简柏林 / thingskit · Commits

Commit 44f00eb011e9e2c07744fbd345ad7ed422e4e60d

Authored by Andrii Shvaika
1 parent f028e566

Improvements to WS API

Inline Side-by-side
Showing 14 changed files with 244 additions and 209 deletions
... ... @@ -19,17 +19,16 @@ import com.google.common.util.concurrent.FutureCallback;
19 19 import com.google.common.util.concurrent.Futures;
20 20 import com.google.common.util.concurrent.ListenableFuture;
21 21 import com.google.common.util.concurrent.MoreExecutors;
  22 +import lombok.Getter;
22 23 import lombok.extern.slf4j.Slf4j;
23 24 import org.checkerframework.checker.nullness.qual.Nullable;
24 25 import org.springframework.beans.factory.annotation.Autowired;
25 26 import org.springframework.beans.factory.annotation.Value;
26 27 import org.springframework.context.annotation.Lazy;
27   -import org.springframework.context.event.EventListener;
  28 +import org.springframework.scheduling.annotation.Scheduled;
28 29 import org.springframework.stereotype.Service;
29 30 import org.thingsboard.common.util.ThingsBoardThreadFactory;
30   -import org.thingsboard.server.common.data.EntityView;
31 31 import org.thingsboard.server.common.data.id.CustomerId;
32   -import org.thingsboard.server.common.data.id.EntityViewId;
33 32 import org.thingsboard.server.common.data.id.TenantId;
34 33 import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery;
35 34 import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
... ... @@ -40,19 +39,12 @@ import org.thingsboard.server.common.data.query.EntityDataQuery;
40 39 import org.thingsboard.server.common.data.query.EntityKey;
41 40 import org.thingsboard.server.common.data.query.EntityKeyType;
42 41 import org.thingsboard.server.common.data.query.TsValue;
43   -import org.thingsboard.server.common.msg.queue.ServiceType;
44   -import org.thingsboard.server.common.msg.queue.TbCallback;
45   -import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
46 42 import org.thingsboard.server.dao.entity.EntityService;
47 43 import org.thingsboard.server.dao.entityview.EntityViewService;
48 44 import org.thingsboard.server.dao.timeseries.TimeseriesService;
49   -import org.thingsboard.server.queue.discovery.ClusterTopologyChangeEvent;
50   -import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
51   -import org.thingsboard.server.queue.discovery.PartitionService;
52 45 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
53 46 import org.thingsboard.server.queue.util.TbCoreComponent;
54   -import org.thingsboard.server.service.queue.TbClusterService;
55   -import org.thingsboard.server.service.security.permission.Operation;
  47 +import org.thingsboard.server.service.executors.DbCallbackExecutorService;
56 48 import org.thingsboard.server.service.telemetry.DefaultTelemetryWebSocketService;
57 49 import org.thingsboard.server.service.telemetry.TelemetryWebSocketService;
58 50 import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef;
... ... @@ -63,12 +55,14 @@ import org.thingsboard.server.service.telemetry.cmd.v2.EntityHistoryCmd;
63 55 import org.thingsboard.server.service.telemetry.cmd.v2.LatestValueCmd;
64 56 import org.thingsboard.server.service.telemetry.cmd.v2.TimeSeriesCmd;
65 57 import org.thingsboard.server.service.telemetry.sub.SubscriptionErrorCode;
66   -import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;
67 58
68 59 import javax.annotation.PostConstruct;
69 60 import javax.annotation.PreDestroy;
70 61 import java.util.ArrayList;
  62 +import java.util.Arrays;
71 63 import java.util.Collection;
  64 +import java.util.Collections;
  65 +import java.util.Comparator;
72 66 import java.util.HashMap;
73 67 import java.util.LinkedHashMap;
74 68 import java.util.LinkedHashSet;
... ... @@ -79,6 +73,12 @@ import java.util.concurrent.ConcurrentHashMap;
79 73 import java.util.concurrent.ExecutionException;
80 74 import java.util.concurrent.ExecutorService;
81 75 import java.util.concurrent.Executors;
  76 +import java.util.concurrent.ScheduledExecutorService;
  77 +import java.util.concurrent.ScheduledFuture;
  78 +import java.util.concurrent.ThreadFactory;
  79 +import java.util.concurrent.TimeUnit;
  80 +import java.util.concurrent.atomic.AtomicInteger;
  81 +import java.util.concurrent.atomic.AtomicLong;
