refactor: 动作之设备输出

1、兼容多设备输出

refactor: 动作之设备输出
1、兼容多设备输出
云中非
1 parent 2cec1d4e
Showing 1 changed file with 84 additions and 62 deletions
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/yunteng/scene/ReactState.java
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/yunteng/scene/ReactState.java
View file @13530c9
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/yunteng/scene/ReactState.java
View file @13530c9
@@ -33,121 +33,116 @@ import java.util.concurrent.ExecutionException;
 @Slf4j
 class ReactState {
-    /**场景联动主键*/
+    /**
+     * 场景联动主键
+     */
     String reactId;
-    /**场景联动的触发器状态
+    /**
+     * 场景联动的触发器状态
      * 键：设备主键
      * 值：设备指标参与的触发器
      */
-    private ConcurrentHashMap<String, TriggerState > triggerState = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<String, TriggerState> triggerState = new ConcurrentHashMap<>();
     private ConcurrentHashMap<String, TriggerState> clearState = new ConcurrentHashMap<>();
-    /**场景联动的执行条件状态
+    /**
+     * 场景联动的执行条件状态
      * 键：设备主键
      * 值：设备指标参与的触发器
      */
     private ConcurrentHashMap<String, TriggerState> conditionState = new ConcurrentHashMap<>();
-
-
-
     private final List<TriggerDTO> triggers;
-    /**【场景联动的执行条件】懒加载*/
+    /**
+     * 【场景联动的执行条件】懒加载
+     */
     private final List<DoCondition> conditions;
-    /**【场景联动的执行集合】懒加载*/
+    /**
+     * 【场景联动的执行集合】懒加载
+     */
     private final List<DoAction> actions;
-
-
-
     private RuleNodeState state;
-    ReactState(String reactId,TbContext ctx, TbSceneReactNodeConfig config) {
+    ReactState(String reactId, TbContext ctx, TbSceneReactNodeConfig config) {
         this.reactId = reactId;
         TriggerService triggerService = SpringBeanUtils.getBean(TriggerService.class);
         this.triggers = triggerService.getTrigger(reactId);
         DoConditionService conditionService = SpringBeanUtils.getBean(DoConditionService.class);
         this.conditions = conditionService.getConditions(reactId);
         DoActionService actionService = SpringBeanUtils.getBean(DoActionService.class);
-        this.actions =actionService.getActions(reactId);
+        this.actions = actionService.getActions(reactId);
     }
-
-    public void process(TbContext ctx, TbMsg msg,String deviceId) throws ExecutionException, InterruptedException {
-
-
+    public void process(TbContext ctx, TbMsg msg, String deviceId) throws ExecutionException, InterruptedException {
-        if( actions == null){
+        if (actions == null) {
             ctx.tellSuccess(msg);
         }
         boolean matched;
-        if(triggers == null || triggers.isEmpty()){
+        if (triggers == null || triggers.isEmpty()) {
             matched = true;
-        }else{
+        } else {
             matched = false;
-            for(TriggerDTO trigger: triggers){
-                TriggerState triggerState = getOrCreateTriggerState(trigger,deviceId);
-                matched = triggerState.process(ctx,msg);
-                if(matched){
+            for (TriggerDTO trigger : triggers) {
+                TriggerState triggerState = getOrCreateTriggerState(trigger, deviceId);
+                matched = triggerState.process(ctx, msg);
+                if (matched) {
                     break;
                 }
             }
         }
-        if(matched && conditions.size() >0 ){
+        if (matched && conditions.size() > 0) {
             matched = false;
-            for(DoCondition item:conditions){
+            for (DoCondition item : conditions) {
                 List<String> entityIds = item.getEntityId();
-                if(entityIds == null || entityIds.isEmpty()){
+                if (entityIds == null || entityIds.isEmpty()) {
                     matched = true;
                     break;
                 }
-                for(String id:entityIds){
-                    TriggerState conditionState = getOrCreateConditionState(item.getId(),id,item.getTriggerCondition());
-                    if( conditionState == null
-                            || conditionState.process(ctx,msg)){
+                for (String id : entityIds) {
+                    TriggerState conditionState = getOrCreateConditionState(item.getId(), id, item.getTriggerCondition());
+                    if (conditionState == null
+                            || conditionState.process(ctx, msg)) {
                         matched = true;
                         break;
                     }
                 }
-                if(matched){
+                if (matched) {
                     break;
                 }
             }
         }
-        if(matched){
-            for(DoAction item: actions){
-                pushMsg(ctx, msg,  item);
+        if (matched) {
+            for (DoAction item : actions) {
+                pushMsg(ctx, msg, item);
             }
-        }else{
+        } else {
             ctx.tellSuccess(msg);
         }
     }
-
-
-
-
-    protected TriggerState  getOrCreateTriggerState(TriggerDTO trigger,String deviceId) {
+    protected TriggerState getOrCreateTriggerState(TriggerDTO trigger, String deviceId) {
         String triggerId = trigger.getId();
-        String cacheKey =triggerId+deviceId;
-        if(triggerState.containsKey(cacheKey)){
+        String cacheKey = triggerId + deviceId;
+        if (triggerState.containsKey(cacheKey)) {
             return triggerState.get(cacheKey);
         }
-        if(trigger.getEntityId().contains(deviceId)){
+        if (trigger.getEntityId().contains(deviceId)) {
             TriggerState state = createTriggerState(deviceId, trigger.getTriggerCondition());
             triggerState.put(cacheKey, state);
             return state;
@@ -157,13 +152,13 @@ class ReactState {
     }
     protected TriggerState getOrCreateConditionState(String conditionId, String deviceId, AlarmRule condition) {
-        String cacheKey =conditionId+deviceId;
-        if(conditionState.containsKey(cacheKey)){
+        String cacheKey = conditionId + deviceId;
+        if (conditionState.containsKey(cacheKey)) {
             return conditionState.get(cacheKey);
-        }else{
+        } else {
             TriggerState state = createTriggerState(deviceId, condition);
             conditionState.put(cacheKey, state);
-            return  state;
+            return state;
         }
     }
@@ -171,12 +166,12 @@ class ReactState {
     @NotNull
     private TriggerState createTriggerState(String deviceId, AlarmRule rule) {
         Set<AlarmConditionFilterKey> filterKeys = new HashSet<>();
-        for(AlarmConditionFilter filter :rule.getCondition().getCondition()){
+        for (AlarmConditionFilter filter : rule.getCondition().getCondition()) {
             filterKeys.add(filter.getKey());
         }
-        TriggerState state = new TriggerState(deviceId,rule, filterKeys,null);
+        TriggerState state = new TriggerState(deviceId, rule, filterKeys, null);
-        return  state;
+        return state;
     }
     private void pushMsg(TbContext ctx, TbMsg msg, DoAction action) {
@@ -184,38 +179,65 @@ class ReactState {
                 new TbMsgMetaData();
         String relationType = "";
         TbMsg newMsg = null;
-        switch(action.getOutTarget()){
+        switch (action.getOutTarget()) {
             case DEVICE_OUT:
-                relationType = "RPC Request";
-                newMsg = rpcMsg(ctx,msg,action.getDoContext());
+                rpcMsg(ctx, msg, action.getDeviceId(), action.getDoContext());
                 break;
             case SCENE_ACT:
-                //TODO: 场景联动关联场景联动
-                relationType = "Alarm Updated";
+                //TODO: 场景联动关联消息通知
                 break;
             case MSG_NOTIFY:
-                relationType = "Message";
+                noticeMsg(ctx, msg, action.getDoContext());
                 break;
             default:
                 ctx.tellSuccess(msg);
                 break;
         }
-        if(newMsg != null){
+        if (newMsg != null) {
             ctx.tellNext(newMsg, relationType);
         }
     }
-    private TbMsg rpcMsg(TbContext ctx, TbMsg msg, JsonNode context){
+    /**
+     * 设备输出
+     *
+     * @param ctx
+     * @param msg
+     * @param devices
+     * @param context
+     */
+    private void rpcMsg(TbContext ctx, TbMsg msg, List<String> devices, JsonNode context) {
+        String lastMsgQueueName = msg.getQueueName();
+        TbMsgMetaData metaData = msg.getMetaData();
+        for (String id : devices) {
+            TbMsg newMsg = ctx.newMsg(lastMsgQueueName != null ? lastMsgQueueName : ServiceQueue.MAIN
+                    , msg.getType()
+                    , msg.getOriginator()
+                    , msg != null ? msg.getCustomerId() : null
+                    , metaData
+                    , JacksonUtil.toString(context));
+            ctx.tellNext(newMsg, "RPC Request");
+        }
+    }
+
+    /**
+     * 告警通知
+     *
+     * @param ctx
+     * @param msg
+     * @param context
+     */
+    private void noticeMsg(TbContext ctx, TbMsg msg, JsonNode context) {
         String lastMsgQueueName = msg.getQueueName();
         TbMsgMetaData metaData = msg.getMetaData();
-                metaData.putValue(DataConstants.IS_CLEARED_ALARM, Boolean.TRUE.toString());
+        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
                 , JacksonUtil.toString(context));
-        return newMsg;
+        ctx.tellNext(newMsg, "Message");
     }
 }