fix: 场景联动调用远程RPC

@@ -14,9 +14,11 @@ import org.thingsboard.server.common.data.device.profile.AlarmConditionFilter;
 import org.thingsboard.server.common.data.device.profile.AlarmConditionFilterKey;
 import org.thingsboard.server.common.data.rule.RuleNodeState;
 import org.thingsboard.server.common.data.yunteng.dto.TriggerDTO;
+import org.thingsboard.server.common.data.yunteng.enums.ActionTypeEnum;
 import org.thingsboard.server.common.data.yunteng.utils.SpringBeanUtils;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
+import org.thingsboard.server.common.msg.queue.ServiceQueue;
 import org.thingsboard.server.dao.yunteng.entities.DoAction;
 import org.thingsboard.server.dao.yunteng.entities.DoCondition;
 import org.thingsboard.server.dao.yunteng.service.DoActionService;
@@ -78,9 +80,8 @@ class ReactState {
         }
 
         if(matched){
-            // TODO 输出动作
             for(DoAction item: actions){
-//                pushMsg();
+                pushMsg(ctx, msg,  item);
             }
         }
 
@@ -110,26 +111,43 @@ class ReactState {
         return null;
     }
 
-    private void pushMsg(TbContext ctx, TbMsg msg, TbAlarmResult alarmResult,DoAction action) {
-        JsonNode jsonNodes = JacksonUtil.valueToTree(alarmResult.getAlarm());
-        String data = jsonNodes.toString();
+    private void pushMsg(TbContext ctx, TbMsg msg, DoAction action) {
         TbMsgMetaData metaData = //lastMsgMetaData != null ? lastMsgMetaData.copy() :
                 new TbMsgMetaData();
-        String relationType;
-        if (alarmResult.isCreated()) {
-            relationType = "Alarm Created";
-            metaData.putValue(DataConstants.IS_NEW_ALARM, Boolean.TRUE.toString());
-        } else if (alarmResult.isUpdated()) {
-            relationType = "Alarm Updated";
-            metaData.putValue(DataConstants.IS_EXISTING_ALARM, Boolean.TRUE.toString());
-        } else if (alarmResult.isSeverityUpdated()) {
-            relationType = "Alarm Severity Updated";
-            metaData.putValue(DataConstants.IS_EXISTING_ALARM, Boolean.TRUE.toString());
-            metaData.putValue(DataConstants.IS_SEVERITY_UPDATED_ALARM, Boolean.TRUE.toString());
-        } else {
-            relationType = "Alarm Cleared";
-            metaData.putValue(DataConstants.IS_CLEARED_ALARM, Boolean.TRUE.toString());
+        String relationType = "";
+        TbMsg newMsg = null;
+        switch(action.getOutTarget()){
+            case DEVICE_OUT:
+                relationType = "RPC Request";
+                newMsg = rpcMsg(ctx,msg,action.getDoContext());
+                break;
+            case SCENE_ACT:
+                //TODO: 场景联动关联场景联动
+                relationType = "Alarm Updated";
+                break;
+            case MSG_NOTIFY:
+                relationType = "Message";
+                break;
+            default:
+                ctx.tellSuccess(msg);
+                break;
         }
-        ctx.tellNext(msg, relationType);
+
+        if(newMsg != null){
+            ctx.tellNext(newMsg, relationType);
+        }
+    }
+
+    private TbMsg rpcMsg(TbContext ctx, TbMsg msg,String context){
+        String lastMsgQueueName = msg.getQueueName();
+        TbMsgMetaData metaData = msg.getMetaData();
+                metaData.putValue(DataConstants.IS_CLEARED_ALARM, Boolean.TRUE.toString());
+        TbMsg newMsg = ctx.newMsg(lastMsgQueueName != null ? lastMsgQueueName : ServiceQueue.MAIN
+                , msg.getType()
+                , msg.getOriginator()
+                , msg != null ? msg.getCustomerId() : null
+                , metaData
+                , context);
+        return newMsg;
     }
 }

@@ -67,7 +67,7 @@ class TriggerState {
     private volatile Alarm currentAlarm;
     private volatile boolean initialFetchDone;
     private volatile TbMsgMetaData lastMsgMetaData;
-    private volatile String lastMsgQueueName;
+
     private final DynamicPredicateValueCtx dynamicPredicateValueCtx;
     private DataSnapshot latestValues;
 
@@ -87,7 +87,6 @@ class TriggerState {
             latestValues = fetchLatestValues(ctx, originator);
         }
         lastMsgMetaData = msg.getMetaData();
-        lastMsgQueueName = msg.getQueueName();
         SnapshotUpdate update = null;
         if(msg.getType().equals(SessionMsgType.POST_TELEMETRY_REQUEST.name())){
             update =  processTelemetry(ctx,msg);