Merge branch 'ljl022510' into 'master'

Ljl022510

See merge request huang/thingsboard3.3.2!50

@@ -74,7 +74,7 @@ public class YtDeviceController extends BaseController {
     }
     DeviceDTO newDeviceDTO = null;
     boolean isIncludeRelation = false;
-    String gateWayDeviceId = deviceDTO.getGateWayDeviceId();
+    String gateWayDeviceId = deviceDTO.getGatewayId();
     DeviceDTO gateWayDevice = null;
     if (StringUtils.isNotEmpty(gateWayDeviceId)) {
       gateWayDevice =

@@ -38,7 +38,8 @@ public class DeviceDTO extends TenantDTO {
   private String profileId;
 
   @ApiModelProperty(value = "关联网关设备")
-  private String gateWayDeviceId;
+  private String gatewayId;
+  private String gatewayName;
 
   @ApiModelProperty(value = "设备凭证")
   private YtCredentialsDto deviceToken;

@@ -24,6 +24,7 @@ public class YtDevice extends TenantBaseEntity {
   private JsonNode deviceInfo;
   private String profileId;
   private String tbDeviceId;
+  private String gatewayId;
   private String deviceTypeId;
   private String label;
   @TableField(typeHandler = EnumTypeHandler.class)

@@ -25,6 +25,8 @@
         <result property="description" column="description"/>
         <result property="customerId" column="customer_id"/>
         <result property="organizationId" column="organization_id"/>
+        <result property="gatewayId" column="gateway_id"/>
+        <result property="gatewayName" column="gateway_name"/>
         <association property="deviceProfile" javaType="org.thingsboard.server.common.data.yunteng.dto.DeviceProfileDTO">
             <result property="name" column="profile_name"/>
             <result property="transportType" column="transport_type"/>
@@ -41,6 +43,7 @@
         <result property="createdTime" column="created_time"/>
         <result property="lastOnlineTime" column="last_online_time"/>
     </resultMap>
+
     <sql id="basicColumns">
         ifd.id,ifd.name,ifd.device_info,ifd.profile_id,ifd.active_time,ifd.tenant_id,ifd.description
         ,ifd.tb_device_id,ifd.label,ifd.last_connect_time,ifd.device_type,ifd.device_state,ifd.create_time,ifd.update_time,ifd.creator,
@@ -48,6 +51,7 @@
     </sql>
     <sql id="detailColumns">
         <include refid="basicColumns"/>
+        ,ifd.gateway_id,idg.name gateway_name
         ,ifdp.name AS profile_name,ifdp.transport_type
         ,io.name AS organization_name
     </sql>
@@ -109,6 +113,7 @@
         FROM iotfs_device ifd
         LEFT JOIN device_profile ifdp ON ifd.profile_id = CAST (ifdp.id AS VARCHAR)
         LEFT JOIN iotfs_organization io ON io.id = ifd.organization_id
+        LEFT JOIN iotfs_device idg ON idg.id = ifd.gateway_id
         <where>
             <if test="tenantId !=null and tenantId !=''">
                 AND ifd.tenant_id = #{tenantId}

@@ -4,31 +4,19 @@
 package org.thingsboard.rule.engine.yunteng.scene;
 
 import com.fasterxml.jackson.databind.JsonNode;
-import com.google.gson.JsonParser;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.rule.engine.action.TbAlarmResult;
 import org.thingsboard.rule.engine.api.TbContext;
-import org.thingsboard.rule.engine.profile.*;
-import org.thingsboard.rule.engine.profile.state.PersistedAlarmRuleState;
-import org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNode;
-import org.thingsboard.rule.engine.yunteng.utils.TriggerRuleState;
 import org.thingsboard.server.common.data.DataConstants;
-import org.thingsboard.server.common.data.device.profile.*;
-import org.thingsboard.server.common.data.exception.ApiUsageLimitsExceededException;
-import org.thingsboard.server.common.data.id.EntityId;
-import org.thingsboard.server.common.data.id.TenantId;
-import org.thingsboard.server.common.data.kv.AttributeKvEntry;
-import org.thingsboard.server.common.data.kv.KvEntry;
-import org.thingsboard.server.common.data.query.EntityKeyType;
+import org.thingsboard.server.common.data.device.profile.AlarmCondition;
+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.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.common.msg.session.SessionMsgType;
-import org.thingsboard.server.common.transport.adaptor.JsonConverter;
 import org.thingsboard.server.dao.yunteng.entities.DoAction;
 import org.thingsboard.server.dao.yunteng.entities.DoCondition;
 import org.thingsboard.server.dao.yunteng.service.DoActionService;
@@ -36,6 +24,7 @@ import org.thingsboard.server.dao.yunteng.service.DoConditionService;
 import org.thingsboard.server.dao.yunteng.service.TriggerService;
 
 import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 
 @Slf4j
@@ -48,7 +37,7 @@ class ReactState {
      * 键：设备主键
      * 值：设备指标参与的触发器
      */
-    private Map<String, TriggerState> triggerState;
+    private ConcurrentHashMap<String, TriggerState> triggerState = new ConcurrentHashMap<>();
 
 
 
@@ -78,21 +67,17 @@ class ReactState {
 
 
     public void process(TbContext ctx, TbMsg msg,String deviceId) throws ExecutionException, InterruptedException {
-        TriggerState triggerRuleState = getOrCreateTriggerState(deviceId);
+        TriggerState triggerState = getOrCreateTriggerState(deviceId);
         boolean matched = false;
-        if(triggerRuleState != null && actions != null){
-            if(msg.getType().equals(SessionMsgType.POST_TELEMETRY_REQUEST.name())){
-                matched = processTelemetry(ctx,msg,triggerRuleState);
-            }else{
-                matched = processAttributes(ctx,msg,triggerRuleState);
-            }
+        if(triggerState != null && actions != null){
+            matched = triggerState.process(ctx,msg);
         }
 
         if(matched){
             // TODO 执行条件
         }
 
-        if(true){
+        if(matched){
             // TODO 输出动作
             for(DoAction item: actions){
 //                pushMsg();
@@ -103,69 +88,26 @@ class ReactState {
 
 
 
-    private boolean processAttributes(TbContext ctx, TbMsg msg,TriggerState triggerRuleState) throws ExecutionException, InterruptedException {
-        boolean stateChanged = false;
-        Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData()));
-        if (!attributes.isEmpty()) {
-            SnapshotUpdate update = triggerRuleState.merge(attributes);
-            stateChanged |= triggerRuleState.process(ctx, msg, update);
-
-        }
-        return stateChanged;
-    }
-
-    protected boolean processTelemetry(TbContext ctx, TbMsg msg,TriggerState triggerRuleState) throws ExecutionException, InterruptedException {
-        boolean stateChanged = false;
-        Map<Long, List<KvEntry>> tsKvMap = JsonConverter.convertToSortedTelemetry(new JsonParser().parse(msg.getData()), TbMsgTimeseriesNode.getTs(msg));
-        for (Map.Entry<Long, List<KvEntry>> entry : tsKvMap.entrySet()) {
-            Long ts = entry.getKey();
-            List<KvEntry> data = entry.getValue();
-            SnapshotUpdate update = triggerRuleState.merge( ts, data);
-            if (update.hasUpdate()) {
-                stateChanged |= triggerRuleState.process(ctx, msg, update);
-            }
-        }
-
-        return stateChanged;
-    }
-
-
-
-    private static EntityKeyType getKeyTypeFromScope(String scope) {
-        switch (scope) {
-            case DataConstants.CLIENT_SCOPE:
-                return EntityKeyType.CLIENT_ATTRIBUTE;
-            case DataConstants.SHARED_SCOPE:
-                return EntityKeyType.SHARED_ATTRIBUTE;
-            case DataConstants.SERVER_SCOPE:
-                return EntityKeyType.SERVER_ATTRIBUTE;
-        }
-        return EntityKeyType.ATTRIBUTE;
-    }
-
-
-
 
 
 
     protected TriggerState getOrCreateTriggerState(String deviceId) {
-        TriggerState triggerRuleState = triggerState.computeIfAbsent(deviceId
-                ,a -> {
-                    TriggerService triggerService = SpringBeanUtils.getBean(TriggerService.class);
-                    TriggerDTO trigger =triggerService.getTrigger(reactId,deviceId);
-                    if(trigger == null){
-                        return null;
-                    }
-                    AlarmCondition condition = trigger.getTriggerCondition();
-                    Set<AlarmConditionFilterKey> filterKeys = new HashSet<>();
-                    for(AlarmConditionFilter filter :condition.getCondition()){
-                        filterKeys.add(filter.getKey());
-                    }
-                    TriggerState state = new TriggerState(deviceId,condition, filterKeys,null);
-                    triggerState.put(a, state);
-                    return state;
-                });
-        return triggerRuleState;
+        if(triggerState.containsKey(deviceId)){
+            return triggerState.get(deviceId);
+        }
+        TriggerService triggerService = SpringBeanUtils.getBean(TriggerService.class);
+        TriggerDTO trigger =triggerService.getTrigger(reactId,deviceId);
+        if(trigger != null){
+            AlarmCondition condition = trigger.getTriggerCondition();
+            Set<AlarmConditionFilterKey> filterKeys = new HashSet<>();
+            for(AlarmConditionFilter filter :condition.getCondition()){
+                filterKeys.add(filter.getKey());
+            }
+            TriggerState state = new TriggerState(deviceId,condition, filterKeys,null);
+            triggerState.put(deviceId, state);
+            return  state;
+        }
+        return null;
     }
 
     private void pushMsg(TbContext ctx, TbMsg msg, TbAlarmResult alarmResult,DoAction action) {

@@ -18,6 +18,7 @@ package org.thingsboard.rule.engine.yunteng.scene;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.gson.JsonParser;
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
@@ -28,6 +29,7 @@ import org.thingsboard.rule.engine.api.TbContext;
 import org.thingsboard.rule.engine.profile.*;
 import org.thingsboard.rule.engine.profile.state.PersistedAlarmRuleState;
 import org.thingsboard.rule.engine.profile.state.PersistedAlarmState;
+import org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNode;
 import org.thingsboard.rule.engine.yunteng.utils.TriggerRuleState;
 import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.Device;
@@ -46,6 +48,8 @@ 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.common.msg.session.SessionMsgType;
+import org.thingsboard.server.common.transport.adaptor.JsonConverter;
 import org.thingsboard.server.dao.alarm.AlarmOperationResult;
 import org.thingsboard.server.dao.sql.query.EntityKeyMapping;
 import org.thingsboard.server.dao.yunteng.service.TriggerService;
@@ -64,7 +68,6 @@ class TriggerState {
     private volatile boolean initialFetchDone;
     private volatile TbMsgMetaData lastMsgMetaData;
     private volatile String lastMsgQueueName;
-    private volatile DataSnapshot dataSnapshot;
     private final DynamicPredicateValueCtx dynamicPredicateValueCtx;
     private DataSnapshot latestValues;
 
@@ -79,18 +82,29 @@ class TriggerState {
     }
 
 
-    public boolean process(TbContext ctx, TbMsg msg,  SnapshotUpdate update) throws ExecutionException, InterruptedException {
+    public boolean process(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException {
         if (latestValues == null) {
             latestValues = fetchLatestValues(ctx, originator);
         }
         lastMsgMetaData = msg.getMetaData();
         lastMsgQueueName = msg.getQueueName();
-        this.dataSnapshot = latestValues;
-        return createOrClearAlarms(ctx, msg, latestValues, update, TriggerRuleState::eval);
+        SnapshotUpdate update = null;
+        if(msg.getType().equals(SessionMsgType.POST_TELEMETRY_REQUEST.name())){
+            update =  processTelemetry(ctx,msg);
+        }else{
+            update =   processAttributes(ctx,msg);
+        }
+
+        if (update != null && update.hasUpdate()) {
+            return createOrClearAlarms(ctx, msg, latestValues, update, TriggerRuleState::eval);
+        }
+        return false;
     }
 
+
+
     public boolean process(TbContext ctx, long ts) throws ExecutionException, InterruptedException {
-        return createOrClearAlarms(ctx, null, ts, null, (alarmState, tsParam) -> alarmState.eval(tsParam, dataSnapshot));
+        return createOrClearAlarms(ctx, null, ts, null, (alarmState, tsParam) -> alarmState.eval(tsParam, latestValues));
     }
 
     public <T> boolean createOrClearAlarms(TbContext ctx, TbMsg msg, T data, SnapshotUpdate update, BiFunction<TriggerRuleState, T, AlarmEvalResult> evalFunction) {
@@ -101,11 +115,21 @@ class TriggerState {
         AlarmEvalResult evalResult = evalFunction.apply(ruleState, data);
         if (AlarmEvalResult.TRUE.equals(evalResult)) {
             stateUpdate = true;
+        } else if (AlarmEvalResult.FALSE.equals(evalResult)) {
+            stateUpdate = clearAlarmState(stateUpdate, ruleState);
         }
         return stateUpdate;
     }
 
 
+    public boolean clearAlarmState(boolean stateUpdate, TriggerRuleState state) {
+        if (state != null) {
+            state.clear();
+            stateUpdate |= state.checkUpdate();
+        }
+        return stateUpdate;
+    }
+
     public boolean validateUpdate(SnapshotUpdate update, TriggerRuleState state) {
         if (update != null) {
             //Check that the update type and that keys match.
@@ -152,7 +176,24 @@ class TriggerState {
 
 
 
+    protected SnapshotUpdate processAttributes(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException {
+        Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData()));
+        if (!attributes.isEmpty()) {
+            return merge(attributes);
+        }
+        return null;
+    }
 
+    protected SnapshotUpdate processTelemetry(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException {
+        Map<Long, List<KvEntry>> tsKvMap = JsonConverter.convertToSortedTelemetry(new JsonParser().parse(msg.getData()), TbMsgTimeseriesNode.getTs(msg));
+        for (Map.Entry<Long, List<KvEntry>> entry : tsKvMap.entrySet()) {
+            Long ts = entry.getKey();
+            List<KvEntry> data = entry.getValue();
+            return merge( ts, data);
+        }
+
+        return null;
+    }
 
 
 
@@ -238,7 +279,7 @@ class TriggerState {
                 throw new RuntimeException("Can't parse entry: " + entry.getDataType());
         }
     }
-    SnapshotUpdate merge( Long newTs, List<KvEntry> data) {
+    private SnapshotUpdate merge( Long newTs, List<KvEntry> data) {
         Set<AlarmConditionFilterKey> keys = new HashSet<>();
         for (KvEntry entry : data) {
             AlarmConditionFilterKey entityKey = new AlarmConditionFilterKey(AlarmConditionKeyType.TIME_SERIES, entry.getKey());
@@ -250,7 +291,7 @@ class TriggerState {
         return new SnapshotUpdate(AlarmConditionKeyType.TIME_SERIES, keys);
     }
 
-    SnapshotUpdate merge(Set<AttributeKvEntry> attributes) {
+    private SnapshotUpdate merge(Set<AttributeKvEntry> attributes) {
         long newTs = 0;
         Set<AlarmConditionFilterKey> keys = new HashSet<>();
         for (AttributeKvEntry entry : attributes) {