CustomerId to TbMsg and SessionInfo

Viacheslav Klimov · Andrew Shvayka
1 parent 0ff4e6ac
Showing 17 changed files with 68 additions and 44 deletions
application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
application/src/main/java/org/thingsboard/server/controller/BaseController.java
application/src/main/java/org/thingsboard/server/service/device/DeviceProvisionServiceImpl.java
application/src/main/java/org/thingsboard/server/service/queue/TbClusterService.java
application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java
common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mTransportContextServer.java
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mTransportServiceImpl.java
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbMsgCountNode.java
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmState.java
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java
--- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
View file @030d80d
+++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
View file @030d80d
@@ -41,6 +41,7 @@ import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.TenantProfile;
 import org.thingsboard.server.common.data.alarm.Alarm;
 import org.thingsboard.server.common.data.asset.Asset;
+import org.thingsboard.server.common.data.id.CustomerId;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.EdgeId;
 import org.thingsboard.server.common.data.id.EntityId;
@@ -269,7 +270,12 @@ class DefaultTbContext implements TbContext {
     @Override
     public TbMsg newMsg(String queueName, String type, EntityId originator, TbMsgMetaData metaData, String data) {
-        return TbMsg.newMsg(queueName, type, originator, metaData, data, nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId());
+        return newMsg(queueName, type, originator, null, metaData, data);
+    }
+
+    @Override
+    public TbMsg newMsg(String queueName, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data) {
+        return TbMsg.newMsg(queueName, type, originator, customerId, metaData, data, nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId());
     }
     @Override
--- a/application/src/main/java/org/thingsboard/server/controller/BaseController.java
View file @030d80d
+++ b/application/src/main/java/org/thingsboard/server/controller/BaseController.java
View file @030d80d
@@ -919,7 +919,7 @@ public abstract class BaseController {
                         entityNode.put("endTs", extractParameter(Long.class, 2, additionalInfo));
                     }
                 }