82 82 import java.util.stream.Collectors;
83 83
84 84 @Slf4j
... ... @@ -87,29 +87,15 @@ import java.util.stream.Collectors;
87 87 public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubscriptionService {
88 88
89 89 private static final int DEFAULT_LIMIT = 100;
90   - private final Set<TopicPartitionInfo> currentPartitions = ConcurrentHashMap.newKeySet();
91 90 private final Map<String, Map<Integer, TbEntityDataSubCtx>> subscriptionsBySessionId = new ConcurrentHashMap<>();
92 91
93 92 @Autowired
94 93 private TelemetryWebSocketService wsService;
95 94
96 95 @Autowired
97   - private EntityViewService entityViewService;
98   -
99   - @Autowired
100 96 private EntityService entityService;
101 97
102 98 @Autowired
103   - private PartitionService partitionService;
104   -
105   - @Autowired
106   - private TbClusterService clusterService;
107   -
108   - @Autowired
109   - @Lazy
110   - private SubscriptionManagerService subscriptionManagerService;
111   -
112   - @Autowired
113 99 @Lazy
114 100 private TbLocalSubscriptionService localSubscriptionService;
115 101
... ... @@ -119,18 +105,38 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
119 105 @Autowired
120 106 private TbServiceInfoProvider serviceInfoProvider;
121 107
  108 + @Autowired
  109 + @Getter
  110 + private DbCallbackExecutorService dbCallbackExecutor;
  111 +
  112 + private ScheduledExecutorService scheduler;
  113 +
122 114 @Value("${database.ts.type}")
123 115 private String databaseTsType;
  116 + @Value("${server.ws.dynamic_page_link_refresh_interval:6}")
  117 + private long dynamicPageLinkRefreshInterval;
  118 + @Value("${server.ws.dynamic_page_link_refresh_pool_size:1}")
  119 + private int dynamicPageLinkRefreshPoolSize;
124 120
125 121 private ExecutorService wsCallBackExecutor;
126 122 private boolean tsInSqlDB;
127 123 private String serviceId;
  124 + private AtomicInteger regularQueryInvocationCnt = new AtomicInteger();
  125 + private AtomicInteger dynamicQueryInvocationCnt = new AtomicInteger();
  126 + private AtomicLong regularQueryTimeSpent = new AtomicLong();
  127 + private AtomicLong dynamicQueryTimeSpent = new AtomicLong();
128 128
129 129 @PostConstruct
130 130 public void initExecutor() {
131 131 serviceId = serviceInfoProvider.getServiceId();
132 132 wsCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("ws-entity-sub-callback"));
133 133 tsInSqlDB = databaseTsType.equalsIgnoreCase("sql") || databaseTsType.equalsIgnoreCase("timescale");
  134 + ThreadFactory tbThreadFactory = ThingsBoardThreadFactory.forName("ws-entity-sub-scheduler");
  135 + if (dynamicPageLinkRefreshPoolSize == 1) {
  136 + scheduler = Executors.newSingleThreadScheduledExecutor(tbThreadFactory);
  137 + } else {
  138 + scheduler = Executors.newScheduledThreadPool(dynamicPageLinkRefreshPoolSize, tbThreadFactory);
  139 + }
134 140 }
135 141
136 142 @PreDestroy
... ... @@ -141,44 +147,18 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
141 147 }
142 148
143 149 @Override
144   - @EventListener(PartitionChangeEvent.class)
145   - public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
146   - if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) {
147   - currentPartitions.clear();
148   - currentPartitions.addAll(partitionChangeEvent.getPartitions());
149   - }
150   - }
151   -
152   - @Override
153   - @EventListener(ClusterTopologyChangeEvent.class)
154   - public void onApplicationEvent(ClusterTopologyChangeEvent event) {
155   - if (event.getServiceQueueKeys().stream().anyMatch(key -> ServiceType.TB_CORE.equals(key.getServiceType()))) {
156   - /*
157   - * If the cluster topology has changed, we need to push all current subscriptions to SubscriptionManagerService again.
158   - * Otherwise, the SubscriptionManagerService may "forget" those subscriptions in case of restart.
159   - * Although this is resource consuming operation, it is cheaper than sending ping/pong commands periodically
160   - * It is also cheaper then caching the subscriptions by entity id and then lookup of those caches every time we have new telemetry in SubscriptionManagerService.
161   - * Even if we cache locally the list of active subscriptions by entity id, it is still time consuming operation to get them from cache
162   - * Since number of subscriptions is usually much less then number of devices that are pushing data.
163   -// */
164   -// subscriptionsBySessionId.values().forEach(map -> map.values()
165   -// .forEach(sub -> pushSubscriptionToManagerService(sub, false)));
166   - }
167   - }
168   -
169   - @Override
170 150 public void handleCmd(TelemetryWebSocketSessionRef session, EntityDataCmd cmd) {
171 151 TbEntityDataSubCtx ctx = getSubCtx(session.getSessionId(), cmd.getCmdId());
172 152 if (ctx != null) {
173 153 log.debug("[{}][{}] Updating existing subscriptions using: {}", session.getSessionId(), cmd.getCmdId(), cmd);
174 154 if (cmd.getLatestCmd() != null || cmd.getTsCmd() != null || cmd.getHistoryCmd() != null) {
175   - Collection<Integer> oldSubIds = ctx.clearSubscriptions();
176   - oldSubIds.forEach(subId -> localSubscriptionService.cancelSubscription(serviceId, subId));
  155 + clearSubs(ctx);
177 156 }
178 157 } else {
179 158 log.debug("[{}][{}] Creating new subscription using: {}", session.getSessionId(), cmd.getCmdId(), cmd);
180 159 ctx = createSubCtx(session, cmd);
181 160 }
  161 + ctx.setCurrentCmd(cmd);
182 162 if (cmd.getQuery() != null) {
183 163 if (ctx.getQuery() == null) {
184 164 log.debug("[{}][{}] Initializing data using query: {}", session.getSessionId(), cmd.getCmdId(), cmd.getQuery());
... ... @@ -197,13 +177,26 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
197 177 }
198 178 });
199 179 }
  180 + long start = System.currentTimeMillis();
200 181 PageData<EntityData> data = entityService.findEntityDataByQuery(tenantId, customerId, ctx.getQuery());
  182 + long end = System.currentTimeMillis();
  183 + regularQueryInvocationCnt.incrementAndGet();
  184 + regularQueryTimeSpent.addAndGet(end - start);
  185 +
