Merge branch '20230705' into 'master_dev'

fix: 场景联动命令下发同步异步问题修复

See merge request yunteng/thingskit!204

@@ -171,5 +171,7 @@ public interface FastIotConstants {
     public static String TARGET_ID = "target";
     /**实控设备名称*/
     public static String TARGET_NAME = "deviceName";
+    /**RPC单项双向*/
+    public static String ONEWAY = "oneway";
   }
 }

-/**
- * 设备场景联动状态
- */
+/** 设备场景联动状态 */
 package org.thingsboard.rule.engine.yunteng.scene;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.jetbrains.annotations.NotNull;
 import org.thingsboard.rule.engine.api.TbContext;
 import org.thingsboard.server.common.data.DataConstants;
-import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.alarm.Alarm;
 import org.thingsboard.server.common.data.alarm.AlarmStatus;
 import org.thingsboard.server.common.data.device.profile.AlarmConditionFilter;
 import org.thingsboard.server.common.data.device.profile.AlarmConditionFilterKey;
 import org.thingsboard.server.common.data.device.profile.AlarmRule;
 import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.data.yunteng.constant.FastIotConstants;
 import org.thingsboard.server.common.data.yunteng.dto.ActionAlarmDTO;
 import org.thingsboard.server.common.data.yunteng.dto.AlarmInfoDTO;
 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.enums.ScopeEnum;
-import org.thingsboard.server.common.data.yunteng.enums.TkAlarmSeverity;
-import org.thingsboard.server.common.data.yunteng.enums.TriggerTypeEnum;
+import org.thingsboard.server.common.data.yunteng.enums.*;
 import org.thingsboard.server.common.data.yunteng.utils.JacksonUtil;
 import org.thingsboard.server.common.data.yunteng.utils.SpringBeanUtils;
 import org.thingsboard.server.common.msg.TbMsg;