-                TbMsg tbMsg = TbMsg.newMsg(msgType, entityId, metaData, TbMsgDataType.JSON, json.writeValueAsString(entityNode));
+                TbMsg tbMsg = TbMsg.newMsg(msgType, entityId, customerId, metaData, TbMsgDataType.JSON, json.writeValueAsString(entityNode));
                 TenantId tenantId = user.getTenantId();
                 if (tenantId.isNullUid()) {
                     if (entity instanceof HasTenantId) {
--- a/application/src/main/java/org/thingsboard/server/service/device/DeviceProvisionServiceImpl.java
View file @030d80d
+++ b/application/src/main/java/org/thingsboard/server/service/device/DeviceProvisionServiceImpl.java
View file @030d80d
@@ -230,7 +230,7 @@ public class DeviceProvisionServiceImpl implements DeviceProvisionService {
     private void pushProvisionEventToRuleEngine(ProvisionRequest request, Device device, String type) {
         try {
             JsonNode entityNode = JacksonUtil.valueToTree(request);
-            TbMsg msg = TbMsg.newMsg(type, device.getId(), createTbMsgMetaData(device), JacksonUtil.toString(entityNode));
+            TbMsg msg = TbMsg.newMsg(type, device.getId(), device.getCustomerId(), createTbMsgMetaData(device), JacksonUtil.toString(entityNode));
             sendToRuleEngine(device.getTenantId(), msg, null);
         } catch (IllegalArgumentException e) {
             log.warn("[{}] Failed to push device action to rule engine: {}", device.getId(), type, e);
@@ -240,7 +240,7 @@ public class DeviceProvisionServiceImpl implements DeviceProvisionService {
     private void pushDeviceCreatedEventToRuleEngine(Device device) {
         try {
             ObjectNode entityNode = JacksonUtil.OBJECT_MAPPER.valueToTree(device);
-            TbMsg msg = TbMsg.newMsg(DataConstants.ENTITY_CREATED, device.getId(), createTbMsgMetaData(device), JacksonUtil.OBJECT_MAPPER.writeValueAsString(entityNode));
+            TbMsg msg = TbMsg.newMsg(DataConstants.ENTITY_CREATED, device.getId(), device.getCustomerId(), createTbMsgMetaData(device), JacksonUtil.OBJECT_MAPPER.writeValueAsString(entityNode));
             sendToRuleEngine(device.getTenantId(), msg, null);
         } catch (JsonProcessingException | IllegalArgumentException e) {
             log.warn("[{}] Failed to push device action to rule engine: {}", device.getId(), DataConstants.ENTITY_CREATED, e);
--- a/application/src/main/java/org/thingsboard/server/service/queue/TbClusterService.java
View file @030d80d
+++ b/application/src/main/java/org/thingsboard/server/service/queue/TbClusterService.java
View file @030d80d
@@ -22,7 +22,6 @@ import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.TbResource;
 import org.thingsboard.server.common.data.Tenant;
 import org.thingsboard.server.common.data.TenantProfile;
-import org.thingsboard.server.common.data.id.CustomerId;
 import org.thingsboard.server.common.data.id.EdgeId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
--- a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java
View file @030d80d
+++ b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java
View file @030d80d
@@ -412,8 +412,8 @@ public class DefaultTransportApiService implements TransportApiService {
         return DeviceInfoProto.newBuilder()
                 .setTenantIdMSB(device.getTenantId().getId().getMostSignificantBits())
                 .setTenantIdLSB(device.getTenantId().getId().getLeastSignificantBits())
-                .setCustomerIdMSB(device.getCustomerId().getId().getMostSignificantBits())
-                .setCustomerIdLSB(device.getCustomerId().getId().getLeastSignificantBits())
+                .setCustomerIdMSB(Optional.ofNullable(device.getCustomerId()).map(customerId -> customerId.getId().getMostSignificantBits()).orElse(0L))
+                .setCustomerIdLSB(Optional.ofNullable(device.getCustomerId()).map(customerId -> customerId.getId().getLeastSignificantBits()).orElse(0L))
                 .setDeviceIdMSB(device.getId().getId().getMostSignificantBits())
                 .setDeviceIdLSB(device.getId().getId().getLeastSignificantBits())
                 .setDeviceName(device.getName())
--- a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java
View file @030d80d
+++ b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java
View file @030d80d
@@ -19,10 +19,10 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.InvalidProtocolBufferException;
 import lombok.AccessLevel;
-import lombok.Builder;
 import lombok.Data;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.id.CustomerId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.EntityIdFactory;
@@ -67,8 +67,7 @@ public final class TbMsg implements Serializable {
     transient private final TbMsgCallback callback;
     public static TbMsg newMsg(String queueName, String type, EntityId originator, TbMsgMetaData metaData, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
-        return new TbMsg(queueName, UUID.randomUUID(), System.currentTimeMillis(), type, originator, null,
-                metaData.copy(), TbMsgDataType.JSON, data, ruleChainId, ruleNodeId, 0, TbMsgCallback.EMPTY);
+        return newMsg(queueName, type, originator, null, metaData, data, ruleChainId, ruleNodeId);
     }
     public static TbMsg newMsg(String queueName, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
@@ -77,13 +76,21 @@ public final class TbMsg implements Serializable {
     }
     public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data) {
-        return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, null, metaData.copy(), TbMsgDataType.JSON, data, null, null, 0, TbMsgCallback.EMPTY);
+        return newMsg(type, originator, null, metaData, data);
+    }
+
+    public static TbMsg newMsg(String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data) {
+        return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId, metaData.copy(), TbMsgDataType.JSON, data, null, null, 0, TbMsgCallback.EMPTY);
     }
     // REALLY NEW MSG
     public static TbMsg newMsg(String queueName, String type, EntityId originator, TbMsgMetaData metaData, String data) {
-        return new TbMsg(queueName, UUID.randomUUID(), System.currentTimeMillis(), type, originator, null, metaData.copy(), TbMsgDataType.JSON, data, null, null, 0, TbMsgCallback.EMPTY);
+        return newMsg(queueName, type, originator, null, metaData, data);
+    }
+
+    public static TbMsg newMsg(String queueName, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data) {
+        return new TbMsg(queueName, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId, metaData.copy(), TbMsgDataType.JSON, data, null, null, 0, TbMsgCallback.EMPTY);
     }
     public static TbMsg newMsg(String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, TbMsgDataType dataType, String data) {
@@ -91,7 +98,7 @@ public final class TbMsg implements Serializable {
     }
     public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data) {
-        return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, null, metaData.copy(), dataType, data, null, null, 0, TbMsgCallback.EMPTY);
+        return newMsg(type, originator, null, metaData, dataType, data);
     }
     // For Tests only
@@ -145,7 +152,15 @@ public final class TbMsg implements Serializable {
         }
         this.type = type;
         this.originator = originator;
-        this.customerId = (customerId == null || customerId.isNullUid()) ? null : customerId;
+        if (customerId == null || customerId.isNullUid()) {
+            if (originator.getEntityType() == EntityType.CUSTOMER) {
+                this.customerId = (CustomerId) originator;
+            } else {
+                this.customerId = null;
+            }
+        } else {
+            this.customerId = customerId;
+        }
         this.metaData = metaData;
         this.dataType = dataType;
         this.data = data;
--- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mTransportContextServer.java
View file @030d80d
+++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mTransportContextServer.java
View file @030d80d
@@ -141,6 +141,8 @@ public class LwM2mTransportContextServer extends TransportContext {
                 .setDeviceIdLSB(msg.getDeviceInfo().getDeviceIdLSB())
                 .setTenantIdMSB(msg.getDeviceInfo().getTenantIdMSB())
                 .setTenantIdLSB(msg.getDeviceInfo().getTenantIdLSB())
+                .setCustomerIdMSB(msg.getDeviceInfo().getCustomerIdMSB())
+                .setCustomerIdLSB(msg.getDeviceInfo().getCustomerIdLSB())
                 .setDeviceName(msg.getDeviceInfo().getDeviceName())
                 .setDeviceType(msg.getDeviceInfo().getDeviceType())
                 .setDeviceProfileIdLSB(msg.getDeviceInfo().getDeviceProfileIdLSB())
--- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mTransportServiceImpl.java
View file @030d80d
+++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mTransportServiceImpl.java
View file @030d80d
@@ -1177,6 +1177,8 @@ public class LwM2mTransportServiceImpl implements LwM2mTransportService {
                         .setDeviceIdLSB(msg.getDeviceInfo().getDeviceIdLSB())
                         .setTenantIdMSB(msg.getDeviceInfo().getTenantIdMSB())
                         .setTenantIdLSB(msg.getDeviceInfo().getTenantIdLSB())
+                        .setCustomerIdMSB(msg.getDeviceInfo().getCustomerIdMSB())
+                        .setCustomerIdLSB(msg.getDeviceInfo().getCustomerIdLSB())
                         .setDeviceName(msg.getDeviceInfo().getDeviceName())
                         .setDeviceType(msg.getDeviceInfo().getDeviceType())
                         .setDeviceProfileIdLSB(msg.getDeviceInfo().getDeviceProfileIdLSB())
--- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java
View file @030d80d
+++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java
View file @030d80d
@@ -45,6 +45,8 @@ public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext imple
                 .setDeviceIdLSB(deviceInfo.getDeviceId().getId().getLeastSignificantBits())
                 .setTenantIdMSB(deviceInfo.getTenantId().getId().getMostSignificantBits())
                 .setTenantIdLSB(deviceInfo.getTenantId().getId().getLeastSignificantBits())
+                .setCustomerIdMSB(deviceInfo.getCustomerId().getId().getMostSignificantBits())
+                .setCustomerIdLSB(deviceInfo.getCustomerId().getId().getLeastSignificantBits())
                 .setDeviceName(deviceInfo.getDeviceName())
                 .setDeviceType(deviceInfo.getDeviceType())
                 .setGwSessionIdMSB(parent.getSessionId().getMostSignificantBits())
--- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java
View file @030d80d
+++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java
View file @030d80d
@@ -16,16 +16,15 @@
 package org.thingsboard.rule.engine.api;
 import io.netty.channel.EventLoopGroup;
-import org.springframework.data.redis.core.RedisTemplate;
 import org.thingsboard.common.util.ListeningExecutor;
 import org.thingsboard.rule.engine.api.sms.SmsSenderFactory;
-import org.thingsboard.server.common.data.ApiUsageRecordKey;
 import org.thingsboard.server.common.data.Customer;
 import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.TenantProfile;
 import org.thingsboard.server.common.data.alarm.Alarm;
 import org.thingsboard.server.common.data.asset.Asset;
+import org.thingsboard.server.common.data.id.CustomerId;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.EdgeId;
 import org.thingsboard.server.common.data.id.EntityId;
@@ -144,6 +143,8 @@ public interface TbContext {
     TbMsg newMsg(String queueName, String type, EntityId originator, TbMsgMetaData metaData, String data);
+    TbMsg newMsg(String queueName, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data);
+
     TbMsg transformMsg(TbMsg origMsg, String type, EntityId originator, TbMsgMetaData metaData, String data);
     TbMsg customerCreatedMsg(Customer customer, RuleNodeId ruleNodeId);
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java
View file @030d80d
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java
View file @030d80d
@@ -136,7 +136,7 @@ public class TbCopyAttributesToEntityViewNode implements TbNode {
     }
     private void transformAndTellNext(TbContext ctx, TbMsg msg, EntityView entityView) {
-        ctx.enqueueForTellNext(ctx.newMsg(msg.getQueueName(), msg.getType(), entityView.getId(), msg.getMetaData(), msg.getData()), SUCCESS);
+        ctx.enqueueForTellNext(ctx.newMsg(msg.getQueueName(), msg.getType(), entityView.getId(), msg.getCustomerId(), msg.getMetaData(), msg.getData()), SUCCESS);
     }
     private boolean attributeContainsInEntityView(String scope, String attrKey, EntityView entityView) {
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbMsgCountNode.java
View file @030d80d
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbMsgCountNode.java
View file @030d80d
@@ -63,7 +63,7 @@ public class TbMsgCountNode implements TbNode {
         TbMsgCountNodeConfiguration config = TbNodeUtils.convert(configuration, TbMsgCountNodeConfiguration.class);
         this.delay = TimeUnit.SECONDS.toMillis(config.getInterval());
         this.telemetryPrefix = config.getTelemetryPrefix();
-        scheduleTickMsg(ctx);
+        scheduleTickMsg(ctx, null);
     }
@@ -78,23 +78,23 @@ public class TbMsgCountNode implements TbNode {
             TbMsgMetaData metaData = new TbMsgMetaData();
             metaData.putValue("delta", Long.toString(System.currentTimeMillis() - lastScheduledTs + delay));
-            TbMsg tbMsg = TbMsg.newMsg(msg.getQueueName(), SessionMsgType.POST_TELEMETRY_REQUEST.name(), ctx.getTenantId(), metaData, gson.toJson(telemetryJson));
+            TbMsg tbMsg = TbMsg.newMsg(msg.getQueueName(), SessionMsgType.POST_TELEMETRY_REQUEST.name(), ctx.getTenantId(), msg.getCustomerId(), metaData, gson.toJson(telemetryJson));
             ctx.enqueueForTellNext(tbMsg, SUCCESS);
-            scheduleTickMsg(ctx);
+            scheduleTickMsg(ctx, tbMsg);
         } else {
             messagesProcessed.incrementAndGet();
             ctx.ack(msg);
         }
     }
-    private void scheduleTickMsg(TbContext ctx) {
+    private void scheduleTickMsg(TbContext ctx, TbMsg msg) {
         long curTs = System.currentTimeMillis();
         if (lastScheduledTs == 0L) {
             lastScheduledTs = curTs;
         }
         lastScheduledTs = lastScheduledTs + delay;
         long curDelay = Math.max(0L, (lastScheduledTs - curTs));
-        TbMsg tickMsg = ctx.newMsg(ServiceQueue.MAIN, TB_MSG_COUNT_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), "");
+        TbMsg tickMsg = ctx.newMsg(ServiceQueue.MAIN, TB_MSG_COUNT_NODE_MSG, ctx.getSelfId(), msg != null ? msg.getCustomerId() : null, new TbMsgMetaData(), "");
         nextTickId = tickMsg.getId();
         ctx.tellSelf(tickMsg, curDelay);
     }
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java
View file @030d80d
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java
View file @030d80d
@@ -100,7 +100,7 @@ public class TbMsgGeneratorNode implements TbNode {
     @Override
     public void onMsg(TbContext ctx, TbMsg msg) {
         if (initialized && msg.getType().equals(TB_MSG_GENERATOR_NODE_MSG) && msg.getId().equals(nextTickId)) {
-            withCallback(generate(ctx),
+            withCallback(generate(ctx, msg),
                     m -> {
                         if (initialized && (config.getMsgCount() == TbMsgGeneratorNodeConfiguration.UNLIMITED_MSG_COUNT || currentMsgCount < config.getMsgCount())) {
                             ctx.enqueueForTellNext(m, SUCCESS);
@@ -130,16 +130,16 @@ public class TbMsgGeneratorNode implements TbNode {
         ctx.tellSelf(tickMsg, curDelay);
     }
-    private ListenableFuture<TbMsg> generate(TbContext ctx) {
+    private ListenableFuture<TbMsg> generate(TbContext ctx, TbMsg msg) {
         return ctx.getJsExecutor().executeAsync(() -> {
             if (prevMsg == null) {
-                prevMsg = ctx.newMsg(ServiceQueue.MAIN, "", originatorId, new TbMsgMetaData(), "{}");
+                prevMsg = ctx.newMsg(ServiceQueue.MAIN, "", originatorId, msg.getCustomerId(), new TbMsgMetaData(), "{}");
             }
             if (initialized) {
                 ctx.logJsEvalRequest();
                 TbMsg generated = jsEngine.executeGenerate(prevMsg);
                 ctx.logJsEvalResponse();
-                prevMsg = ctx.newMsg(ServiceQueue.MAIN, generated.getType(), originatorId, generated.getMetaData(), generated.getData());
+                prevMsg = ctx.newMsg(ServiceQueue.MAIN, generated.getType(), originatorId, msg.getCustomerId(), generated.getMetaData(), generated.getData());
             }
             return prevMsg;
         });
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java
View file @030d80d
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java
View file @030d80d
@@ -70,7 +70,7 @@ public class TbMsgDelayNode implements TbNode {
         } else {
             if (pendingMsgs.size() < config.getMaxPendingMsgs()) {
                 pendingMsgs.put(msg.getId(), msg);
-                TbMsg tickMsg = ctx.newMsg(ServiceQueue.MAIN, TB_MSG_DELAY_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), msg.getId().toString());
+                TbMsg tickMsg = ctx.newMsg(ServiceQueue.MAIN, TB_MSG_DELAY_NODE_MSG, ctx.getSelfId(), msg.getCustomerId(), new TbMsgMetaData(), msg.getId().toString());
                 ctx.tellSelf(tickMsg, getDelay(msg));
                 ctx.ack(msg);
             } else {
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmState.java
View file @030d80d
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmState.java
View file @030d80d
@@ -74,15 +74,15 @@ class AlarmState {
         lastMsgMetaData = msg.getMetaData();
         lastMsgQueueName = msg.getQueueName();
         this.dataSnapshot = data;
-        return createOrClearAlarms(ctx, data, update, AlarmRuleState::eval);
+        return createOrClearAlarms(ctx, msg, data, update, AlarmRuleState::eval);
     }
     public boolean process(TbContext ctx, long ts) throws ExecutionException, InterruptedException {
         initCurrentAlarm(ctx);
-        return createOrClearAlarms(ctx, ts, null, AlarmRuleState::eval);
+        return createOrClearAlarms(ctx, null, ts, null, AlarmRuleState::eval);
     }
-    public <T> boolean createOrClearAlarms(TbContext ctx, T data, SnapshotUpdate update, BiFunction<AlarmRuleState, T, AlarmEvalResult> evalFunction) {
+    public <T> boolean createOrClearAlarms(TbContext ctx, TbMsg msg, T data, SnapshotUpdate update, BiFunction<AlarmRuleState, T, AlarmEvalResult> evalFunction) {
         boolean stateUpdate = false;
         AlarmRuleState resultState = null;
         log.debug("[{}] processing update: {}", alarmDefinition.getId(), data);
@@ -103,7 +103,7 @@ class AlarmState {
         if (resultState != null) {
             TbAlarmResult result = calculateAlarmResult(ctx, resultState);
             if (result != null) {
-                pushMsg(ctx, result, resultState);
+                pushMsg(ctx, msg, result, resultState);
             }
             stateUpdate = clearAlarmState(stateUpdate, clearState);
         } else if (currentAlarm != null && clearState != null) {
@@ -122,7 +122,7 @@ class AlarmState {
                 );
                 DonAsynchron.withCallback(alarmClearOperationResult,
                         result -> {
-                            pushMsg(ctx, new TbAlarmResult(false, false, true, result.getAlarm()), clearState);
+                            pushMsg(ctx, msg, new TbAlarmResult(false, false, true, result.getAlarm()), clearState);
                         },
                         throwable -> {
                             throw new RuntimeException(throwable);
@@ -165,7 +165,7 @@ class AlarmState {
         }
     }
-    public void pushMsg(TbContext ctx, TbAlarmResult alarmResult, AlarmRuleState ruleState) {
+    public void pushMsg(TbContext ctx, TbMsg msg, TbAlarmResult alarmResult, AlarmRuleState ruleState) {
         JsonNode jsonNodes = JacksonUtil.valueToTree(alarmResult.getAlarm());
         String data = jsonNodes.toString();
         TbMsgMetaData metaData = lastMsgMetaData != null ? lastMsgMetaData.copy() : new TbMsgMetaData();
@@ -185,7 +185,8 @@ class AlarmState {
             metaData.putValue(DataConstants.IS_CLEARED_ALARM, Boolean.TRUE.toString());
         }
         setAlarmConditionMetadata(ruleState, metaData);
-        TbMsg newMsg = ctx.newMsg(lastMsgQueueName != null ? lastMsgQueueName : ServiceQueue.MAIN, "ALARM", originator, metaData, data);
+        TbMsg newMsg = ctx.newMsg(lastMsgQueueName != null ? lastMsgQueueName : ServiceQueue.MAIN, "ALARM",
+                originator, msg != null ? msg.getCustomerId() : null, metaData, data);
         ctx.tellNext(newMsg, relationType);
     }
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java
View file @030d80d
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java
View file @030d80d
@@ -74,7 +74,7 @@ public class TbDeviceProfileNode implements TbNode {
         this.config = TbNodeUtils.convert(configuration, TbDeviceProfileNodeConfiguration.class);
         this.cache = ctx.getDeviceProfileCache();
         this.ctx = ctx;
-        scheduleAlarmHarvesting(ctx);
+        scheduleAlarmHarvesting(ctx, null);
         ctx.addDeviceProfileListeners(this::onProfileUpdate, this::onDeviceUpdate);
         if (config.isFetchAlarmRulesStateOnStart()) {
             log.info("[{}] Fetching alarm rule state", ctx.getSelfId());
@@ -108,7 +108,7 @@ public class TbDeviceProfileNode implements TbNode {
     public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException {
         EntityType originatorType = msg.getOriginator().getEntityType();
         if (msg.getType().equals(PERIODIC_MSG_TYPE)) {
-            scheduleAlarmHarvesting(ctx);
+            scheduleAlarmHarvesting(ctx, msg);
             harvestAlarms(ctx, System.currentTimeMillis());
         } else if (msg.getType().equals(PROFILE_UPDATE_MSG_TYPE)) {
             updateProfile(ctx, new DeviceProfileId(UUID.fromString(msg.getData())));
@@ -168,8 +168,8 @@ public class TbDeviceProfileNode implements TbNode {
         return deviceState;
     }
-    protected void scheduleAlarmHarvesting(TbContext ctx) {
-        TbMsg periodicCheck = TbMsg.newMsg(PERIODIC_MSG_TYPE, ctx.getTenantId(), TbMsgMetaData.EMPTY, "{}");
+    protected void scheduleAlarmHarvesting(TbContext ctx, TbMsg msg) {
+        TbMsg periodicCheck = TbMsg.newMsg(PERIODIC_MSG_TYPE, ctx.getTenantId(), msg != null ? msg.getCustomerId() : null, TbMsgMetaData.EMPTY, "{}");
         ctx.tellSelf(periodicCheck, TimeUnit.MINUTES.toMillis(1));
     }
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java
View file @030d80d
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java
View file @030d80d
@@ -16,10 +16,6 @@
 package org.thingsboard.rule.engine.rpc;
 import com.datastax.oss.driver.api.core.uuid.Uuids;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
 import com.google.gson.Gson;
 import com.google.gson.JsonElement;
 import com.google.gson.JsonObject;
@@ -116,10 +112,10 @@ public class TbSendRPCRequestNode implements TbNode {
             ctx.getRpcService().sendRpcRequestToDevice(request, ruleEngineDeviceRpcResponse -> {
                 if (!ruleEngineDeviceRpcResponse.getError().isPresent()) {
-                    TbMsg next = ctx.newMsg(msg.getQueueName(), msg.getType(), msg.getOriginator(), msg.getMetaData(), ruleEngineDeviceRpcResponse.getResponse().orElse("{}"));
+                    TbMsg next = ctx.newMsg(msg.getQueueName(), msg.getType(), msg.getOriginator(), msg.getCustomerId(), msg.getMetaData(), ruleEngineDeviceRpcResponse.getResponse().orElse("{}"));
                     ctx.enqueueForTellNext(next, TbRelationTypes.SUCCESS);
                 } else {
-                    TbMsg next = ctx.newMsg(msg.getQueueName(), msg.getType(), msg.getOriginator(), msg.getMetaData(), wrap("error", ruleEngineDeviceRpcResponse.getError().get().name()));
+                    TbMsg next = ctx.newMsg(msg.getQueueName(), msg.getType(), msg.getOriginator(), msg.getCustomerId(), msg.getMetaData(), wrap("error", ruleEngineDeviceRpcResponse.getError().get().name()));
                     ctx.tellFailure(next, new RuntimeException(ruleEngineDeviceRpcResponse.getError().get().name()));
                 }
             });