201 186 if (log.isTraceEnabled()) {
202 187 data.getData().forEach(ed -> {
203 188 log.trace("[{}][{}] EntityData: {}", session.getSessionId(), cmd.getCmdId(), ed);
204 189 });
205 190 }
206 191 ctx.setData(data);
  192 + ctx.cancelRefreshTask();
  193 + if (ctx.getQuery().getPageLink().isDynamic()) {
  194 + TbEntityDataSubCtx finalCtx = ctx;
  195 + ScheduledFuture<?> task = scheduler.scheduleWithFixedDelay(
  196 + () -> refreshDynamicQuery(tenantId, customerId, finalCtx),
  197 + dynamicPageLinkRefreshInterval, dynamicPageLinkRefreshInterval, TimeUnit.SECONDS);
  198 + finalCtx.setRefreshTask(task);
  199 + }
207 200 }
208 201 ListenableFuture<TbEntityDataSubCtx> historyFuture;
209 202 if (cmd.getHistoryCmd() != null) {
... ... @@ -233,6 +226,35 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
233 226 }, wsCallBackExecutor);
234 227 }
235 228
  229 + private void refreshDynamicQuery(TenantId tenantId, CustomerId customerId, TbEntityDataSubCtx finalCtx) {
  230 + try {
  231 + long start = System.currentTimeMillis();
  232 + Collection<Integer> oldSubIds = finalCtx.update(entityService.findEntityDataByQuery(tenantId, customerId, finalCtx.getQuery()));
  233 + long end = System.currentTimeMillis();
  234 + dynamicQueryInvocationCnt.incrementAndGet();
  235 + dynamicQueryTimeSpent.addAndGet(end - start);
  236 + oldSubIds.forEach(subId -> localSubscriptionService.cancelSubscription(serviceId, subId));
  237 + } catch (Exception e) {
  238 + log.warn("[{}][{}] Failed to refresh query", finalCtx.getSessionId(), finalCtx.getCmdId(), e);
  239 + }
  240 + }
  241 +
  242 + @Scheduled(fixedDelayString = "${server.ws.dynamic_page_link_stats:10000}")
  243 + public void printStats() {
  244 + int regularQueryInvocationCntValue = regularQueryInvocationCnt.getAndSet(0);
  245 + long regularQueryInvocationTimeValue = regularQueryTimeSpent.getAndSet(0);
  246 + int dynamicQueryInvocationCntValue = dynamicQueryInvocationCnt.getAndSet(0);
  247 + long dynamicQueryInvocationTimeValue = dynamicQueryTimeSpent.getAndSet(0);
  248 + long dynamicQueryCnt = subscriptionsBySessionId.values().stream().map(Map::values).count();
  249 + log.info("Stats: regularQueryInvocationCnt = [{}], regularQueryInvocationTime = [{}], dynamicQueryCnt = [{}] dynamicQueryInvocationCnt = [{}], dynamicQueryInvocationTime = [{}]",
  250 + regularQueryInvocationCntValue, regularQueryInvocationTimeValue, dynamicQueryCnt, dynamicQueryInvocationCntValue, dynamicQueryInvocationTimeValue);
  251 + }
  252 +
  253 + private void clearSubs(TbEntityDataSubCtx ctx) {
  254 + Collection<Integer> oldSubIds = ctx.clearSubscriptions();
  255 + oldSubIds.forEach(subId -> localSubscriptionService.cancelSubscription(serviceId, subId));
  256 + }
  257 +
236 258 private TbEntityDataSubCtx createSubCtx(TelemetryWebSocketSessionRef sessionRef, EntityDataCmd cmd) {
237 259 Map<Integer, TbEntityDataSubCtx> sessionSubs = subscriptionsBySessionId.computeIfAbsent(sessionRef.getSessionId(), k -> new HashMap<>());
238 260 TbEntityDataSubCtx ctx = new TbEntityDataSubCtx(serviceId, wsService, sessionRef, cmd.getCmdId());
... ... @@ -282,6 +304,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
282 304 }
283 305 wsService.sendWsMsg(ctx.getSessionId(), update);
284 306 createSubscriptions(ctx, keys.stream().map(key -> new EntityKey(EntityKeyType.TIME_SERIES, key)).collect(Collectors.toList()), false);
  307 + ctx.getData().getData().forEach(ed -> ed.getTimeseries().clear());
285 308 }
286 309
287 310 @Override
... ... @@ -378,12 +401,21 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
378 401 }
379 402
380 403 private ListenableFuture<TbEntityDataSubCtx> handleHistoryCmd(TbEntityDataSubCtx ctx, EntityHistoryCmd historyCmd) {
  404 + List<ReadTsKvQuery> finalTsKvQueryList;
381 405 List<ReadTsKvQuery> tsKvQueryList = historyCmd.getKeys().stream().map(key -> new BaseReadTsKvQuery(
382 406 key, historyCmd.getStartTs(), historyCmd.getEndTs(), historyCmd.getInterval(), getLimit(historyCmd.getLimit()), historyCmd.getAgg()
383 407 )).collect(Collectors.toList());
  408 + if (historyCmd.isFetchLatestPreviousPoint()) {
  409 + finalTsKvQueryList = new ArrayList<>(tsKvQueryList);
  410 + tsKvQueryList.addAll(historyCmd.getKeys().stream().map(key -> new BaseReadTsKvQuery(
  411 + key, historyCmd.getStartTs() - TimeUnit.DAYS.toMillis(365), historyCmd.getStartTs(), historyCmd.getInterval(), 1, historyCmd.getAgg()
  412 + )).collect(Collectors.toList()));
  413 + } else {
  414 + finalTsKvQueryList = tsKvQueryList;
  415 + }
384 416 Map<EntityData, ListenableFuture<List<TsKvEntry>>> fetchResultMap = new HashMap<>();
385 417 ctx.getData().getData().forEach(entityData -> fetchResultMap.put(entityData,
386   - tsService.findAll(ctx.getTenantId(), entityData.getEntityId(), tsKvQueryList)));
  418 + tsService.findAll(ctx.getTenantId(), entityData.getEntityId(), finalTsKvQueryList)));