@@ -33,368 +34,371 @@ import org.thingsboard.server.dao.yunteng.entities.TkDoActionEntity;
 import org.thingsboard.server.dao.yunteng.entities.TkDoConditionEntity;
 import org.thingsboard.server.dao.yunteng.service.*;
 
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
-
 @Slf4j
 class ReactState {
 
-    /**
-     * 场景联动主键
-     */
-    String reactId;
-    String reactName;
-    String orgId;
-
-    /**
-     * 场景联动的触发器状态
-     * 键：触发器ID和设备ID为键
-     * 值：设备指标参与的触发器
-     */
-    private ConcurrentHashMap<String, TriggerState> triggerState = new ConcurrentHashMap<>();
-
-    /**
-     * 场景联动的告警清除触发器状态
-     * 键：设备ID为键
-     * 值：设备指标参与的触发器
-     */
-    private ConcurrentHashMap<String, TriggerState> clearState = new ConcurrentHashMap<>();
-    /**
-     * 设备告警信息
-     * 键：设备ID为键
-     * 值：设备最新告警信息
-     */
-    private ConcurrentHashMap<String, Alarm> currentAlarms = new ConcurrentHashMap<>();
-
-    /**
-     * 场景联动的执行条件状态
-     * 键：触发器ID和设备ID为键
-     * 值：设备指标参与的触发器
-     */
-    private ConcurrentHashMap<String, TriggerState> conditionState = new ConcurrentHashMap<>();
-
+  /** 场景联动主键 */
+  String reactId;
+
+  String reactName;
+  String orgId;
+
+  /** 场景联动的触发器状态 键：触发器ID和设备ID为键 值：设备指标参与的触发器 */
+  private ConcurrentHashMap<String, TriggerState> triggerState = new ConcurrentHashMap<>();
+
+  /** 场景联动的告警清除触发器状态 键：设备ID为键 值：设备指标参与的触发器 */
+  private ConcurrentHashMap<String, TriggerState> clearState = new ConcurrentHashMap<>();
+  /** 设备告警信息 键：设备ID为键 值：设备最新告警信息 */
+  private ConcurrentHashMap<String, Alarm> currentAlarms = new ConcurrentHashMap<>();
+
+  /** 场景联动的执行条件状态 键：触发器ID和设备ID为键 值：设备指标参与的触发器 */
+  private ConcurrentHashMap<String, TriggerState> conditionState = new ConcurrentHashMap<>();
+
+  private final List<TriggerDTO> triggers;
+  private TkDoActionEntity alarmAction = null;
+  /** 【场景联动的执行条件】懒加载 */
+  private final List<TkDoConditionEntity> conditions;
+  /** 【场景联动的执行集合】懒加载 */
+  private final List<TkDoActionEntity> actions;
+
+  private final TkNoticeService noticeService;
+  private TkDeviceService ytDeviceService;
+
+  ReactState(String reactId, TbContext ctx, TbSceneReactNodeConfig config) {
+    this.reactId = reactId;
+    this.reactName = config.getNames().get(reactId);
+    this.orgId = config.getOrgs().get(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.getActionsByAll(reactId);
+    this.actions.addAll(actionService.getActionsByPart(reactId));
+    for (TkDoActionEntity action : actions) {
+      /** 动作中只有1个告警通知 */
+      if (ActionTypeEnum.MSG_NOTIFY.equals(action.getOutTarget())) {
+        this.alarmAction = action;
+        break;
+      }
+    }
+    this.noticeService = SpringBeanUtils.getBean(TkNoticeService.class);
+    this.ytDeviceService = SpringBeanUtils.getBean(TkDeviceService.class);
+  }
 
-    private final List<TriggerDTO> triggers;
-    private TkDoActionEntity alarmAction = null;
-    /**
-     * 【场景联动的执行条件】懒加载
-     */
-    private final List<TkDoConditionEntity> conditions;
-    /**
-     * 【场景联动的执行集合】懒加载
-     */
-    private final List<TkDoActionEntity> actions;
-    private final TkNoticeService noticeService;
-    private TkDeviceService ytDeviceService;
+  public void process(TbContext ctx, TbMsg msg, String profileId, String deviceId)
+      throws ExecutionException, InterruptedException {
 
+    StringBuilder detail = new StringBuilder();
+    if (actions == null) {
+      ctx.tellSuccess(msg);
+    }
 
-    ReactState(String reactId, TbContext ctx, TbSceneReactNodeConfig config) {
-        this.reactId = reactId;
-        this.reactName = config.getNames().get(reactId);
-        this.orgId = config.getOrgs().get(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.getActionsByAll(reactId);
-        this.actions.addAll(actionService.getActionsByPart(reactId));
-        for (TkDoActionEntity action : actions) {
-            /**动作中只有1个告警通知*/
-            if (ActionTypeEnum.MSG_NOTIFY.equals(action.getOutTarget())) {
-                this.alarmAction = action;
-                break;
-            }
+    boolean matched = false;
+    if (triggers == null || triggers.isEmpty()) {
+      matched = true;
+    } else {
+      boolean itemMatched = false;
+      for (TriggerDTO trigger : triggers) {
+        TriggerState triggerState = getOrCreateTriggerState(trigger, profileId, deviceId);
+        if (triggerState == null) {
+          continue;
+        }
+        itemMatched = triggerState.process(ctx, msg);
+        if (itemMatched) {
+          matched = true;
+          detail.append(
+              triggerState.getAlarmDetails() == null ? "" : triggerState.getAlarmDetails());
+          if (this.alarmAction != null) {
+            noticeMsg(
+                ctx,
+                msg,
+                alarmAction,
+                deviceId,
+                triggerState.getAlarmDetails(),
+                triggerState.getLatestValues().getTs());
+          }
+        } else if (currentAlarms.containsKey(deviceId) && this.alarmAction != null) {
+          // 清除设备告警
+          for (AlarmConditionFilterKey entityKey : triggerState.getEntityKeys()) {
+            clearAlarm(ctx, msg, deviceId, entityKey.getKey());
+          }
         }
-        this.noticeService = SpringBeanUtils.getBean(TkNoticeService.class);
-        this.ytDeviceService = SpringBeanUtils.getBean(TkDeviceService.class);
+      }
     }
 
-
-    public void process(TbContext ctx, TbMsg msg,String profileId, String deviceId) throws ExecutionException, InterruptedException {
-
-
-        StringBuilder detail = new StringBuilder();
-        if (actions == null) {
-            ctx.tellSuccess(msg);
+    if (matched && conditions.size() > 0) {
+      matched = false;
+      for (TkDoConditionEntity item : conditions) {
+        List<String> entityIds = item.getEntityId();
+        if (entityIds == null || entityIds.isEmpty()) {
+          matched = true;
+          break;
         }
-
-        boolean matched = false;
-        if (triggers == null || triggers.isEmpty()) {
+        for (String id : entityIds) {
+          TriggerState conditionState = getOrCreateConditionState(item, profileId, id);
+          if (conditionState == null || conditionState.process(ctx, msg)) {
+            detail.append("；");
+            detail.append(conditionState.getAlarmDetails());
             matched = true;
-        } else {
-            boolean itemMatched = false;
-            for (TriggerDTO trigger : triggers) {
-                TriggerState triggerState = getOrCreateTriggerState(trigger,profileId, deviceId);
-                if (triggerState == null) {
-                    continue;
-                }
-                itemMatched = triggerState.process(ctx, msg);
-                if (itemMatched) {
-                    matched = true;
-                    detail.append(triggerState.getAlarmDetails()==null?"":triggerState.getAlarmDetails());
-                    if (this.alarmAction != null) {
-                        noticeMsg(ctx, msg, alarmAction, deviceId, triggerState.getAlarmDetails(), triggerState.getLatestValues().getTs());
-                    }
-                } else if (currentAlarms.containsKey(deviceId) && this.alarmAction != null) {
-                    //清除设备告警
-                    for (AlarmConditionFilterKey entityKey : triggerState.getEntityKeys()) {
-                        clearAlarm(ctx, msg, deviceId, entityKey.getKey());
-                    }
-                }
-            }
+            break;
+          }
         }
-
-
-        if (matched && conditions.size() > 0) {
-            matched = false;
-            for (TkDoConditionEntity item : conditions) {
-                List<String> entityIds = item.getEntityId();
-                if (entityIds == null || entityIds.isEmpty()) {
-                    matched = true;
-                    break;
-                }
-                for (String id : entityIds) {
-                    TriggerState conditionState = getOrCreateConditionState(item, profileId,id);
-                    if (conditionState == null
-                            || conditionState.process(ctx, msg)) {
-                        detail.append("；");
-                        detail.append(conditionState.getAlarmDetails());
-                        matched = true;
-                        break;
-                    }
-                }
-                if (matched) {
-                    break;
-                }
-            }
-        }
-
         if (matched) {
-            for (TkDoActionEntity item : actions) {
-                if (ActionTypeEnum.MSG_NOTIFY.equals(item.getOutTarget())) {
-                    continue;
-                }
-                pushMsg(ctx, msg, item, detail.toString());
-            }
-        } else {
-            ctx.tellSuccess(msg);
+          break;
         }
-
+      }
     }
 
-
-    protected TriggerState getOrCreateTriggerState(TriggerDTO trigger, String profileId, String deviceId) {
-        String triggerId = trigger.getId();
-        String cacheKey = triggerId + deviceId;
-        if (triggerState.containsKey(cacheKey)) {
-            return triggerState.get(cacheKey);
-        }
-        if ((trigger.getEntityType().equals(ScopeEnum.PART) && trigger.getEntityId().contains(deviceId))
-                ||(trigger.getEntityType().equals(ScopeEnum.ALL) && trigger.getDeviceProfileId().equals(profileId))  ) {
-            TriggerState state = createTriggerState(deviceId, trigger.getTriggerCondition());
-            triggerState.put(cacheKey, state);
-            return state;
+    if (matched) {
+      for (TkDoActionEntity item : actions) {
+        if (ActionTypeEnum.MSG_NOTIFY.equals(item.getOutTarget())) {
+          continue;
         }
-        return null;
-
+        pushMsg(ctx, msg, item, detail.toString());
+      }
+    } else {
+      ctx.tellSuccess(msg);
     }
-
-    protected TriggerState getOrCreateClearState(String deviceId, String key) {
-        String cacheKey = deviceId + key;
-        if (triggerState.containsKey(cacheKey)) {
-            return triggerState.get(cacheKey);
-        }
-        TriggerState state = null;
-        ActionAlarmDTO alarm = JacksonUtil.convertValue(alarmAction.getDoContext(), ActionAlarmDTO.class);
-        if (alarm != null && alarm.getClearRule() != null) {
-            for (TriggerDTO rule : alarm.getClearRule()) {
-                if ((ScopeEnum.PART.equals(rule.getEntityType()) && !rule.getEntityId().contains(deviceId))
-//                        || !alarmAction.getDeviceId().contains(deviceId)
-                ) {
-                    continue;
-                }
-
-                for (AlarmConditionFilter filter : rule.getTriggerCondition().getCondition().getCondition()) {
-                    String tempKey = filter.getKey().getKey();
-                    if (key.equals(tempKey)) {
-                        state = createTriggerState(deviceId, rule.getTriggerCondition());
-                        triggerState.put(cacheKey, state);
-                        break;
-                    }
-                }
-                if (state != null) {
-                    break;
-                }
-            }
-            return state;
-        }
-        return null;
-
+  }
+
+  protected TriggerState getOrCreateTriggerState(
+      TriggerDTO trigger, String profileId, String deviceId) {
+    String triggerId = trigger.getId();
+    String cacheKey = triggerId + deviceId;
+    if (triggerState.containsKey(cacheKey)) {
+      return triggerState.get(cacheKey);
     }
+    if ((trigger.getEntityType().equals(ScopeEnum.PART) && trigger.getEntityId().contains(deviceId))
+        || (trigger.getEntityType().equals(ScopeEnum.ALL)
+            && trigger.getDeviceProfileId().equals(profileId))) {
+      TriggerState state = createTriggerState(deviceId, trigger.getTriggerCondition());
+      triggerState.put(cacheKey, state);
+      return state;
+    }
+    return null;
+  }
 
-    protected TriggerState getOrCreateConditionState(TkDoConditionEntity condition, String profileId,String deviceId) {
-        String cacheKey = condition.getId() + deviceId;
-        if (conditionState.containsKey(cacheKey)) {
-            return conditionState.get(cacheKey);
+  protected TriggerState getOrCreateClearState(String deviceId, String key) {
+    String cacheKey = deviceId + key;
+    if (triggerState.containsKey(cacheKey)) {
+      return triggerState.get(cacheKey);
+    }
+    TriggerState state = null;
+    ActionAlarmDTO alarm =
+        JacksonUtil.convertValue(alarmAction.getDoContext(), ActionAlarmDTO.class);
+    if (alarm != null && alarm.getClearRule() != null) {
+      for (TriggerDTO rule : alarm.getClearRule()) {
+        if ((ScopeEnum.PART.equals(rule.getEntityType()) && !rule.getEntityId().contains(deviceId))
+        //                        || !alarmAction.getDeviceId().contains(deviceId)
+        ) {
+          continue;
         }
-        if ((condition.getEntityType().equals(ScopeEnum.PART) && condition.getEntityId().contains(deviceId) )
-                ||(condition.getEntityType().equals(ScopeEnum.ALL) && condition.getDeviceProfileId().equals(profileId))  ) {
-            TriggerState state = createTriggerState(deviceId, condition.getTriggerCondition());
+
+        for (AlarmConditionFilter filter :
+            rule.getTriggerCondition().getCondition().getCondition()) {
+          String tempKey = filter.getKey().getKey();
+          if (key.equals(tempKey)) {
+            state = createTriggerState(deviceId, rule.getTriggerCondition());
             triggerState.put(cacheKey, state);
-            return state;
+            break;
+          }
         }
-        return null;
-    }
-
-    @NotNull
-    private TriggerState createTriggerState(String deviceId, AlarmRule rule) {
-        Set<AlarmConditionFilterKey> filterKeys = new HashSet<>();
-        for (AlarmConditionFilter filter : rule.getCondition().getCondition()) {
-            filterKeys.add(filter.getKey());
+        if (state != null) {
+          break;
         }
-        TriggerState state = new TriggerState(deviceId, rule, filterKeys, rule.getAlarmDetails(), null);
-
-        return state;
+      }
+      return state;
     }
-
-    private void pushMsg(TbContext ctx, TbMsg msg, TkDoActionEntity action, String detail) {
-        TbMsgMetaData metaData = //lastMsgMetaData != null ? lastMsgMetaData.copy() :
-                new TbMsgMetaData();
-        String relationType = "";
-        TbMsg newMsg = null;
-        switch (action.getOutTarget()) {
-            case DEVICE_OUT:
-                List<String> rpcDevices = action.getDeviceId();
-                if(ScopeEnum.ALL.equals(action.getEntityType())){
-                    rpcDevices = ytDeviceService.rpcDevices(action.getTenantId(),orgId,action.getDeviceProfileId());
-                }
-                rpcMsg(ctx, msg,rpcDevices , action.getDoContext());
-                break;
-            case SCENE_ACT:
-                reactMsg(ctx, msg, action);
-                break;
-            default:
-                ctx.tellSuccess(msg);
-                break;
-        }
+    return null;
+  }
+
+  protected TriggerState getOrCreateConditionState(
+      TkDoConditionEntity condition, String profileId, String deviceId) {
+    String cacheKey = condition.getId() + deviceId;
+    if (conditionState.containsKey(cacheKey)) {
+      return conditionState.get(cacheKey);
     }
-
-    /**
-     * 设备输出
-     *
-     * @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();
-        metaData.putValue(DataConstants.PERSISTENT,"true");
-        for (String id : devices) {
-            TbMsg newMsg = ctx.newMsg(lastMsgQueueName != null ? lastMsgQueueName : ServiceQueue.MAIN
-                    , msg.getType()
-                    , new DeviceId(UUID.fromString(id))
-                    , msg != null ? msg.getCustomerId() : null
-                    , metaData
-                    , JacksonUtil.toString(context));
-            ctx.tellNext(newMsg, "RPC Request");
-        }
+    if ((condition.getEntityType().equals(ScopeEnum.PART)
+            && condition.getEntityId().contains(deviceId))
+        || (condition.getEntityType().equals(ScopeEnum.ALL)
+            && condition.getDeviceProfileId().equals(profileId))) {
+      TriggerState state = createTriggerState(deviceId, condition.getTriggerCondition());
+      triggerState.put(cacheKey, state);
+      return state;
     }
-
-    /**
-     * 告警通知
-     *
-     * @param ctx
-     * @param msg
-     * @param action
-     */
-    private void noticeMsg(TbContext ctx, TbMsg msg, TkDoActionEntity action, String deviceId, String detailStr, long startTs) {
-
-        DeviceId entityId = new DeviceId(UUID.fromString(deviceId));
-        ActionAlarmDTO actionAlarm = JacksonUtil.convertValue(action.getDoContext(), ActionAlarmDTO.class);
-
-
-        Alarm currentAlarm = new Alarm();
-        currentAlarm.setType(reactName);
-        currentAlarm.setStatus(AlarmStatus.ACTIVE_UNACK);
-        currentAlarm.setSeverity(actionAlarm.getAlarmLevel());
-        if (startTs == 0L) {
-            startTs = System.currentTimeMillis();
-        }
-        currentAlarm.setStartTs(startTs);
-        currentAlarm.setEndTs(currentAlarm.getStartTs());
-        ObjectNode detailData = JacksonUtil.newObjectNode();
-        if(StringUtils.isNotEmpty(detailStr)){
-            detailData.put("msg", detailStr);
-        }
-        detailData.put("data", JacksonUtil.toJsonNode(msg.getData()));
-        currentAlarm.setDetails(detailData);
-        currentAlarm.setOriginator(entityId);
-        currentAlarm.setTenantId(ctx.getTenantId());
-        currentAlarm.setPropagate(false);
-
-        currentAlarm = ctx.getAlarmService().createOrUpdateAlarm(currentAlarm);
-        Alarm oldAlarm = currentAlarms.get(deviceId);
-        if(oldAlarm != null
-                && currentAlarm.getId().equals(oldAlarm.getId())
-                && currentAlarm.getSeverity().equals(oldAlarm.getSeverity())){
-            return;
-        }
-        ytDeviceService.freshAlarmStatus(entityId, 1);
-        currentAlarms.put(deviceId, currentAlarm);
-        alarmMsg(ctx, msg, currentAlarm, "Alarm Cleared");
-
-        AlarmInfoDTO formData = new AlarmInfoDTO();
-        formData.setDeviceName(msg.getMetaData().getData().get("deviceName"));
-        formData.setDetails(JacksonUtil.toString(detailData));
-        formData.setType(currentAlarm.getType());
-        formData.setCreateTs(currentAlarm.getCreatedTime());
-        formData.setStartTs(currentAlarm.getStartTs());
-        formData.setEndTs(currentAlarm.getEndTs());
-        formData.setStatus(currentAlarm.getStatus().name());
-        formData.setDeviceId(deviceId);
-        formData.setTenantId(currentAlarm.getTenantId().getId().toString());
-        formData.setSeverity(TkAlarmSeverity.getLabel(currentAlarm.getSeverity().name()));
-        noticeService.alert(action.getAlarmProfileId(), formData);
+    return null;
+  }
+
+  @NotNull
+  private TriggerState createTriggerState(String deviceId, AlarmRule rule) {
+    Set<AlarmConditionFilterKey> filterKeys = new HashSet<>();
+    for (AlarmConditionFilter filter : rule.getCondition().getCondition()) {
+      filterKeys.add(filter.getKey());
     }
-
-    private void clearAlarm(TbContext ctx, TbMsg msg, String deviceId, String key) throws ExecutionException, InterruptedException {
-        TriggerState clearState = getOrCreateClearState(deviceId, key);
-        if (clearState != null && clearState.process(ctx, msg)) {
-            ctx.getAlarmService().clearAlarmForResult(ctx.getTenantId(), currentAlarms.get(deviceId).getId(), null, System.currentTimeMillis());
-            ytDeviceService.freshAlarmStatus(new DeviceId(UUID.fromString(deviceId)), 0);
-            alarmMsg(ctx, msg, currentAlarms.get(deviceId), "Alarm Cleared");
-            currentAlarms.remove(deviceId);
+    TriggerState state = new TriggerState(deviceId, rule, filterKeys, rule.getAlarmDetails(), null);
+
+    return state;
+  }
+
+  private void pushMsg(TbContext ctx, TbMsg msg, TkDoActionEntity action, String detail) {
+    TbMsgMetaData metaData = // lastMsgMetaData != null ? lastMsgMetaData.copy() :
+        new TbMsgMetaData();
+    String relationType = "";
+    TbMsg newMsg = null;
+    switch (action.getOutTarget()) {
+      case DEVICE_OUT:
+        List<String> rpcDevices = action.getDeviceId();
+        if (ScopeEnum.ALL.equals(action.getEntityType())) {
+          rpcDevices =
+              ytDeviceService.rpcDevices(action.getTenantId(), orgId, action.getDeviceProfileId());
         }
+        rpcMsg(ctx, msg, rpcDevices, action.getDoContext(), action.getCallType());
+        break;
+      case SCENE_ACT:
+        reactMsg(ctx, msg, action);
+        break;
+      default:
+        ctx.tellSuccess(msg);
+        break;
     }
-
-    private void alarmMsg(TbContext ctx, TbMsg msg, Alarm alarm, String releationType) {
-        String lastMsgQueueName = msg.getQueueName();
-        TbMsgMetaData metaData = msg.getMetaData();
-        String data = JacksonUtil.valueToTree(alarm).toString();
-        TbMsg newMsg = ctx.newMsg(lastMsgQueueName != null ? lastMsgQueueName : ServiceQueue.MAIN, "ALARM",
-                alarm.getOriginator(), msg != null ? msg.getCustomerId() : null, metaData, data);
-        ctx.enqueueForTellNext(newMsg, releationType);
+  }
+
+  /**
+   * 设备输出
+   *
+   * @param ctx
+   * @param msg
+   * @param devices
+   * @param context
+   */
+  private void rpcMsg(
+      TbContext ctx, TbMsg msg, List<String> devices, JsonNode context, CallTypeEnum callType) {
+    String lastMsgQueueName = msg.getQueueName();
+    TbMsgMetaData metaData = msg.getMetaData();
+    metaData.putValue(DataConstants.PERSISTENT, "true");
+    metaData.putValue(
+        FastIotConstants.Rpc.ONEWAY, (callType == CallTypeEnum.SYNC ? true : false) + "");
+    for (String id : devices) {
+      TbMsg newMsg =
+          ctx.newMsg(
+              lastMsgQueueName != null ? lastMsgQueueName : ServiceQueue.MAIN,
+              msg.getType(),
+              new DeviceId(UUID.fromString(id)),
+              msg != null ? msg.getCustomerId() : null,
+              metaData,
+              JacksonUtil.toString(context));
+      ctx.tellNext(newMsg, "RPC Request");
     }
-
-
-    private void reactMsg(TbContext ctx, TbMsg msg, TkDoActionEntity action) {
-        //TODO: 场景联动关联消息通知
-        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
-                , JacksonUtil.toString(action.getDoContext()));
-        ctx.tellNext(newMsg, "Message");
+  }
+
+  /**
+   * 告警通知
+   *
+   * @param ctx
+   * @param msg
+   * @param action
+   */
+  private void noticeMsg(
+      TbContext ctx,
+      TbMsg msg,
+      TkDoActionEntity action,
+      String deviceId,
+      String detailStr,
+      long startTs) {
+
+    DeviceId entityId = new DeviceId(UUID.fromString(deviceId));
+    ActionAlarmDTO actionAlarm =
+        JacksonUtil.convertValue(action.getDoContext(), ActionAlarmDTO.class);
+
+    Alarm currentAlarm = new Alarm();
+    currentAlarm.setType(reactName);
+    currentAlarm.setStatus(AlarmStatus.ACTIVE_UNACK);
+    currentAlarm.setSeverity(actionAlarm.getAlarmLevel());
+    if (startTs == 0L) {
+      startTs = System.currentTimeMillis();
+    }
+    currentAlarm.setStartTs(startTs);
+    currentAlarm.setEndTs(currentAlarm.getStartTs());
+    ObjectNode detailData = JacksonUtil.newObjectNode();
+    if (StringUtils.isNotEmpty(detailStr)) {
+      detailData.put("msg", detailStr);
+    }
+    detailData.put("data", JacksonUtil.toJsonNode(msg.getData()));
+    currentAlarm.setDetails(detailData);
+    currentAlarm.setOriginator(entityId);
+    currentAlarm.setTenantId(ctx.getTenantId());
+    currentAlarm.setPropagate(false);
+
+    currentAlarm = ctx.getAlarmService().createOrUpdateAlarm(currentAlarm);
+    Alarm oldAlarm = currentAlarms.get(deviceId);
+    if (oldAlarm != null
+        && currentAlarm.getId().equals(oldAlarm.getId())
+        && currentAlarm.getSeverity().equals(oldAlarm.getSeverity())) {
+      return;
+    }
+    ytDeviceService.freshAlarmStatus(entityId, 1);
+    currentAlarms.put(deviceId, currentAlarm);
+    alarmMsg(ctx, msg, currentAlarm, "Alarm Cleared");
+
+    AlarmInfoDTO formData = new AlarmInfoDTO();
+    formData.setDeviceName(msg.getMetaData().getData().get("deviceName"));
+    formData.setDetails(JacksonUtil.toString(detailData));
+    formData.setType(currentAlarm.getType());
+    formData.setCreateTs(currentAlarm.getCreatedTime());
+    formData.setStartTs(currentAlarm.getStartTs());
+    formData.setEndTs(currentAlarm.getEndTs());
+    formData.setStatus(currentAlarm.getStatus().name());
+    formData.setDeviceId(deviceId);
+    formData.setTenantId(currentAlarm.getTenantId().getId().toString());
+    formData.setSeverity(TkAlarmSeverity.getLabel(currentAlarm.getSeverity().name()));
+    noticeService.alert(action.getAlarmProfileId(), formData);
+  }
+
+  private void clearAlarm(TbContext ctx, TbMsg msg, String deviceId, String key)
+      throws ExecutionException, InterruptedException {
+    TriggerState clearState = getOrCreateClearState(deviceId, key);
+    if (clearState != null && clearState.process(ctx, msg)) {
+      ctx.getAlarmService()
+          .clearAlarmForResult(
+              ctx.getTenantId(),
+              currentAlarms.get(deviceId).getId(),
+              null,
+              System.currentTimeMillis());
+      ytDeviceService.freshAlarmStatus(new DeviceId(UUID.fromString(deviceId)), 0);
+      alarmMsg(ctx, msg, currentAlarms.get(deviceId), "Alarm Cleared");
+      currentAlarms.remove(deviceId);
     }
+  }
+
+  private void alarmMsg(TbContext ctx, TbMsg msg, Alarm alarm, String releationType) {
+    String lastMsgQueueName = msg.getQueueName();
+    TbMsgMetaData metaData = msg.getMetaData();
+    String data = JacksonUtil.valueToTree(alarm).toString();
+    TbMsg newMsg =
+        ctx.newMsg(
+            lastMsgQueueName != null ? lastMsgQueueName : ServiceQueue.MAIN,
+            "ALARM",
+            alarm.getOriginator(),
+            msg != null ? msg.getCustomerId() : null,
+            metaData,
+            data);
+    ctx.enqueueForTellNext(newMsg, releationType);
+  }
+
+  private void reactMsg(TbContext ctx, TbMsg msg, TkDoActionEntity action) {
+    // TODO: 场景联动关联消息通知
+    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,
+            JacksonUtil.toString(action.getDoContext()));
+    ctx.tellNext(newMsg, "Message");
+  }
 }