Improvements

Andrii Shvaika · Andrew Shvayka
1 parent aec2d388
Showing 10 changed files with 70 additions and 39 deletions
application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java
application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmDataSubCtx.java
application/src/main/java/org/thingsboard/server/service/telemetry/DefaultAlarmSubscriptionService.java
common/data/src/main/java/org/thingsboard/server/common/data/query/AlarmDataQuery.java
common/data/src/main/java/org/thingsboard/server/common/data/query/RelationsQueryFilter.java
dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java
dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultEntityQueryRepository.java
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java
--- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
View file @2eb7949
+++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
View file @2eb7949
@@ -76,6 +76,7 @@ import org.thingsboard.server.service.script.JsExecutorService;
 import org.thingsboard.server.service.script.JsInvokeService;
 import org.thingsboard.server.service.session.DeviceSessionCacheService;
 import org.thingsboard.server.service.state.DeviceStateService;
+import org.thingsboard.server.service.telemetry.AlarmSubscriptionService;
 import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
 import org.thingsboard.server.service.transport.TbCoreToTransportService;
@@ -169,10 +170,6 @@ public class ActorSystemContext {
     @Autowired
     @Getter
-    private AlarmService alarmService;
-
-    @Autowired
-    @Getter
     private RelationService relationService;
     @Autowired
@@ -189,6 +186,10 @@ public class ActorSystemContext {
     @Autowired
     @Getter
+    private AlarmSubscriptionService alarmService;
+
+    @Autowired
+    @Getter
     private JsInvokeService jsSandbox;
     @Autowired
--- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
View file @2eb7949
+++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
View file @2eb7949
@@ -22,6 +22,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.thingsboard.common.util.ListeningExecutor;
 import org.thingsboard.rule.engine.api.MailService;
+import org.thingsboard.rule.engine.api.RuleEngineAlarmService;
 import org.thingsboard.rule.engine.api.RuleEngineRpcService;
 import org.thingsboard.rule.engine.api.RuleEngineTelemetryService;
 import org.thingsboard.rule.engine.api.ScriptEngine;
@@ -351,7 +352,7 @@ class DefaultTbContext implements TbContext {
     }
     @Override
-    public AlarmService getAlarmService() {
+    public RuleEngineAlarmService getAlarmService() {
         return mainCtx.getAlarmService();
     }
--- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java
View file @2eb7949
+++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java
View file @2eb7949
@@ -266,16 +266,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
             AlarmDataUpdate update = new AlarmDataUpdate(cmd.getCmdId(), new PageData<>(Collections.emptyList(), 1, 0, false), null);
             wsService.sendWsMsg(ctx.getSessionId(), update);
         } else {
-            ctx.setLastFetchTs(System.currentTimeMillis());
-            PageData<AlarmData> alarms = alarmService.findAlarmDataByQueryForEntities(ctx.getTenantId(), ctx.getCustomerId(),
-                    ctx.getQuery().getPageLink(), ctx.getOrderedEntityIds());
-            alarms = ctx.setAndMergeAlarmsData(alarms);
-            AlarmDataUpdate update = new AlarmDataUpdate(cmd.getCmdId(), alarms, null);
-            wsService.sendWsMsg(ctx.getSessionId(), update);
-            if (adq.getPageLink().getTimeWindow() > 0) {
-                //TODO: refresh list of entities periodically (similar to time-series subscription).
-                createAlarmSubscriptions(ctx);
-            }
+            ctx.fetchAlarmsAndCreateSubscriptions();
         }
     }
@@ -321,7 +312,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
     private TbAlarmDataSubCtx createSubCtx(TelemetryWebSocketSessionRef sessionRef, AlarmDataCmd cmd) {
         Map<Integer, TbAbstractDataSubCtx> sessionSubs = subscriptionsBySessionId.computeIfAbsent(sessionRef.getSessionId(), k -> new HashMap<>());
-        TbAlarmDataSubCtx ctx = new TbAlarmDataSubCtx(serviceId, wsService, sessionRef, cmd.getCmdId());
+        TbAlarmDataSubCtx ctx = new TbAlarmDataSubCtx(serviceId, wsService, localSubscriptionService, alarmService, sessionRef, cmd.getCmdId());
         ctx.setQuery(cmd.getQuery());
         sessionSubs.put(cmd.getCmdId(), ctx);
         return ctx;
@@ -467,11 +458,6 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
         createTelemetrySubscriptions(ctx, keys, true);
     }
-    private void createAlarmSubscriptions(TbAlarmDataSubCtx ctx) {
-        List<TbSubscription> subscriptions = ctx.createSubscriptions();
-        subscriptions.forEach(localSubscriptionService::addSubscription);
-    }
-
     private void createTelemetrySubscriptions(TbEntityDataSubCtx ctx, List<EntityKey> keys, boolean latest) {
         List<TbSubscription> tbSubs = ctx.createSubscriptions(keys, latest);
         tbSubs.forEach(sub -> localSubscriptionService.addSubscription(sub));
--- a/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmDataSubCtx.java
View file @2eb7949
+++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmDataSubCtx.java
View file @2eb7949
@@ -27,8 +27,10 @@ import org.thingsboard.server.common.data.query.AlarmData;
 import org.thingsboard.server.common.data.query.AlarmDataPageLink;
 import org.thingsboard.server.common.data.query.AlarmDataQuery;
 import org.thingsboard.server.common.data.query.EntityData;
+import org.thingsboard.server.dao.alarm.AlarmService;
 import org.thingsboard.server.service.telemetry.TelemetryWebSocketService;
 import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef;
+import org.thingsboard.server.service.telemetry.cmd.v2.AlarmDataCmd;
 import org.thingsboard.server.service.telemetry.cmd.v2.AlarmDataUpdate;
 import org.thingsboard.server.service.telemetry.sub.AlarmSubscriptionUpdate;
@@ -45,6 +47,8 @@ import java.util.stream.Collectors;
 @Slf4j
 public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx<AlarmDataQuery> {
+    private final TbLocalSubscriptionService localSubscriptionService;
+    private final AlarmService alarmService;
     @Getter
     @Setter
     private final LinkedHashMap<EntityId, EntityData> entitiesMap;
@@ -59,15 +63,32 @@ public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx<AlarmDataQuery> {
     private boolean tooManyEntities;
     private Map<Integer, EntityId> subToEntityIdMap;
-    @Setter
-    private long lastFetchTs;
-    public TbAlarmDataSubCtx(String serviceId, TelemetryWebSocketService wsService, TelemetryWebSocketSessionRef sessionRef, int cmdId) {
+    public TbAlarmDataSubCtx(String serviceId, TelemetryWebSocketService wsService,
+                             TbLocalSubscriptionService localSubscriptionService,
+                             AlarmService alarmService,
+                             TelemetryWebSocketSessionRef sessionRef, int cmdId) {
         super(serviceId, wsService, sessionRef, cmdId);
+        this.localSubscriptionService = localSubscriptionService;
+        this.alarmService = alarmService;
         this.entitiesMap = new LinkedHashMap<>();
         this.alarmsMap = new HashMap<>();
     }
+    public void fetchAlarmsAndCreateSubscriptions() {
+        PageData<AlarmData> alarms = alarmService.findAlarmDataByQueryForEntities(getTenantId(), getCustomerId(),
+                query.getPageLink(), getOrderedEntityIds());
+        alarms = setAndMergeAlarmsData(alarms);
+        AlarmDataUpdate update = new AlarmDataUpdate(cmdId, alarms, null);
+        wsService.sendWsMsg(getSessionId(), update);
+        if (query.getPageLink().getTimeWindow() > 0) {
+            clearSubscriptions();
+            //TODO: refresh list of entities periodically (similar to time-series subscription).
+            List<TbSubscription> subscriptions = createSubscriptions();
+            subscriptions.forEach(localSubscriptionService::addSubscription);
+        }
+    }
+
     public void setEntitiesData(PageData<EntityData> entitiesData) {
         entitiesMap.clear();
         tooManyEntities = entitiesData.hasNext();
@@ -99,31 +120,42 @@ public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx<AlarmDataQuery> {
     public List<TbSubscription> createSubscriptions() {
         this.subToEntityIdMap = new HashMap<>();
         AlarmDataPageLink pageLink = query.getPageLink();
+        long startTs = System.currentTimeMillis() - pageLink.getTimeWindow();
         List<TbSubscription> result = new ArrayList<>();
         for (EntityData entityData : entitiesMap.values()) {
             int subIdx = sessionRef.getSessionSubIdSeq().incrementAndGet();
             subToEntityIdMap.put(subIdx, entityData.getEntityId());
             log.trace("[{}][{}][{}] Creating alarms subscription for [{}] with query: {}", serviceId, cmdId, subIdx, entityData.getEntityId(), pageLink);
             result.add(TbAlarmsSubscription.builder()
+                    .type(TbSubscriptionType.ALARMS)
                     .serviceId(serviceId)
                     .sessionId(sessionRef.getSessionId())
                     .subscriptionId(subIdx)
                     .tenantId(sessionRef.getSecurityCtx().getTenantId())
                     .entityId(entityData.getEntityId())
                     .updateConsumer(this::sendWsMsg)
-                    .ts(lastFetchTs)
+                    .ts(startTs)
                     .build());
         }
         return result;
     }
+    public void clearSubscriptions() {
+        if (subToEntityIdMap != null) {
+            for (Integer subId : subToEntityIdMap.keySet()) {
+                localSubscriptionService.cancelSubscription(sessionRef.getSessionId(), subId);
+            }
+            subToEntityIdMap.clear();
+        }
+    }
+
     private void sendWsMsg(String sessionId, AlarmSubscriptionUpdate subscriptionUpdate) {
         Alarm alarm = subscriptionUpdate.getAlarm();
         AlarmId alarmId = alarm.getId();
         if (subscriptionUpdate.isAlarmDeleted()) {
             Alarm deleted = alarmsMap.remove(alarmId);
             if (deleted != null) {
-                //TODO: invalidate current page;
+                fetchAlarmsAndCreateSubscriptions();
             }
         } else {
             AlarmData current = alarmsMap.get(alarmId);
@@ -131,14 +163,14 @@ public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx<AlarmDataQuery> {
             boolean matchesFilter = filter(alarm);
             if (onCurrentPage) {
                 if (matchesFilter) {
-                    AlarmData updated = new AlarmData(alarm, current.getName(), current.getEntityId());
+                    AlarmData updated = new AlarmData(alarm, current.getOriginatorName(), current.getEntityId());
                     alarmsMap.put(alarmId, updated);
                     wsService.sendWsMsg(sessionId, new AlarmDataUpdate(cmdId, null, Collections.singletonList(updated)));
                 } else {
-                    //TODO: invalidate current page;
+                    fetchAlarmsAndCreateSubscriptions();
                 }
             } else if (matchesFilter && query.getPageLink().getPage() == 0) {
-                //TODO: invalidate current page;
+                fetchAlarmsAndCreateSubscriptions();
             }
         }
     }
--- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultAlarmSubscriptionService.java
View file @2eb7949
+++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultAlarmSubscriptionService.java
View file @2eb7949
@@ -166,7 +166,7 @@ public class DefaultAlarmSubscriptionService extends AbstractSubscriptionService
                 TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId);
                 if (currentPartitions.contains(tpi)) {
                     if (subscriptionManagerService.isPresent()) {
-                        subscriptionManagerService.get().onAlarmUpdate(tenantId, entityId, alarm);
+                        subscriptionManagerService.get().onAlarmUpdate(tenantId, entityId, alarm, TbCallback.EMPTY);
                     } else {
                         log.warn("Possible misconfiguration because subscriptionManagerService is null!");
                     }
@@ -186,7 +186,7 @@ public class DefaultAlarmSubscriptionService extends AbstractSubscriptionService
                 TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId);
                 if (currentPartitions.contains(tpi)) {
                     if (subscriptionManagerService.isPresent()) {
-                        subscriptionManagerService.get().onAlarmDeleted(tenantId, entityId, alarm);
+                        subscriptionManagerService.get().onAlarmDeleted(tenantId, entityId, alarm, TbCallback.EMPTY);
                     } else {
                         log.warn("Possible misconfiguration because subscriptionManagerService is null!");
                     }
--- a/common/data/src/main/java/org/thingsboard/server/common/data/query/AlarmDataQuery.java
View file @2eb7949
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/query/AlarmDataQuery.java
View file @2eb7949
@@ -16,7 +16,6 @@
 package org.thingsboard.server.common.data.query;
 import com.fasterxml.jackson.annotation.JsonIgnore;
-import lombok.Getter;
 import lombok.ToString;
 import java.util.List;
--- a/common/data/src/main/java/org/thingsboard/server/common/data/query/RelationsQueryFilter.java
View file @2eb7949
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/query/RelationsQueryFilter.java
View file @2eb7949
@@ -19,6 +19,7 @@ import lombok.Data;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.relation.EntitySearchDirection;
 import org.thingsboard.server.common.data.relation.EntityTypeFilter;
+import org.thingsboard.server.common.data.relation.RelationTypeGroup;
 import java.util.List;
--- a/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java
View file @2eb7949
+++ b/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java
View file @2eb7949
@@ -61,6 +61,7 @@ public class DefaultAlarmQueryRepository implements AlarmQueryRepository {
         alarmFieldColumnMap.put("severity", ModelConstants.ALARM_SEVERITY_PROPERTY);
         alarmFieldColumnMap.put("originator_id", ModelConstants.ALARM_ORIGINATOR_ID_PROPERTY);
         alarmFieldColumnMap.put("originator_type", ModelConstants.ALARM_ORIGINATOR_TYPE_PROPERTY);
+        alarmFieldColumnMap.put("originator", "originator_name");
     }
     public static final String SELECT_ORIGINATOR_NAME = " CASE" +
@@ -124,7 +125,7 @@ public class DefaultAlarmQueryRepository implements AlarmQueryRepository {
         EntityDataSortOrder sortOrder = pageLink.getSortOrder();
         if (sortOrder != null && sortOrder.getKey().getType().equals(EntityKeyType.ALARM_FIELD)) {
             String sortOrderKey = sortOrder.getKey().getKey();
-            sortPart.append("a.").append(alarmFieldColumnMap.getOrDefault(sortOrderKey, sortOrderKey))
+            sortPart.append(alarmFieldColumnMap.getOrDefault(sortOrderKey, sortOrderKey))
                     .append(" ").append(sortOrder.getDirection().name());
             ctx.addUuidListParameter("entity_ids", orderedEntityIds.stream().map(EntityId::getId).collect(Collectors.toList()));
             if (pageLink.isSearchPropagatedAlarms()) {
--- a/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultEntityQueryRepository.java
View file @2eb7949
+++ b/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultEntityQueryRepository.java
View file @2eb7949
@@ -447,14 +447,21 @@ public class DefaultEntityQueryRepository implements EntityQueryRepository {
                 if (!single) {
                     whereFilter.append(" (");
                 }
-                whereFilter.append(" re.relation_type = :where_relation_type").append(entityTypeFilterIdx).append(" and re.")
-                        .append(entityFilter.getDirection().equals(EntitySearchDirection.FROM) ? "to" : "from")
-                        .append("_type in (:where_entity_types").append(entityTypeFilterIdx).append(")");
+                List<String> whereEntityTypes = etf.getEntityTypes().stream().map(EntityType::name).collect(Collectors.toList());
+                whereFilter
+                        .append(" re.relation_type = :where_relation_type").append(entityTypeFilterIdx);
+                if (!whereEntityTypes.isEmpty()) {
+                    whereFilter.append(" and re.")
+                            .append(entityFilter.getDirection().equals(EntitySearchDirection.FROM) ? "to" : "from")
+                            .append("_type in (:where_entity_types").append(entityTypeFilterIdx).append(")");
+                }
                 if (!single) {
                     whereFilter.append(" )");
                 }
                 ctx.addStringParameter("where_relation_type" + entityTypeFilterIdx, relationType);
-                ctx.addStringListParameter("where_entity_types" + entityTypeFilterIdx, etf.getEntityTypes().stream().map(EntityType::name).collect(Collectors.toList()));
+                if (!whereEntityTypes.isEmpty()) {
+                    ctx.addStringListParameter("where_entity_types" + entityTypeFilterIdx, whereEntityTypes);
+                }
                 entityTypeFilterIdx++;
             }
         } else {
--- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java
View file @2eb7949
+++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java
View file @2eb7949
@@ -30,6 +30,7 @@ import org.mockito.Mock;
 import org.mockito.runners.MockitoJUnitRunner;
 import org.mockito.stubbing.Answer;
 import org.thingsboard.common.util.ListeningExecutor;
+import org.thingsboard.rule.engine.api.RuleEngineAlarmService;
 import org.thingsboard.rule.engine.api.ScriptEngine;
 import org.thingsboard.rule.engine.api.TbContext;
 import org.thingsboard.rule.engine.api.TbNodeConfiguration;
@@ -43,6 +44,7 @@ import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgDataType;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
+import org.thingsboard.server.dao.alarm.AlarmOperationResult;
 import org.thingsboard.server.dao.alarm.AlarmService;
 import javax.script.ScriptException;
@@ -79,7 +81,7 @@ public class TbAlarmNodeTest {
     @Mock
     private TbContext ctx;
     @Mock
-    private AlarmService alarmService;
+    private RuleEngineAlarmService alarmService;
     @Mock
     private ScriptEngine detailsJs;
@@ -289,7 +291,8 @@ public class TbAlarmNodeTest {
         when(detailsJs.executeJsonAsync(msg)).thenReturn(Futures.immediateFuture(null));
         when(alarmService.findLatestByOriginatorAndType(tenantId, originator, "SomeType")).thenReturn(Futures.immediateFuture(activeAlarm));
-        when(alarmService.clearAlarm(eq(activeAlarm.getTenantId()), eq(activeAlarm.getId()), org.mockito.Mockito.any(JsonNode.class), anyLong())).thenReturn(Futures.immediateFuture(true));
+        when(alarmService.clearAlarm(eq(activeAlarm.getTenantId()), eq(activeAlarm.getId()), org.mockito.Mockito.any(JsonNode.class), anyLong()))
+                .thenReturn(Futures.immediateFuture( false));
         when(alarmService.findAlarmByIdAsync(eq(activeAlarm.getTenantId()), eq(activeAlarm.getId()))).thenReturn(Futures.immediateFuture(activeAlarm));
 //        doAnswer((Answer<Alarm>) invocationOnMock -> (Alarm) (invocationOnMock.getArguments())[0]).when(alarmService).createOrUpdateAlarm(activeAlarm);