387 419 return Futures.transform(Futures.allAsList(fetchResultMap.values()), f -> {
388 420 fetchResultMap.forEach((entityData, future) -> {
389 421 Map<String, List<TsValue>> keyData = new LinkedHashMap<>();
... ... @@ -394,6 +426,11 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
394 426 entityTsData.forEach(entry -> keyData.get(entry.getKey()).add(new TsValue(entry.getTs(), entry.getValueAsString())));
395 427 }
396 428 keyData.forEach((k, v) -> entityData.getTimeseries().put(k, v.toArray(new TsValue[v.size()])));
  429 + if (historyCmd.isFetchLatestPreviousPoint()) {
  430 + entityData.getTimeseries().values().forEach(dataArray -> {
  431 + Arrays.sort(dataArray, (o1, o2) -> Long.compare(o2.getTs(), o1.getTs()));
  432 + });
  433 + }
397 434 } catch (InterruptedException | ExecutionException e) {
398 435 log.warn("[{}][{}][{}] Failed to fetch historical data", ctx.getSessionId(), ctx.getCmdId(), entityData.getEntityId(), e);
399 436 wsService.sendWsMsg(ctx.getSessionId(),
... ... @@ -408,119 +445,29 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
408 445 update = new EntityDataUpdate(ctx.getCmdId(), null, ctx.getData().getData());
409 446 }
410 447 wsService.sendWsMsg(ctx.getSessionId(), update);
  448 + ctx.getData().getData().forEach(ed -> ed.getTimeseries().clear());
411 449 return ctx;
412 450 }, wsCallBackExecutor);
413 451 }
414 452
415 453 @Override
416 454 public void cancelSubscription(String sessionId, EntityDataUnsubscribeCmd cmd) {
417   - TbEntityDataSubCtx ctx = getSubCtx(sessionId, cmd.getCmdId());
  455 + cleanupAndCancel(getSubCtx(sessionId, cmd.getCmdId()));
418 456 }
419 457
420   -// //TODO 3.1: replace null callbacks with callbacks from websocket service.
421   -// @Override
422   -// public void addSubscription(TbSubscription subscription) {
423   -// EntityId entityId = subscription.getEntityId();
424   -// // Telemetry subscription on Entity Views are handled differently, because we need to allow only certain keys and time ranges;
425   -// if (entityId.getEntityType().equals(EntityType.ENTITY_VIEW) && TbSubscriptionType.TIMESERIES.equals(subscription.getType())) {
426   -// subscription = resolveEntityViewSubscription((TbTimeseriesSubscription) subscription);
427   -// }
428   -// pushSubscriptionToManagerService(subscription, true);
429   -// registerSubscription(subscription);
430   -// }
431   -
432   -// private void pushSubscriptionToManagerService(TbSubscription subscription, boolean pushToLocalService) {
433   -// TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, subscription.getTenantId(), subscription.getEntityId());
434   -// if (currentPartitions.contains(tpi)) {
435   -// // Subscription is managed on the same server;
436   -// if (pushToLocalService) {
437   -// subscriptionManagerService.addSubscription(subscription, TbCallback.EMPTY);
438   -// }
439   -// } else {
440   -// // Push to the queue;
441   -// TransportProtos.ToCoreMsg toCoreMsg = TbSubscriptionUtils.toNewSubscriptionProto(subscription);
442   -// clusterService.pushMsgToCore(tpi, subscription.getEntityId().getId(), toCoreMsg, null);
443   -// }
444   -// }
445   -
446   - @Override
447   - public void onSubscriptionUpdate(String sessionId, SubscriptionUpdate update, TbCallback callback) {
448   -// TbSubscription subscription = subscriptionsBySessionId
449   -// .getOrDefault(sessionId, Collections.emptyMap()).get(update.getSubscriptionId());
450   -// if (subscription != null) {
451   -// switch (subscription.getType()) {
452   -// case TIMESERIES:
453   -// TbTimeseriesSubscription tsSub = (TbTimeseriesSubscription) subscription;
454   -// update.getLatestValues().forEach((key, value) -> tsSub.getKeyStates().put(key, value));
455   -// break;
456   -// case ATTRIBUTES:
457   -// TbAttributeSubscription attrSub = (TbAttributeSubscription) subscription;
458   -// update.getLatestValues().forEach((key, value) -> attrSub.getKeyStates().put(key, value));
459   -// break;
460   -// }
461   -// wsService.sendWsMsg(sessionId, update);
462   -// }
463   -// callback.onSuccess();
  458 + private void cleanupAndCancel(TbEntityDataSubCtx ctx) {
  459 + if (ctx != null) {
  460 + ctx.cancelRefreshTask();
  461 + clearSubs(ctx);
  462 + }
464 463 }
465 464
466   -// @Override
467   -// public void cancelSubscription(String sessionId, int subscriptionId) {
468   -// log.debug("[{}][{}] Going to remove subscription.", sessionId, subscriptionId);
469   -// Map<Integer, TbSubscription> sessionSubscriptions = subscriptionsBySessionId.get(sessionId);
470   -// if (sessionSubscriptions != null) {
471   -// TbSubscription subscription = sessionSubscriptions.remove(subscriptionId);
472   -// if (subscription != null) {
473   -// if (sessionSubscriptions.isEmpty()) {
474   -// subscriptionsBySessionId.remove(sessionId);
475   -// }
476   -// TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, subscription.getTenantId(), subscription.getEntityId());
477   -// if (currentPartitions.contains(tpi)) {
478   -// // Subscription is managed on the same server;
479   -// subscriptionManagerService.cancelSubscription(sessionId, subscriptionId, TbCallback.EMPTY);
480   -// } else {
481   -// // Push to the queue;
482   -// TransportProtos.ToCoreMsg toCoreMsg = TbSubscriptionUtils.toCloseSubscriptionProto(subscription);
483   -// clusterService.pushMsgToCore(tpi, subscription.getEntityId().getId(), toCoreMsg, null);
484   -// }
485   -// } else {
486   -// log.debug("[{}][{}] Subscription not found!", sessionId, subscriptionId);
487   -// }
488   -// } else {
489   -// log.debug("[{}] No session subscriptions found!", sessionId);
490   -// }
491   -// }
492   -
493 465 @Override
494 466 public void cancelAllSessionSubscriptions(String sessionId) {
495   -// Map<Integer, TbSubscription> subscriptions = subscriptionsBySessionId.get(sessionId);
496   -// if (subscriptions != null) {
497   -// Set<Integer> toRemove = new HashSet<>(subscriptions.keySet());
498   -// toRemove.forEach(id -> cancelSubscription(sessionId, id));
499   -// }
500   - }
501   -
502   - private TbSubscription resolveEntityViewSubscription(TbTimeseriesSubscription subscription) {
503   - EntityView entityView = entityViewService.findEntityViewById(TenantId.SYS_TENANT_ID, new EntityViewId(subscription.getEntityId().getId()));
504   -
505   - Map<String, Long> keyStates;
506   - if (subscription.isAllKeys()) {
507   - keyStates = entityView.getKeys().getTimeseries().stream().collect(Collectors.toMap(k -> k, k -> 0L));
508   - } else {
509   - keyStates = subscription.getKeyStates().entrySet()
510   - .stream().filter(entry -> entityView.getKeys().getTimeseries().contains(entry.getKey()))
511   - .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
  467 + Map<Integer, TbEntityDataSubCtx> sessionSubs = subscriptionsBySessionId.remove(sessionId);
  468 + if (sessionSubs != null) {
  469 + sessionSubs.values().forEach(this::cleanupAndCancel);
512 470 }
513   -
514   - return TbTimeseriesSubscription.builder()
515   - .serviceId(subscription.getServiceId())
516   - .sessionId(subscription.getSessionId())
517   - .subscriptionId(subscription.getSubscriptionId())
518   - .tenantId(subscription.getTenantId())
519   - .entityId(entityView.getEntityId())
520   - .startTime(entityView.getStartTimeMs())
521   - .endTime(entityView.getEndTimeMs())
522   - .allKeys(false)
523   - .keyStates(keyStates).build();
524 471 }
525 472
526 473 private int getLimit(int limit) {
... ...
... ... @@ -28,6 +28,7 @@ import org.thingsboard.server.common.data.query.EntityKeyType;
28 28 import org.thingsboard.server.common.data.query.TsValue;
29 29 import org.thingsboard.server.service.telemetry.TelemetryWebSocketService;
30 30 import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef;
  31 +import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd;
31 32 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate;
32 33 import org.thingsboard.server.service.telemetry.cmd.v2.LatestValueCmd;
33 34 import org.thingsboard.server.service.telemetry.cmd.v2.TimeSeriesCmd;
... ... @@ -39,9 +40,14 @@ import java.util.Collection;
39 40 import java.util.Collections;
40 41 import java.util.Comparator;
41 42 import java.util.HashMap;
  43 +import java.util.HashSet;
42 44 import java.util.List;
43 45 import java.util.Map;
44 46 import java.util.Optional;
  47 +import java.util.Set;
  48 +import java.util.concurrent.ScheduledFuture;
  49 +import java.util.function.Function;
  50 +import java.util.stream.Collectors;
45 51
46 52 @Slf4j
47 53 @Data
... ... @@ -53,12 +59,14 @@ public class TbEntityDataSubCtx {
53 59 private final TelemetryWebSocketSessionRef sessionRef;
54 60 private final int cmdId;
55 61 private EntityDataQuery query;
56   - private LatestValueCmd latestCmd;
57 62 private TimeSeriesCmd tsCmd;
58 63 private PageData<EntityData> data;
59 64 private boolean initialDataSent;
60 65 private List<TbSubscription> tbSubs;
61 66 private Map<Integer, EntityId> subToEntityIdMap;
  67 + private volatile ScheduledFuture<?> refreshTask;
  68 + private TimeSeriesCmd curTsCmd;
  69 + private LatestValueCmd latestValueCmd;
62 70
63 71 public TbEntityDataSubCtx(String serviceId, TelemetryWebSocketService wsService, TelemetryWebSocketSessionRef sessionRef, int cmdId) {
64 72 this.serviceId = serviceId;
... ... @@ -86,34 +94,43 @@ public class TbEntityDataSubCtx {
86 94 public List<TbSubscription> createSubscriptions(List<EntityKey> keys, boolean resultToLatestValues) {
87 95 this.subToEntityIdMap = new HashMap<>();
88 96 tbSubs = new ArrayList<>();
89   - Map<EntityKeyType, List<EntityKey>> keysByType = new HashMap<>();
90   - keys.forEach(key -> keysByType.computeIfAbsent(key.getType(), k -> new ArrayList<>()).add(key));
  97 + Map<EntityKeyType, List<EntityKey>> keysByType = getEntityKeyByTypeMap(keys);
91 98 for (EntityData entityData : data.getData()) {
92   - keysByType.forEach((keysType, keysList) -> {
93   - int subIdx = sessionRef.getSessionSubIdSeq().incrementAndGet();
94   - subToEntityIdMap.put(subIdx, entityData.getEntityId());
95   - switch (keysType) {
96   - case TIME_SERIES:
97   - tbSubs.add(createTsSub(entityData, subIdx, keysList, resultToLatestValues));
98   - break;
99   - case CLIENT_ATTRIBUTE:
100   - tbSubs.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.CLIENT_SCOPE, keysList));
101   - break;
102   - case SHARED_ATTRIBUTE:
103   - tbSubs.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.SHARED_SCOPE, keysList));
104   - break;
105   - case SERVER_ATTRIBUTE:
106   - tbSubs.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.SERVER_SCOPE, keysList));
107   - break;
108   - case ATTRIBUTE:
109   - tbSubs.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.ANY_SCOPE, keysList));
110   - break;
111   - }
112   - });
  99 + addSubscription(entityData, keysByType, resultToLatestValues);
