diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java index fc08ebfac1..0e322f203a 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java @@ -124,7 +124,7 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer @Override public void addSubscription(TbSubscription subscription, TbCallback callback) { - log.trace("[{}][{}][{}] Registering remote subscription for entity [{}]", + log.trace("[{}][{}][{}] Registering subscription for entity [{}]", subscription.getServiceId(), subscription.getSessionId(), subscription.getSubscriptionId(), subscription.getEntityId()); TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, subscription.getTenantId(), subscription.getEntityId()); if (currentPartitions.contains(tpi)) { diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java index 83412acee0..d342446db3 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java @@ -145,6 +145,9 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc if (wsCallBackExecutor != null) { wsCallBackExecutor.shutdownNow(); } + if (scheduler != null) { + scheduler.shutdownNow(); + } } @Override @@ -230,11 +233,12 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc private void refreshDynamicQuery(TenantId tenantId, CustomerId customerId, TbEntityDataSubCtx finalCtx) { try { long start = System.currentTimeMillis(); - Collection oldSubIds = finalCtx.update(entityService.findEntityDataByQuery(tenantId, customerId, finalCtx.getQuery())); + TbEntityDataSubCtx.TbEntityDataSubCtxUpdateResult result = finalCtx.update(entityService.findEntityDataByQuery(tenantId, customerId, finalCtx.getQuery())); long end = System.currentTimeMillis(); dynamicQueryInvocationCnt.incrementAndGet(); dynamicQueryTimeSpent.addAndGet(end - start); - oldSubIds.forEach(subId -> localSubscriptionService.cancelSubscription(serviceId, subId)); + result.getSubsToCancel().forEach(subId -> localSubscriptionService.cancelSubscription(finalCtx.getSessionId(), subId)); + result.getSubsToAdd().forEach(localSubscriptionService::addSubscription); } catch (Exception e) { log.warn("[{}][{}] Failed to refresh query", finalCtx.getSessionId(), finalCtx.getCmdId(), e); } @@ -340,21 +344,6 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc }, wsCallBackExecutor); } - private List getReadTsKvQueries(GetTsCmd cmd) { - List finalTsKvQueryList; - List queries = cmd.getKeys().stream().map(key -> new BaseReadTsKvQuery(key, cmd.getStartTs(), cmd.getEndTs(), cmd.getInterval(), - getLimit(cmd.getLimit()), cmd.getAgg())).collect(Collectors.toList()); - if (cmd.isFetchLatestPreviousPoint()) { - finalTsKvQueryList = new ArrayList<>(queries); - finalTsKvQueryList.addAll(cmd.getKeys().stream().map(key -> new BaseReadTsKvQuery( - key, cmd.getStartTs() - TimeUnit.DAYS.toMillis(365), cmd.getStartTs(), cmd.getInterval(), 1, cmd.getAgg() - )).collect(Collectors.toList())); - } else { - finalTsKvQueryList = queries; - } - return finalTsKvQueryList; - } - private void handleLatestCmd(TbEntityDataSubCtx ctx, LatestValueCmd latestCmd) { log.trace("[{}][{}] Going to process latest command: {}", ctx.getSessionId(), ctx.getCmdId(), latestCmd); //Fetch the latest values for telemetry keys (in case they are not copied from NoSQL to SQL DB in hybrid mode. diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java index 1f60b7aab0..50541ccfcd 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.service.subscription; +import lombok.AllArgsConstructor; import lombok.Data; import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.common.data.id.CustomerId; @@ -62,7 +63,6 @@ public class TbEntityDataSubCtx { private TimeSeriesCmd tsCmd; private PageData data; private boolean initialDataSent; - private List tbSubs; private Map subToEntityIdMap; private volatile ScheduledFuture refreshTask; private TimeSeriesCmd curTsCmd; @@ -93,10 +93,10 @@ public class TbEntityDataSubCtx { public List createSubscriptions(List keys, boolean resultToLatestValues) { this.subToEntityIdMap = new HashMap<>(); - tbSubs = new ArrayList<>(); + List tbSubs = new ArrayList<>(); Map> keysByType = getEntityKeyByTypeMap(keys); for (EntityData entityData : data.getData()) { - addSubscription(entityData, keysByType, resultToLatestValues); + tbSubs.addAll(addSubscriptions(entityData, keysByType, resultToLatestValues)); } return tbSubs; } @@ -107,33 +107,35 @@ public class TbEntityDataSubCtx { return keysByType; } - private void addSubscription(EntityData entityData, Map> keysByType, boolean resultToLatestValues) { + private List addSubscriptions(EntityData entityData, Map> keysByType, boolean resultToLatestValues) { + List subscriptionList = new ArrayList<>(); keysByType.forEach((keysType, keysList) -> { int subIdx = sessionRef.getSessionSubIdSeq().incrementAndGet(); subToEntityIdMap.put(subIdx, entityData.getEntityId()); switch (keysType) { case TIME_SERIES: - tbSubs.add(createTsSub(entityData, subIdx, keysList, resultToLatestValues)); + subscriptionList.add(createTsSub(entityData, subIdx, keysList, resultToLatestValues)); break; case CLIENT_ATTRIBUTE: - tbSubs.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.CLIENT_SCOPE, keysList)); + subscriptionList.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.CLIENT_SCOPE, keysList)); break; case SHARED_ATTRIBUTE: - tbSubs.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.SHARED_SCOPE, keysList)); + subscriptionList.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.SHARED_SCOPE, keysList)); break; case SERVER_ATTRIBUTE: - tbSubs.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.SERVER_SCOPE, keysList)); + subscriptionList.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.SERVER_SCOPE, keysList)); break; case ATTRIBUTE: - tbSubs.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.ANY_SCOPE, keysList)); + subscriptionList.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.ANY_SCOPE, keysList)); break; } }); + return subscriptionList; } private TbSubscription createAttrSub(EntityData entityData, int subIdx, EntityKeyType keysType, TbAttributeSubscriptionScope scope, List subKeys) { Map keyStates = buildKeyStats(entityData, keysType, subKeys); - log.trace("[{}][{}][{}] Creating attributes subscription with keys: {}", serviceId, cmdId, subIdx, keyStates); + log.trace("[{}][{}][{}] Creating attributes subscription for [{}] with keys: {}", serviceId, cmdId, subIdx, entityData.getEntityId(), keyStates); return TbAttributeSubscription.builder() .serviceId(serviceId) .sessionId(sessionRef.getSessionId()) @@ -156,7 +158,7 @@ public class TbEntityDataSubCtx { keyStates.put(k, ts); }); } - log.trace("[{}][{}][{}] Creating time-series subscription with keys: {}", serviceId, cmdId, subIdx, keyStates); + log.trace("[{}][{}][{}] Creating time-series subscription for [{}] with keys: {}", serviceId, cmdId, subIdx, entityData.getEntityId(), keyStates); return TbTimeseriesSubscription.builder() .serviceId(serviceId) .sessionId(sessionRef.getSessionId()) @@ -304,7 +306,7 @@ public class TbEntityDataSubCtx { } } - public Collection update(PageData newData) { + public TbEntityDataSubCtxUpdateResult update(PageData newData) { Map oldDataMap; if (data != null && !data.getData().isEmpty()) { oldDataMap = data.getData().stream().collect(Collectors.toMap(EntityData::getEntityId, Function.identity())); @@ -314,20 +316,21 @@ public class TbEntityDataSubCtx { Map newDataMap = newData.getData().stream().collect(Collectors.toMap(EntityData::getEntityId, Function.identity())); if (oldDataMap.size() == newDataMap.size() && oldDataMap.keySet().equals(newDataMap.keySet())) { log.trace("[{}][{}] No updates to entity data found", sessionRef.getSessionId(), cmdId); - return Collections.emptyList(); + return TbEntityDataSubCtxUpdateResult.EMPTY; } else { this.data = newData; - List subIdsToRemove = new ArrayList<>(); + List subIdsToCancel = new ArrayList<>(); + List subsToAdd = new ArrayList<>(); Set currentSubs = new HashSet<>(); subToEntityIdMap.forEach((subId, entityId) -> { if (!newDataMap.containsKey(entityId)) { - subIdsToRemove.add(subId); + subIdsToCancel.add(subId); } else { currentSubs.add(entityId); } }); - log.trace("[{}][{}] Subscriptions that are invalid: {}", sessionRef.getSessionId(), cmdId, subIdsToRemove); - subIdsToRemove.forEach(subToEntityIdMap::remove); + log.trace("[{}][{}] Subscriptions that are invalid: {}", sessionRef.getSessionId(), cmdId, subIdsToCancel); + subIdsToCancel.forEach(subToEntityIdMap::remove); List newSubsList = newDataMap.entrySet().stream().filter(entry -> !currentSubs.contains(entry.getKey())).map(Map.Entry::getValue).collect(Collectors.toList()); if (!newSubsList.isEmpty()) { boolean resultToLatestValues; @@ -346,13 +349,13 @@ public class TbEntityDataSubCtx { newSubsList.forEach( entity -> { log.trace("[{}][{}] Found new subscription for entity: {}", sessionRef.getSessionId(), cmdId, entity.getEntityId()); - addSubscription(entity, keysByType, resultToLatestValues); + subsToAdd.addAll(addSubscriptions(entity, keysByType, resultToLatestValues)); } ); } } wsService.sendWsMsg(sessionRef.getSessionId(), new EntityDataUpdate(cmdId, data, null)); - return subIdsToRemove; + return new TbEntityDataSubCtxUpdateResult(subIdsToCancel, subsToAdd); } } @@ -360,4 +363,14 @@ public class TbEntityDataSubCtx { curTsCmd = cmd.getTsCmd(); latestValueCmd = cmd.getLatestCmd(); } + + @Data + @AllArgsConstructor + public static class TbEntityDataSubCtxUpdateResult { + + private static TbEntityDataSubCtxUpdateResult EMPTY = new TbEntityDataSubCtxUpdateResult(Collections.emptyList(), Collections.emptyList()); + + private List subsToCancel; + private List subsToAdd; + } }