113 100 }
114 101 return tbSubs;
115 102 }
116 103
  104 + private Map<EntityKeyType, List<EntityKey>> getEntityKeyByTypeMap(List<EntityKey> keys) {
  105 + Map<EntityKeyType, List<EntityKey>> keysByType = new HashMap<>();
  106 + keys.forEach(key -> keysByType.computeIfAbsent(key.getType(), k -> new ArrayList<>()).add(key));
  107 + return keysByType;
  108 + }
  109 +
  110 + private void addSubscription(EntityData entityData, Map<EntityKeyType, List<EntityKey>> keysByType, boolean resultToLatestValues) {
  111 + keysByType.forEach((keysType, keysList) -> {
  112 + int subIdx = sessionRef.getSessionSubIdSeq().incrementAndGet();
  113 + subToEntityIdMap.put(subIdx, entityData.getEntityId());
  114 + switch (keysType) {
  115 + case TIME_SERIES:
  116 + tbSubs.add(createTsSub(entityData, subIdx, keysList, resultToLatestValues));
  117 + break;
  118 + case CLIENT_ATTRIBUTE:
  119 + tbSubs.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.CLIENT_SCOPE, keysList));
  120 + break;
  121 + case SHARED_ATTRIBUTE:
  122 + tbSubs.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.SHARED_SCOPE, keysList));
  123 + break;
  124 + case SERVER_ATTRIBUTE:
  125 + tbSubs.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.SERVER_SCOPE, keysList));
  126 + break;
  127 + case ATTRIBUTE:
  128 + tbSubs.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.ANY_SCOPE, keysList));
  129 + break;
  130 + }
  131 + });
  132 + }
  133 +
117 134 private TbSubscription createAttrSub(EntityData entityData, int subIdx, EntityKeyType keysType, TbAttributeSubscriptionScope scope, List<EntityKey> subKeys) {
118 135 Map<String, Long> keyStates = buildKeyStats(entityData, keysType, subKeys);
119 136 log.trace("[{}][{}][{}] Creating attributes subscription with keys: {}", serviceId, cmdId, subIdx, keyStates);
... ... @@ -275,4 +292,74 @@ public class TbEntityDataSubCtx {
275 292 return Collections.emptyList();
276 293 }
277 294 }
  295 +
  296 + public void setRefreshTask(ScheduledFuture<?> task) {
  297 + this.refreshTask = task;
  298 + }
  299 +
  300 + public void cancelRefreshTask() {
  301 + if (this.refreshTask != null) {
  302 + log.trace("[{}][{}] Canceling old refresh task", sessionRef.getSessionId(), cmdId);
  303 + this.refreshTask.cancel(true);
  304 + }
  305 + }
  306 +
  307 + public Collection<Integer> update(PageData<EntityData> newData) {
  308 + Map<EntityId, EntityData> oldDataMap;
  309 + if (data != null && !data.getData().isEmpty()) {
  310 + oldDataMap = data.getData().stream().collect(Collectors.toMap(EntityData::getEntityId, Function.identity()));
  311 + } else {
  312 + oldDataMap = Collections.emptyMap();
  313 + }
  314 + Map<EntityId, EntityData> newDataMap = newData.getData().stream().collect(Collectors.toMap(EntityData::getEntityId, Function.identity()));
  315 + if (oldDataMap.size() == newDataMap.size() && oldDataMap.keySet().equals(newDataMap.keySet())) {
  316 + log.trace("[{}][{}] No updates to entity data found", sessionRef.getSessionId(), cmdId);
  317 + return Collections.emptyList();
  318 + } else {
  319 + this.data = newData;
  320 + List<Integer> subIdsToRemove = new ArrayList<>();
  321 + Set<EntityId> currentSubs = new HashSet<>();
  322 + subToEntityIdMap.forEach((subId, entityId) -> {
  323 + if (!newDataMap.containsKey(entityId)) {
  324 + subIdsToRemove.add(subId);
  325 + } else {
  326 + currentSubs.add(entityId);
  327 + }
  328 + });
  329 + log.trace("[{}][{}] Subscriptions that are invalid: {}", sessionRef.getSessionId(), cmdId, subIdsToRemove);
  330 + subIdsToRemove.forEach(subToEntityIdMap::remove);
  331 + List<EntityData> newSubsList = newDataMap.entrySet().stream().filter(entry -> !currentSubs.contains(entry.getKey())).map(Map.Entry::getValue).collect(Collectors.toList());
  332 + if (!newSubsList.isEmpty()) {
  333 + boolean resultToLatestValues;
  334 + List<EntityKey> keys = null;
  335 + if (curTsCmd != null) {
  336 + resultToLatestValues = false;
  337 + keys = curTsCmd.getKeys().stream().map(key -> new EntityKey(EntityKeyType.TIME_SERIES, key)).collect(Collectors.toList());
  338 + } else if (latestValueCmd != null) {
  339 + resultToLatestValues = true;
  340 + keys = latestValueCmd.getKeys();
  341 + } else {
  342 + resultToLatestValues = true;
  343 + }
  344 + if (keys != null && !keys.isEmpty()) {
  345 + Map<EntityKeyType, List<EntityKey>> keysByType = getEntityKeyByTypeMap(keys);
  346 + newSubsList.forEach(
  347 + entity -> {
  348 + log.trace("[{}][{}] Found new subscription for entity: {}", sessionRef.getSessionId(), cmdId, entity.getEntityId());
  349 + if (curTsCmd != null) {
  350 + addSubscription(entity, keysByType, resultToLatestValues);
  351 + }
  352 + }
  353 + );
  354 + }
  355 + }
  356 + wsService.sendWsMsg(sessionRef.getSessionId(), new EntityDataUpdate(cmdId, data, null));
  357 + return subIdsToRemove;
  358 + }
  359 + }
  360 +
  361 + public void setCurrentCmd(EntityDataCmd cmd) {
  362 + curTsCmd = cmd.getTsCmd();
  363 + latestValueCmd = cmd.getLatestCmd();
  364 + }
278 365 }
... ...
... ... @@ -15,13 +15,9 @@
15 15 */
16 16 package org.thingsboard.server.service.subscription;
17 17
18   -import org.thingsboard.server.common.msg.queue.TbCallback;
19   -import org.thingsboard.server.queue.discovery.ClusterTopologyChangeEvent;
20   -import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
21 18 import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef;
22 19 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd;
23 20 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUnsubscribeCmd;
24   -import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;
25 21
26 22 public interface TbEntityDataSubscriptionService {
27 23
... ... @@ -31,9 +27,4 @@ public interface TbEntityDataSubscriptionService {
31 27
32 28 void cancelAllSessionSubscriptions(String sessionId);
33 29
34   - void onSubscriptionUpdate(String sessionId, SubscriptionUpdate update, TbCallback callback);
35   -
36   - void onApplicationEvent(PartitionChangeEvent event);
37   -
38   - void onApplicationEvent(ClusterTopologyChangeEvent event);
39 30 }
... ...
... ... @@ -697,15 +697,18 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
697 697 }
698 698
699 699 private void sendWsMsg(TelemetryWebSocketSessionRef sessionRef, int cmdId, Object update) {
700   - executor.submit(() -> {
701   - try {
702   - msgEndpoint.send(sessionRef, cmdId, jsonMapper.writeValueAsString(update));
703   - } catch (JsonProcessingException e) {
704   - log.warn("[{}] Failed to encode reply: {}", sessionRef.getSessionId(), update, e);
705   - } catch (IOException e) {
706   - log.warn("[{}] Failed to send reply: {}", sessionRef.getSessionId(), update, e);
707   - }
708   - });
  700 + try {
  701 + String msg = jsonMapper.writeValueAsString(update);
  702 + executor.submit(() -> {
  703 + try {
  704 + msgEndpoint.send(sessionRef, cmdId, msg);
  705 + } catch (IOException e) {
  706 + log.warn("[{}] Failed to send reply: {}", sessionRef.getSessionId(), update, e);
  707 + }
  708 + });
  709 + } catch (JsonProcessingException e) {
  710 + log.warn("[{}] Failed to encode reply: {}", sessionRef.getSessionId(), update, e);
  711 + }
709 712 }
710 713
711 714
... ...
... ... @@ -29,5 +29,6 @@ public class EntityHistoryCmd {
29 29 private long interval;
30 30 private int limit;
31 31 private Aggregation agg;
  32 + private boolean fetchLatestPreviousPoint;
32 33
33 34 }
... ...
... ... @@ -46,6 +46,8 @@ server:
46 46 max_subscriptions_per_regular_user: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SUBSCRIPTIONS_PER_REGULAR_USER:0}"
47 47 max_subscriptions_per_public_user: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SUBSCRIPTIONS_PER_PUBLIC_USER:0}"
48 48 max_updates_per_session: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_UPDATES_PER_SESSION:300:1,3000:60}"
  49 + dynamic_page_link_refresh_interval: "${TB_SERVER_WS_DYNAMIC_PAGE_LINK_REFRESH_INTERVAL_SEC:6}"
  50 + dynamic_page_link_refresh_pool_size: "${TB_SERVER_WS_DYNAMIC_PAGE_LINK_REFRESH_POOL_SIZE:1}"
49 51 rest:
50 52 limits:
51 53 tenant:
... ...
... ... @@ -23,27 +23,27 @@ public class BaseReadTsKvQuery extends BaseTsKvQuery implements ReadTsKvQuery {
23 23 private final long interval;
24 24 private final int limit;
25 25 private final Aggregation aggregation;
26   - private final String orderBy;
  26 + private final String order;
27 27
28 28 public BaseReadTsKvQuery(String key, long startTs, long endTs, long interval, int limit, Aggregation aggregation) {
29 29 this(key, startTs, endTs, interval, limit, aggregation, "DESC");
30 30 }
31 31
32 32 public BaseReadTsKvQuery(String key, long startTs, long endTs, long interval, int limit, Aggregation aggregation,
33   - String orderBy) {
  33 + String order) {
34 34 super(key, startTs, endTs);
35 35 this.interval = interval;
36 36 this.limit = limit;
37 37 this.aggregation = aggregation;
38   - this.orderBy = orderBy;
  38 + this.order = order;
39 39 }
40 40
41 41 public BaseReadTsKvQuery(String key, long startTs, long endTs) {
42 42 this(key, startTs, endTs, endTs - startTs, 1, Aggregation.AVG, "DESC");
43 43 }
44 44
45   - public BaseReadTsKvQuery(String key, long startTs, long endTs, int limit, String orderBy) {
46   - this(key, startTs, endTs, endTs - startTs, limit, Aggregation.NONE, orderBy);
  45 + public BaseReadTsKvQuery(String key, long startTs, long endTs, int limit, String order) {
  46 + this(key, startTs, endTs, endTs - startTs, limit, Aggregation.NONE, order);
47 47 }
48 48
49 49 }
... ...
... ... @@ -23,6 +23,6 @@ public interface ReadTsKvQuery extends TsKvQuery {
23 23
24 24 Aggregation getAggregation();
25 25
26   - String getOrderBy();
  26 + String getOrder();
27 27
28 28 }
... ...
... ... @@ -27,13 +27,17 @@ public class EntityDataPageLink {
27 27 private int page;
28 28 private String textSearch;
29 29 private EntityDataSortOrder sortOrder;
  30 + private boolean dynamic = false;
30 31
31 32 public EntityDataPageLink() {
32 33 }
33 34
  35 + public EntityDataPageLink(int pageSize, int page, String textSearch, EntityDataSortOrder sortOrder) {
  36 + this(pageSize, page, textSearch, sortOrder, false);
  37 + }
  38 +
34 39 @JsonIgnore
35 40 public EntityDataPageLink nextPageLink() {
36 41 return new EntityDataPageLink(this.pageSize, this.page+1, this.textSearch, this.sortOrder);
37 42 }
38   -
39 43 }
... ...
... ... @@ -152,7 +152,7 @@ public abstract class AbstractChunkedAggregationTimeseriesDao extends AbstractSq
152 152 query.getEndTs(),
153 153 PageRequest.of(0, query.getLimit(),
154 154 Sort.by(Sort.Direction.fromString(
155   - query.getOrderBy()), "ts")));
  155 + query.getOrder()), "ts")));
156 156 tsKvEntities.forEach(tsKvEntity -> tsKvEntity.setStrKey(query.getKey()));
157 157 return Futures.immediateFuture(DaoUtil.convertDataList(tsKvEntities));
158 158 }
... ...
... ... @@ -110,7 +110,7 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements
110 110 query.getEndTs(),
111 111 PageRequest.of(0, query.getLimit(),
112 112 Sort.by(Sort.Direction.fromString(
113   - query.getOrderBy()), "ts")));
  113 + query.getOrder()), "ts")));
114 114 timescaleTsKvEntities.forEach(tsKvEntity -> tsKvEntity.setStrKey(strKey));
115 115 return Futures.immediateFuture(DaoUtil.convertDataList(timescaleTsKvEntities));
116 116 }
... ...
... ... @@ -170,7 +170,7 @@ public class BaseTimeseriesService implements TimeseriesService {
170 170 } else {
171 171 endTs = query.getEndTs();
172 172 }
173   - return new BaseReadTsKvQuery(query.getKey(), startTs, endTs, query.getInterval(), query.getLimit(), query.getAggregation(), query.getOrderBy());
  173 + return new BaseReadTsKvQuery(query.getKey(), startTs, endTs, query.getInterval(), query.getLimit(), query.getAggregation(), query.getOrder());
174 174 }).collect(Collectors.toList());
175 175 }
176 176
... ...
... ... @@ -168,7 +168,7 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem
168 168 while (stepTs < query.getEndTs()) {
169 169 long startTs = stepTs;
170 170 long endTs = stepTs + step;
171   - ReadTsKvQuery subQuery = new BaseReadTsKvQuery(query.getKey(), startTs, endTs, step, 1, query.getAggregation(), query.getOrderBy());
  171 + ReadTsKvQuery subQuery = new BaseReadTsKvQuery(query.getKey(), startTs, endTs, step, 1, query.getAggregation(), query.getOrder());
172 172 futures.add(findAndAggregateAsync(tenantId, entityId, subQuery, toPartitionTs(startTs), toPartitionTs(endTs)));
173 173 stepTs = endTs;
174 174 }
... ...
... ... @@ -40,7 +40,7 @@ public class TsKvQueryCursor extends QueryCursor {
40 40
41 41 public TsKvQueryCursor(String entityType, UUID entityId, ReadTsKvQuery baseQuery, List<Long> partitions) {
42 42 super(entityType, entityId, baseQuery, partitions);
43   - this.orderBy = baseQuery.getOrderBy();
  43 + this.orderBy = baseQuery.getOrder();
44 44 this.partitionIndex = isDesc() ? partitions.size() - 1 : 0;
45 45 this.data = new ArrayList<>();
46 46 this.currentLimit = baseQuery.getLimit();
... ...