简柏林 / thingskit · Commits

Commit 0bfbe57479f1c818dfb58d491e57a568127cf369

Authored by Igor Kulikov
2 parents bf22c198 f97d74d6

Merge branch 'feature/entity-data-query' of github.com:thingsboard/thingsboard i… 

…nto feature/entity-data-query
Inline Side-by-side
Showing 18 changed files with 352 additions and 288 deletions
@@ -19,17 +19,16 @@ import com.google.common.util.concurrent.FutureCallback; @@ -19,17 +19,16 @@ import com.google.common.util.concurrent.FutureCallback;
19 import com.google.common.util.concurrent.Futures; 19 import com.google.common.util.concurrent.Futures;
20 import com.google.common.util.concurrent.ListenableFuture; 20 import com.google.common.util.concurrent.ListenableFuture;
21 import com.google.common.util.concurrent.MoreExecutors; 21 import com.google.common.util.concurrent.MoreExecutors;
  22 +import lombok.Getter;
22 import lombok.extern.slf4j.Slf4j; 23 import lombok.extern.slf4j.Slf4j;
23 import org.checkerframework.checker.nullness.qual.Nullable; 24 import org.checkerframework.checker.nullness.qual.Nullable;
24 import org.springframework.beans.factory.annotation.Autowired; 25 import org.springframework.beans.factory.annotation.Autowired;
25 import org.springframework.beans.factory.annotation.Value; 26 import org.springframework.beans.factory.annotation.Value;
26 import org.springframework.context.annotation.Lazy; 27 import org.springframework.context.annotation.Lazy;
27 -import org.springframework.context.event.EventListener; 28 +import org.springframework.scheduling.annotation.Scheduled;
28 import org.springframework.stereotype.Service; 29 import org.springframework.stereotype.Service;
29 import org.thingsboard.common.util.ThingsBoardThreadFactory; 30 import org.thingsboard.common.util.ThingsBoardThreadFactory;
30 -import org.thingsboard.server.common.data.EntityView;  
31 import org.thingsboard.server.common.data.id.CustomerId; 31 import org.thingsboard.server.common.data.id.CustomerId;
32 -import org.thingsboard.server.common.data.id.EntityViewId;  
33 import org.thingsboard.server.common.data.id.TenantId; 32 import org.thingsboard.server.common.data.id.TenantId;
34 import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery; 33 import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery;
35 import org.thingsboard.server.common.data.kv.ReadTsKvQuery; 34 import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
@@ -40,19 +39,12 @@ import org.thingsboard.server.common.data.query.EntityDataQuery; @@ -40,19 +39,12 @@ import org.thingsboard.server.common.data.query.EntityDataQuery;
40 import org.thingsboard.server.common.data.query.EntityKey; 39 import org.thingsboard.server.common.data.query.EntityKey;
41 import org.thingsboard.server.common.data.query.EntityKeyType; 40 import org.thingsboard.server.common.data.query.EntityKeyType;
42 import org.thingsboard.server.common.data.query.TsValue; 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 import org.thingsboard.server.dao.entity.EntityService; 42 import org.thingsboard.server.dao.entity.EntityService;
47 import org.thingsboard.server.dao.entityview.EntityViewService; 43 import org.thingsboard.server.dao.entityview.EntityViewService;
48 import org.thingsboard.server.dao.timeseries.TimeseriesService; 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 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; 45 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
53 import org.thingsboard.server.queue.util.TbCoreComponent; 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 import org.thingsboard.server.service.telemetry.DefaultTelemetryWebSocketService; 48 import org.thingsboard.server.service.telemetry.DefaultTelemetryWebSocketService;
57 import org.thingsboard.server.service.telemetry.TelemetryWebSocketService; 49 import org.thingsboard.server.service.telemetry.TelemetryWebSocketService;
58 import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef; 50 import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef;
@@ -60,15 +52,18 @@ import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd; @@ -60,15 +52,18 @@ import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd;
60 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUnsubscribeCmd; 52 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUnsubscribeCmd;
61 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate; 53 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate;
62 import org.thingsboard.server.service.telemetry.cmd.v2.EntityHistoryCmd; 54 import org.thingsboard.server.service.telemetry.cmd.v2.EntityHistoryCmd;
  55 +import org.thingsboard.server.service.telemetry.cmd.v2.GetTsCmd;
63 import org.thingsboard.server.service.telemetry.cmd.v2.LatestValueCmd; 56 import org.thingsboard.server.service.telemetry.cmd.v2.LatestValueCmd;
64 import org.thingsboard.server.service.telemetry.cmd.v2.TimeSeriesCmd; 57 import org.thingsboard.server.service.telemetry.cmd.v2.TimeSeriesCmd;
65 import org.thingsboard.server.service.telemetry.sub.SubscriptionErrorCode; 58 import org.thingsboard.server.service.telemetry.sub.SubscriptionErrorCode;
66 -import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;  
67 59
68 import javax.annotation.PostConstruct; 60 import javax.annotation.PostConstruct;
69 import javax.annotation.PreDestroy; 61 import javax.annotation.PreDestroy;
70 import java.util.ArrayList; 62 import java.util.ArrayList;
  63 +import java.util.Arrays;
71 import java.util.Collection; 64 import java.util.Collection;
  65 +import java.util.Collections;
  66 +import java.util.Comparator;
72 import java.util.HashMap; 67 import java.util.HashMap;
73 import java.util.LinkedHashMap; 68 import java.util.LinkedHashMap;
74 import java.util.LinkedHashSet; 69 import java.util.LinkedHashSet;
@@ -79,6 +74,12 @@ import java.util.concurrent.ConcurrentHashMap; @@ -79,6 +74,12 @@ import java.util.concurrent.ConcurrentHashMap;
79 import java.util.concurrent.ExecutionException; 74 import java.util.concurrent.ExecutionException;
80 import java.util.concurrent.ExecutorService; 75 import java.util.concurrent.ExecutorService;
81 import java.util.concurrent.Executors; 76 import java.util.concurrent.Executors;
  77 +import java.util.concurrent.ScheduledExecutorService;
  78 +import java.util.concurrent.ScheduledFuture;
  79 +import java.util.concurrent.ThreadFactory;
  80 +import java.util.concurrent.TimeUnit;
  81 +import java.util.concurrent.atomic.AtomicInteger;
  82 +import java.util.concurrent.atomic.AtomicLong;
82 import java.util.stream.Collectors; 83 import java.util.stream.Collectors;
83 84
84 @Slf4j 85 @Slf4j
@@ -87,29 +88,15 @@ import java.util.stream.Collectors; @@ -87,29 +88,15 @@ import java.util.stream.Collectors;
87 public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubscriptionService { 88 public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubscriptionService {
88 89
89 private static final int DEFAULT_LIMIT = 100; 90 private static final int DEFAULT_LIMIT = 100;
90 - private final Set<TopicPartitionInfo> currentPartitions = ConcurrentHashMap.newKeySet();  
91 private final Map<String, Map<Integer, TbEntityDataSubCtx>> subscriptionsBySessionId = new ConcurrentHashMap<>(); 91 private final Map<String, Map<Integer, TbEntityDataSubCtx>> subscriptionsBySessionId = new ConcurrentHashMap<>();
92 92
93 @Autowired 93 @Autowired
94 private TelemetryWebSocketService wsService; 94 private TelemetryWebSocketService wsService;
95 95
96 @Autowired 96 @Autowired
97 - private EntityViewService entityViewService;  
98 -  
99 - @Autowired  
100 private EntityService entityService; 97 private EntityService entityService;
101 98
102 @Autowired 99 @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 @Lazy 100 @Lazy
114 private TbLocalSubscriptionService localSubscriptionService; 101 private TbLocalSubscriptionService localSubscriptionService;
115 102
@@ -119,18 +106,38 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc @@ -119,18 +106,38 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
119 @Autowired 106 @Autowired
120 private TbServiceInfoProvider serviceInfoProvider; 107 private TbServiceInfoProvider serviceInfoProvider;
121 108
  109 + @Autowired
  110 + @Getter
  111 + private DbCallbackExecutorService dbCallbackExecutor;
  112 +
  113 + private ScheduledExecutorService scheduler;
  114 +
122 @Value("${database.ts.type}") 115 @Value("${database.ts.type}")
123 private String databaseTsType; 116 private String databaseTsType;
  117 + @Value("${server.ws.dynamic_page_link_refresh_interval:6}")
  118 + private long dynamicPageLinkRefreshInterval;
  119 + @Value("${server.ws.dynamic_page_link_refresh_pool_size:1}")
  120 + private int dynamicPageLinkRefreshPoolSize;
124 121
125 private ExecutorService wsCallBackExecutor; 122 private ExecutorService wsCallBackExecutor;
126 private boolean tsInSqlDB; 123 private boolean tsInSqlDB;
127 private String serviceId; 124 private String serviceId;
  125 + private AtomicInteger regularQueryInvocationCnt = new AtomicInteger();
  126 + private AtomicInteger dynamicQueryInvocationCnt = new AtomicInteger();
  127 + private AtomicLong regularQueryTimeSpent = new AtomicLong();
  128 + private AtomicLong dynamicQueryTimeSpent = new AtomicLong();
128 129
129 @PostConstruct 130 @PostConstruct
130 public void initExecutor() { 131 public void initExecutor() {
131 serviceId = serviceInfoProvider.getServiceId(); 132 serviceId = serviceInfoProvider.getServiceId();
132 wsCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("ws-entity-sub-callback")); 133 wsCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("ws-entity-sub-callback"));
133 tsInSqlDB = databaseTsType.equalsIgnoreCase("sql") || databaseTsType.equalsIgnoreCase("timescale"); 134 tsInSqlDB = databaseTsType.equalsIgnoreCase("sql") || databaseTsType.equalsIgnoreCase("timescale");
  135 + ThreadFactory tbThreadFactory = ThingsBoardThreadFactory.forName("ws-entity-sub-scheduler");
  136 + if (dynamicPageLinkRefreshPoolSize == 1) {
  137 + scheduler = Executors.newSingleThreadScheduledExecutor(tbThreadFactory);
  138 + } else {
  139 + scheduler = Executors.newScheduledThreadPool(dynamicPageLinkRefreshPoolSize, tbThreadFactory);
  140 + }
134 } 141 }
135 142
136 @PreDestroy 143 @PreDestroy
@@ -141,44 +148,18 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc @@ -141,44 +148,18 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
141 } 148 }
142 149
143 @Override 150 @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 public void handleCmd(TelemetryWebSocketSessionRef session, EntityDataCmd cmd) { 151 public void handleCmd(TelemetryWebSocketSessionRef session, EntityDataCmd cmd) {
171 TbEntityDataSubCtx ctx = getSubCtx(session.getSessionId(), cmd.getCmdId()); 152 TbEntityDataSubCtx ctx = getSubCtx(session.getSessionId(), cmd.getCmdId());
172 if (ctx != null) { 153 if (ctx != null) {
173 log.debug("[{}][{}] Updating existing subscriptions using: {}", session.getSessionId(), cmd.getCmdId(), cmd); 154 log.debug("[{}][{}] Updating existing subscriptions using: {}", session.getSessionId(), cmd.getCmdId(), cmd);
174 if (cmd.getLatestCmd() != null || cmd.getTsCmd() != null || cmd.getHistoryCmd() != null) { 155 if (cmd.getLatestCmd() != null || cmd.getTsCmd() != null || cmd.getHistoryCmd() != null) {
175 - Collection<Integer> oldSubIds = ctx.clearSubscriptions();  
176 - oldSubIds.forEach(subId -> localSubscriptionService.cancelSubscription(serviceId, subId)); 156 + clearSubs(ctx);
177 } 157 }
178 } else { 158 } else {
179 log.debug("[{}][{}] Creating new subscription using: {}", session.getSessionId(), cmd.getCmdId(), cmd); 159 log.debug("[{}][{}] Creating new subscription using: {}", session.getSessionId(), cmd.getCmdId(), cmd);
180 ctx = createSubCtx(session, cmd); 160 ctx = createSubCtx(session, cmd);
181 } 161 }
  162 + ctx.setCurrentCmd(cmd);
182 if (cmd.getQuery() != null) { 163 if (cmd.getQuery() != null) {
183 if (ctx.getQuery() == null) { 164 if (ctx.getQuery() == null) {
184 log.debug("[{}][{}] Initializing data using query: {}", session.getSessionId(), cmd.getCmdId(), cmd.getQuery()); 165 log.debug("[{}][{}] Initializing data using query: {}", session.getSessionId(), cmd.getCmdId(), cmd.getQuery());
@@ -197,13 +178,26 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc @@ -197,13 +178,26 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
197 } 178 }
198 }); 179 });
199 } 180 }
  181 + long start = System.currentTimeMillis();
200 PageData<EntityData> data = entityService.findEntityDataByQuery(tenantId, customerId, ctx.getQuery()); 182 PageData<EntityData> data = entityService.findEntityDataByQuery(tenantId, customerId, ctx.getQuery());
  183 + long end = System.currentTimeMillis();
  184 + regularQueryInvocationCnt.incrementAndGet();
  185 + regularQueryTimeSpent.addAndGet(end - start);
  186 +
201 if (log.isTraceEnabled()) { 187 if (log.isTraceEnabled()) {
202 data.getData().forEach(ed -> { 188 data.getData().forEach(ed -> {
203 log.trace("[{}][{}] EntityData: {}", session.getSessionId(), cmd.getCmdId(), ed); 189 log.trace("[{}][{}] EntityData: {}", session.getSessionId(), cmd.getCmdId(), ed);
204 }); 190 });
205 } 191 }
206 ctx.setData(data); 192 ctx.setData(data);
  193 + ctx.cancelRefreshTask();
  194 + if (ctx.getQuery().getPageLink().isDynamic()) {
  195 + TbEntityDataSubCtx finalCtx = ctx;
  196 + ScheduledFuture<?> task = scheduler.scheduleWithFixedDelay(
  197 + () -> refreshDynamicQuery(tenantId, customerId, finalCtx),
  198 + dynamicPageLinkRefreshInterval, dynamicPageLinkRefreshInterval, TimeUnit.SECONDS);
  199 + finalCtx.setRefreshTask(task);
  200 + }
207 } 201 }
208 ListenableFuture<TbEntityDataSubCtx> historyFuture; 202 ListenableFuture<TbEntityDataSubCtx> historyFuture;
209 if (cmd.getHistoryCmd() != null) { 203 if (cmd.getHistoryCmd() != null) {
@@ -233,6 +227,35 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc @@ -233,6 +227,35 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
233 }, wsCallBackExecutor); 227 }, wsCallBackExecutor);
234 } 228 }
235 229
  230 + private void refreshDynamicQuery(TenantId tenantId, CustomerId customerId, TbEntityDataSubCtx finalCtx) {
  231 + try {
  232 + long start = System.currentTimeMillis();
  233 + Collection<Integer> oldSubIds = finalCtx.update(entityService.findEntityDataByQuery(tenantId, customerId, finalCtx.getQuery()));
  234 + long end = System.currentTimeMillis();
  235 + dynamicQueryInvocationCnt.incrementAndGet();
  236 + dynamicQueryTimeSpent.addAndGet(end - start);
  237 + oldSubIds.forEach(subId -> localSubscriptionService.cancelSubscription(serviceId, subId));
  238 + } catch (Exception e) {
  239 + log.warn("[{}][{}] Failed to refresh query", finalCtx.getSessionId(), finalCtx.getCmdId(), e);
  240 + }
  241 + }
  242 +
  243 + @Scheduled(fixedDelayString = "${server.ws.dynamic_page_link_stats:10000}")
  244 + public void printStats() {
  245 + int regularQueryInvocationCntValue = regularQueryInvocationCnt.getAndSet(0);
  246 + long regularQueryInvocationTimeValue = regularQueryTimeSpent.getAndSet(0);
  247 + int dynamicQueryInvocationCntValue = dynamicQueryInvocationCnt.getAndSet(0);
  248 + long dynamicQueryInvocationTimeValue = dynamicQueryTimeSpent.getAndSet(0);
  249 + long dynamicQueryCnt = subscriptionsBySessionId.values().stream().map(Map::values).count();
  250 + log.info("Stats: regularQueryInvocationCnt = [{}], regularQueryInvocationTime = [{}], dynamicQueryCnt = [{}] dynamicQueryInvocationCnt = [{}], dynamicQueryInvocationTime = [{}]",
  251 + regularQueryInvocationCntValue, regularQueryInvocationTimeValue, dynamicQueryCnt, dynamicQueryInvocationCntValue, dynamicQueryInvocationTimeValue);
  252 + }
  253 +
  254 + private void clearSubs(TbEntityDataSubCtx ctx) {
  255 + Collection<Integer> oldSubIds = ctx.clearSubscriptions();
  256 + oldSubIds.forEach(subId -> localSubscriptionService.cancelSubscription(serviceId, subId));
  257 + }
  258 +
236 private TbEntityDataSubCtx createSubCtx(TelemetryWebSocketSessionRef sessionRef, EntityDataCmd cmd) { 259 private TbEntityDataSubCtx createSubCtx(TelemetryWebSocketSessionRef sessionRef, EntityDataCmd cmd) {
237 Map<Integer, TbEntityDataSubCtx> sessionSubs = subscriptionsBySessionId.computeIfAbsent(sessionRef.getSessionId(), k -> new HashMap<>()); 260 Map<Integer, TbEntityDataSubCtx> sessionSubs = subscriptionsBySessionId.computeIfAbsent(sessionRef.getSessionId(), k -> new HashMap<>());
238 TbEntityDataSubCtx ctx = new TbEntityDataSubCtx(serviceId, wsService, sessionRef, cmd.getCmdId()); 261 TbEntityDataSubCtx ctx = new TbEntityDataSubCtx(serviceId, wsService, sessionRef, cmd.getCmdId());
@@ -250,49 +273,86 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc @@ -250,49 +273,86 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
250 } 273 }
251 } 274 }
252 275
253 - private void handleTimeSeriesCmd(TbEntityDataSubCtx ctx, TimeSeriesCmd cmd) {  
254 - List<String> keys = cmd.getKeys(); 276 + private ListenableFuture<TbEntityDataSubCtx> handleTimeSeriesCmd(TbEntityDataSubCtx ctx, TimeSeriesCmd cmd) {
255 log.debug("[{}][{}] Fetching time-series data for last {} ms for keys: ({})", ctx.getSessionId(), ctx.getCmdId(), cmd.getTimeWindow(), cmd.getKeys()); 277 log.debug("[{}][{}] Fetching time-series data for last {} ms for keys: ({})", ctx.getSessionId(), ctx.getCmdId(), cmd.getTimeWindow(), cmd.getKeys());
256 - long startTs = cmd.getStartTs();  
257 - long endTs = cmd.getStartTs() + cmd.getTimeWindow();  
258 -  
259 - Map<EntityData, ListenableFuture<Map<String, List<TsValue>>>> tsFutures = new HashMap<>();  
260 - for (EntityData entityData : ctx.getData().getData()) {  
261 - List<ReadTsKvQuery> queries = keys.stream().map(key -> new BaseReadTsKvQuery(key, startTs, endTs, cmd.getInterval(),  
262 - getLimit(cmd.getLimit()), DefaultTelemetryWebSocketService.getAggregation(cmd.getAgg()))).collect(Collectors.toList());  
263 - ListenableFuture<List<TsKvEntry>> tsDataFutures = tsService.findAll(ctx.getTenantId(), entityData.getEntityId(), queries);  
264 - tsFutures.put(entityData, Futures.transform(tsDataFutures, this::toTsValues, MoreExecutors.directExecutor())); 278 + return handleGetTsCmd(ctx, cmd, true);
  279 + }
  280 +
  281 +
  282 + private ListenableFuture<TbEntityDataSubCtx> handleHistoryCmd(TbEntityDataSubCtx ctx, EntityHistoryCmd cmd) {
  283 + log.debug("[{}][{}] Fetching history data for start {} and end {} ms for keys: ({})", ctx.getSessionId(), ctx.getCmdId(), cmd.getStartTs(), cmd.getEndTs(), cmd.getKeys());
  284 + return handleGetTsCmd(ctx, cmd, false);
  285 + }
  286 +
  287 + private ListenableFuture<TbEntityDataSubCtx> handleGetTsCmd(TbEntityDataSubCtx ctx, GetTsCmd cmd, boolean subscribe) {
  288 + List<String> keys = cmd.getKeys();
  289 + List<ReadTsKvQuery> finalTsKvQueryList;
  290 + List<ReadTsKvQuery> tsKvQueryList = cmd.getKeys().stream().map(key -> new BaseReadTsKvQuery(
  291 + key, cmd.getStartTs(), cmd.getEndTs(), cmd.getInterval(), getLimit(cmd.getLimit()), cmd.getAgg()
  292 + )).collect(Collectors.toList());
  293 + if (cmd.isFetchLatestPreviousPoint()) {
  294 + finalTsKvQueryList = new ArrayList<>(tsKvQueryList);
  295 + tsKvQueryList.addAll(cmd.getKeys().stream().map(key -> new BaseReadTsKvQuery(
  296 + key, cmd.getStartTs() - TimeUnit.DAYS.toMillis(365), cmd.getStartTs(), cmd.getInterval(), 1, cmd.getAgg()
  297 + )).collect(Collectors.toList()));
  298 + } else {
  299 + finalTsKvQueryList = tsKvQueryList;
265 } 300 }
266 - Futures.addCallback(Futures.allAsList(tsFutures.values()), new FutureCallback<List<Map<String, List<TsValue>>>>() {  
267 - @Override  
268 - public void onSuccess(@Nullable List<Map<String, List<TsValue>>> result) {  
269 - tsFutures.forEach((key, value) -> {  
270 - try {  
271 - value.get().forEach((k, v) -> key.getTimeseries().put(k, v.toArray(new TsValue[v.size()])));  
272 - } catch (InterruptedException | ExecutionException e) {  
273 - log.warn("[{}][{}] Failed to lookup time-series data: {}:{}", ctx.getSessionId(), ctx.getCmdId(), key.getEntityId(), keys, e); 301 + Map<EntityData, ListenableFuture<List<TsKvEntry>>> fetchResultMap = new HashMap<>();
  302 + ctx.getData().getData().forEach(entityData -> fetchResultMap.put(entityData,
  303 + tsService.findAll(ctx.getTenantId(), entityData.getEntityId(), finalTsKvQueryList)));
  304 + return Futures.transform(Futures.allAsList(fetchResultMap.values()), f -> {
  305 + fetchResultMap.forEach((entityData, future) -> {
  306 + Map<String, List<TsValue>> keyData = new LinkedHashMap<>();
  307 + cmd.getKeys().forEach(key -> keyData.put(key, new ArrayList<>()));
  308 + try {
  309 + List<TsKvEntry> entityTsData = future.get();
  310 + if (entityTsData != null) {
  311 + entityTsData.forEach(entry -> keyData.get(entry.getKey()).add(new TsValue(entry.getTs(), entry.getValueAsString())));
274 } 312 }
275 - });  
276 - EntityDataUpdate update;  
277 - if (!ctx.isInitialDataSent()) {  
278 - update = new EntityDataUpdate(ctx.getCmdId(), ctx.getData(), null);  
279 - ctx.setInitialDataSent(true);  
280 - } else {  
281 - update = new EntityDataUpdate(ctx.getCmdId(), null, ctx.getData().getData()); 313 + keyData.forEach((k, v) -> entityData.getTimeseries().put(k, v.toArray(new TsValue[v.size()])));
  314 + if (cmd.isFetchLatestPreviousPoint()) {
  315 + entityData.getTimeseries().values().forEach(dataArray -> {
  316 + Arrays.sort(dataArray, (o1, o2) -> Long.compare(o2.getTs(), o1.getTs()));
  317 + });
  318 + }
  319 + } catch (InterruptedException | ExecutionException e) {
  320 + log.warn("[{}][{}][{}] Failed to fetch historical data", ctx.getSessionId(), ctx.getCmdId(), entityData.getEntityId(), e);
  321 + wsService.sendWsMsg(ctx.getSessionId(),
  322 + new EntityDataUpdate(ctx.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR.getCode(), "Failed to fetch historical data!"));
282 } 323 }
283 - wsService.sendWsMsg(ctx.getSessionId(), update);  
284 - createSubscriptions(ctx, keys.stream().map(key -> new EntityKey(EntityKeyType.TIME_SERIES, key)).collect(Collectors.toList()), false); 324 + });
  325 + EntityDataUpdate update;
  326 + if (!ctx.isInitialDataSent()) {
  327 + update = new EntityDataUpdate(ctx.getCmdId(), ctx.getData(), null);
  328 + ctx.setInitialDataSent(true);
  329 + } else {
  330 + update = new EntityDataUpdate(ctx.getCmdId(), null, ctx.getData().getData());
285 } 331 }
286 -  
287 - @Override  
288 - public void onFailure(Throwable t) {  
289 - log.warn("[{}][{}] Failed to process websocket command: {}:{}", ctx.getSessionId(), ctx.getCmdId(), ctx.getQuery(), cmd, t);  
290 - wsService.sendWsMsg(ctx.getSessionId(),  
291 - new EntityDataUpdate(ctx.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR.getCode(), "Failed to process websocket command!")); 332 + wsService.sendWsMsg(ctx.getSessionId(), update);
  333 + if (subscribe) {
  334 + createSubscriptions(ctx, keys.stream().map(key -> new EntityKey(EntityKeyType.TIME_SERIES, key)).collect(Collectors.toList()), false);
292 } 335 }
  336 + ctx.getData().getData().forEach(ed -> ed.getTimeseries().clear());
  337 + return ctx;
293 }, wsCallBackExecutor); 338 }, wsCallBackExecutor);
294 } 339 }
295 340
  341 + private List<ReadTsKvQuery> getReadTsKvQueries(GetTsCmd cmd) {
  342 + List<ReadTsKvQuery> finalTsKvQueryList;
  343 + List<ReadTsKvQuery> queries = cmd.getKeys().stream().map(key -> new BaseReadTsKvQuery(key, cmd.getStartTs(), cmd.getEndTs(), cmd.getInterval(),
  344 + getLimit(cmd.getLimit()), cmd.getAgg())).collect(Collectors.toList());
  345 + if (cmd.isFetchLatestPreviousPoint()) {
  346 + finalTsKvQueryList = new ArrayList<>(queries);
  347 + finalTsKvQueryList.addAll(cmd.getKeys().stream().map(key -> new BaseReadTsKvQuery(
  348 + key, cmd.getStartTs() - TimeUnit.DAYS.toMillis(365), cmd.getStartTs(), cmd.getInterval(), 1, cmd.getAgg()
  349 + )).collect(Collectors.toList()));
  350 + } else {
  351 + finalTsKvQueryList = queries;
  352 + }
  353 + return finalTsKvQueryList;
  354 + }
  355 +
296 private void handleLatestCmd(TbEntityDataSubCtx ctx, LatestValueCmd latestCmd) { 356 private void handleLatestCmd(TbEntityDataSubCtx ctx, LatestValueCmd latestCmd) {
297 log.trace("[{}][{}] Going to process latest command: {}", ctx.getSessionId(), ctx.getCmdId(), latestCmd); 357 log.trace("[{}][{}] Going to process latest command: {}", ctx.getSessionId(), ctx.getCmdId(), latestCmd);
298 //Fetch the latest values for telemetry keys (in case they are not copied from NoSQL to SQL DB in hybrid mode. 358 //Fetch the latest values for telemetry keys (in case they are not copied from NoSQL to SQL DB in hybrid mode.
@@ -377,150 +437,24 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc @@ -377,150 +437,24 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
377 return results; 437 return results;
378 } 438 }
379 439
380 - private ListenableFuture<TbEntityDataSubCtx> handleHistoryCmd(TbEntityDataSubCtx ctx, EntityHistoryCmd historyCmd) {  
381 - List<ReadTsKvQuery> tsKvQueryList = historyCmd.getKeys().stream().map(key -> new BaseReadTsKvQuery(  
382 - key, historyCmd.getStartTs(), historyCmd.getEndTs(), historyCmd.getInterval(), getLimit(historyCmd.getLimit()), historyCmd.getAgg()  
383 - )).collect(Collectors.toList());  
384 - Map<EntityData, ListenableFuture<List<TsKvEntry>>> fetchResultMap = new HashMap<>();  
385 - ctx.getData().getData().forEach(entityData -> fetchResultMap.put(entityData,  
386 - tsService.findAll(ctx.getTenantId(), entityData.getEntityId(), tsKvQueryList)));  
387 - return Futures.transform(Futures.allAsList(fetchResultMap.values()), f -> {  
388 - fetchResultMap.forEach((entityData, future) -> {  
389 - Map<String, List<TsValue>> keyData = new LinkedHashMap<>();  
390 - historyCmd.getKeys().forEach(key -> keyData.put(key, new ArrayList<>()));  
391 - try {  
392 - List<TsKvEntry> entityTsData = future.get();  
393 - if (entityTsData != null) {  
394 - entityTsData.forEach(entry -> keyData.get(entry.getKey()).add(new TsValue(entry.getTs(), entry.getValueAsString())));  
395 - }  
396 - keyData.forEach((k, v) -> entityData.getTimeseries().put(k, v.toArray(new TsValue[v.size()])));  
397 - } catch (InterruptedException | ExecutionException e) {  
398 - log.warn("[{}][{}][{}] Failed to fetch historical data", ctx.getSessionId(), ctx.getCmdId(), entityData.getEntityId(), e);  
399 - wsService.sendWsMsg(ctx.getSessionId(),  
400 - new EntityDataUpdate(ctx.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR.getCode(), "Failed to fetch historical data!"));  
401 - }  
402 - });  
403 - EntityDataUpdate update;  
404 - if (!ctx.isInitialDataSent()) {  
405 - update = new EntityDataUpdate(ctx.getCmdId(), ctx.getData(), null);  
406 - ctx.setInitialDataSent(true);  
407 - } else {  
408 - update = new EntityDataUpdate(ctx.getCmdId(), null, ctx.getData().getData());  
409 - }  
410 - wsService.sendWsMsg(ctx.getSessionId(), update);  
411 - return ctx;  
412 - }, wsCallBackExecutor);  
413 - }  
414 -  
415 @Override 440 @Override
416 public void cancelSubscription(String sessionId, EntityDataUnsubscribeCmd cmd) { 441 public void cancelSubscription(String sessionId, EntityDataUnsubscribeCmd cmd) {
417 - TbEntityDataSubCtx ctx = getSubCtx(sessionId, cmd.getCmdId()); 442 + cleanupAndCancel(getSubCtx(sessionId, cmd.getCmdId()));
418 } 443 }
419 444
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(); 445 + private void cleanupAndCancel(TbEntityDataSubCtx ctx) {
  446 + if (ctx != null) {
  447 + ctx.cancelRefreshTask();
  448 + clearSubs(ctx);
  449 + }
464 } 450 }
465 451
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 @Override 452 @Override
494 public void cancelAllSessionSubscriptions(String sessionId) { 453 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)); 454 + Map<Integer, TbEntityDataSubCtx> sessionSubs = subscriptionsBySessionId.remove(sessionId);
  455 + if (sessionSubs != null) {
  456 + sessionSubs.values().forEach(this::cleanupAndCancel);
512 } 457 }
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 } 458 }
525 459
526 private int getLimit(int limit) { 460 private int getLimit(int limit) {
@@ -28,6 +28,7 @@ import org.thingsboard.server.common.data.query.EntityKeyType; @@ -28,6 +28,7 @@ import org.thingsboard.server.common.data.query.EntityKeyType;
28 import org.thingsboard.server.common.data.query.TsValue; 28 import org.thingsboard.server.common.data.query.TsValue;
29 import org.thingsboard.server.service.telemetry.TelemetryWebSocketService; 29 import org.thingsboard.server.service.telemetry.TelemetryWebSocketService;
30 import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef; 30 import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef;
  31 +import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd;
31 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate; 32 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate;
32 import org.thingsboard.server.service.telemetry.cmd.v2.LatestValueCmd; 33 import org.thingsboard.server.service.telemetry.cmd.v2.LatestValueCmd;
33 import org.thingsboard.server.service.telemetry.cmd.v2.TimeSeriesCmd; 34 import org.thingsboard.server.service.telemetry.cmd.v2.TimeSeriesCmd;
@@ -39,9 +40,14 @@ import java.util.Collection; @@ -39,9 +40,14 @@ import java.util.Collection;
39 import java.util.Collections; 40 import java.util.Collections;
40 import java.util.Comparator; 41 import java.util.Comparator;
41 import java.util.HashMap; 42 import java.util.HashMap;
  43 +import java.util.HashSet;
42 import java.util.List; 44 import java.util.List;
43 import java.util.Map; 45 import java.util.Map;
44 import java.util.Optional; 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 @Slf4j 52 @Slf4j
47 @Data 53 @Data
@@ -53,12 +59,14 @@ public class TbEntityDataSubCtx { @@ -53,12 +59,14 @@ public class TbEntityDataSubCtx {
53 private final TelemetryWebSocketSessionRef sessionRef; 59 private final TelemetryWebSocketSessionRef sessionRef;
54 private final int cmdId; 60 private final int cmdId;
55 private EntityDataQuery query; 61 private EntityDataQuery query;
56 - private LatestValueCmd latestCmd;  
57 private TimeSeriesCmd tsCmd; 62 private TimeSeriesCmd tsCmd;
58 private PageData<EntityData> data; 63 private PageData<EntityData> data;
59 private boolean initialDataSent; 64 private boolean initialDataSent;
60 private List<TbSubscription> tbSubs; 65 private List<TbSubscription> tbSubs;
61 private Map<Integer, EntityId> subToEntityIdMap; 66 private Map<Integer, EntityId> subToEntityIdMap;
  67 + private volatile ScheduledFuture<?> refreshTask;
  68 + private TimeSeriesCmd curTsCmd;
  69 + private LatestValueCmd latestValueCmd;
62 70
63 public TbEntityDataSubCtx(String serviceId, TelemetryWebSocketService wsService, TelemetryWebSocketSessionRef sessionRef, int cmdId) { 71 public TbEntityDataSubCtx(String serviceId, TelemetryWebSocketService wsService, TelemetryWebSocketSessionRef sessionRef, int cmdId) {
64 this.serviceId = serviceId; 72 this.serviceId = serviceId;
@@ -86,34 +94,43 @@ public class TbEntityDataSubCtx { @@ -86,34 +94,43 @@ public class TbEntityDataSubCtx {
86 public List<TbSubscription> createSubscriptions(List<EntityKey> keys, boolean resultToLatestValues) { 94 public List<TbSubscription> createSubscriptions(List<EntityKey> keys, boolean resultToLatestValues) {
87 this.subToEntityIdMap = new HashMap<>(); 95 this.subToEntityIdMap = new HashMap<>();
88 tbSubs = new ArrayList<>(); 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 for (EntityData entityData : data.getData()) { 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 return tbSubs; 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 private TbSubscription createAttrSub(EntityData entityData, int subIdx, EntityKeyType keysType, TbAttributeSubscriptionScope scope, List<EntityKey> subKeys) { 134 private TbSubscription createAttrSub(EntityData entityData, int subIdx, EntityKeyType keysType, TbAttributeSubscriptionScope scope, List<EntityKey> subKeys) {
118 Map<String, Long> keyStates = buildKeyStats(entityData, keysType, subKeys); 135 Map<String, Long> keyStates = buildKeyStats(entityData, keysType, subKeys);
119 log.trace("[{}][{}][{}] Creating attributes subscription with keys: {}", serviceId, cmdId, subIdx, keyStates); 136 log.trace("[{}][{}][{}] Creating attributes subscription with keys: {}", serviceId, cmdId, subIdx, keyStates);
@@ -275,4 +292,74 @@ public class TbEntityDataSubCtx { @@ -275,4 +292,74 @@ public class TbEntityDataSubCtx {
275 return Collections.emptyList(); 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,13 +15,9 @@
15 */ 15 */
16 package org.thingsboard.server.service.subscription; 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 import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef; 18 import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef;
22 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd; 19 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd;
23 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUnsubscribeCmd; 20 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUnsubscribeCmd;
24 -import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;  
25 21
26 public interface TbEntityDataSubscriptionService { 22 public interface TbEntityDataSubscriptionService {
27 23
@@ -31,9 +27,4 @@ public interface TbEntityDataSubscriptionService { @@ -31,9 +27,4 @@ public interface TbEntityDataSubscriptionService {
31 27
32 void cancelAllSessionSubscriptions(String sessionId); 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,15 +697,18 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
697 } 697 }
698 698
699 private void sendWsMsg(TelemetryWebSocketSessionRef sessionRef, int cmdId, Object update) { 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
@@ -21,7 +21,7 @@ import org.thingsboard.server.common.data.kv.Aggregation; @@ -21,7 +21,7 @@ import org.thingsboard.server.common.data.kv.Aggregation;
21 import java.util.List; 21 import java.util.List;
22 22
23 @Data 23 @Data
24 -public class EntityHistoryCmd { 24 +public class EntityHistoryCmd implements GetTsCmd {
25 25
26 private List<String> keys; 26 private List<String> keys;
27 private long startTs; 27 private long startTs;
@@ -29,5 +29,6 @@ public class EntityHistoryCmd { @@ -29,5 +29,6 @@ public class EntityHistoryCmd {
29 private long interval; 29 private long interval;
30 private int limit; 30 private int limit;
31 private Aggregation agg; 31 private Aggregation agg;
  32 + private boolean fetchLatestPreviousPoint;
32 33
33 } 34 }
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.service.telemetry.cmd.v2;
  17 +
  18 +import org.thingsboard.server.common.data.kv.Aggregation;
  19 +
  20 +import java.util.List;
  21 +
  22 +public interface GetTsCmd {
  23 +
  24 + long getStartTs();
  25 +
  26 + long getEndTs();
  27 +
  28 + List<String> getKeys();
  29 +
  30 + long getInterval();
  31 +
  32 + int getLimit();
  33 +
  34 + Aggregation getAgg();
  35 +
  36 + boolean isFetchLatestPreviousPoint();
  37 +
  38 +}
@@ -15,18 +15,26 @@ @@ -15,18 +15,26 @@
15 */ 15 */
16 package org.thingsboard.server.service.telemetry.cmd.v2; 16 package org.thingsboard.server.service.telemetry.cmd.v2;
17 17
  18 +import com.fasterxml.jackson.annotation.JsonIgnore;
18 import lombok.Data; 19 import lombok.Data;
  20 +import org.thingsboard.server.common.data.kv.Aggregation;
19 21
20 import java.util.List; 22 import java.util.List;
21 23
22 @Data 24 @Data
23 -public class TimeSeriesCmd { 25 +public class TimeSeriesCmd implements GetTsCmd {
24 26
25 private List<String> keys; 27 private List<String> keys;
26 private long startTs; 28 private long startTs;
27 private long timeWindow; 29 private long timeWindow;
28 private long interval; 30 private long interval;
29 private int limit; 31 private int limit;
30 - private String agg; 32 + private Aggregation agg;
  33 + private boolean fetchLatestPreviousPoint;
31 34
  35 + @JsonIgnore
  36 + @Override
  37 + public long getEndTs() {
  38 + return startTs + timeWindow;
  39 + }
32 } 40 }
@@ -46,6 +46,8 @@ server: @@ -46,6 +46,8 @@ server:
46 max_subscriptions_per_regular_user: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SUBSCRIPTIONS_PER_REGULAR_USER:0}" 46 max_subscriptions_per_regular_user: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SUBSCRIPTIONS_PER_REGULAR_USER:0}"
47 max_subscriptions_per_public_user: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SUBSCRIPTIONS_PER_PUBLIC_USER:0}" 47 max_subscriptions_per_public_user: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SUBSCRIPTIONS_PER_PUBLIC_USER:0}"
48 max_updates_per_session: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_UPDATES_PER_SESSION:300:1,3000:60}" 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 rest: 51 rest:
50 limits: 52 limits:
51 tenant: 53 tenant:
@@ -194,7 +194,7 @@ public class BaseWebsocketApiTest extends AbstractWebsocketTest { @@ -194,7 +194,7 @@ public class BaseWebsocketApiTest extends AbstractWebsocketTest {
194 194
195 TimeSeriesCmd tsCmd = new TimeSeriesCmd(); 195 TimeSeriesCmd tsCmd = new TimeSeriesCmd();
196 tsCmd.setKeys(Arrays.asList("temperature")); 196 tsCmd.setKeys(Arrays.asList("temperature"));
197 - tsCmd.setAgg(Aggregation.NONE.name()); 197 + tsCmd.setAgg(Aggregation.NONE);
198 tsCmd.setLimit(1000); 198 tsCmd.setLimit(1000);
199 tsCmd.setStartTs(now - TimeUnit.HOURS.toMillis(1)); 199 tsCmd.setStartTs(now - TimeUnit.HOURS.toMillis(1));
200 tsCmd.setTimeWindow(TimeUnit.HOURS.toMillis(1)); 200 tsCmd.setTimeWindow(TimeUnit.HOURS.toMillis(1));
@@ -561,16 +561,12 @@ public class BaseWebsocketApiTest extends AbstractWebsocketTest { @@ -561,16 +561,12 @@ public class BaseWebsocketApiTest extends AbstractWebsocketTest {
561 wsClient.registerWaitForUpdate(); 561 wsClient.registerWaitForUpdate();
562 AttributeKvEntry dataPoint1 = new BaseAttributeKvEntry(now - TimeUnit.MINUTES.toMillis(1), new LongDataEntry("serverAttributeKey", 42L)); 562 AttributeKvEntry dataPoint1 = new BaseAttributeKvEntry(now - TimeUnit.MINUTES.toMillis(1), new LongDataEntry("serverAttributeKey", 42L));
563 List<AttributeKvEntry> tsData = Arrays.asList(dataPoint1); 563 List<AttributeKvEntry> tsData = Arrays.asList(dataPoint1);
564 - sendAttributes(device, TbAttributeSubscriptionScope.SERVER_SCOPE, tsData);  
565 -  
566 Thread.sleep(100); 564 Thread.sleep(100);
567 565
568 - cmd = new EntityDataCmd(1, edq, null, latestCmd, null);  
569 - wrapper = new TelemetryPluginCmdsWrapper();  
570 - wrapper.setEntityDataCmds(Collections.singletonList(cmd)); 566 + sendAttributes(device, TbAttributeSubscriptionScope.SERVER_SCOPE, tsData);
571 567
572 msg = wsClient.waitForUpdate(); 568 msg = wsClient.waitForUpdate();
573 - 569 + Assert.assertNotNull(msg);
574 update = mapper.readValue(msg, EntityDataUpdate.class); 570 update = mapper.readValue(msg, EntityDataUpdate.class);
575 Assert.assertEquals(1, update.getCmdId()); 571 Assert.assertEquals(1, update.getCmdId());
576 List<EntityData> eData = update.getUpdate(); 572 List<EntityData> eData = update.getUpdate();
@@ -23,27 +23,27 @@ public class BaseReadTsKvQuery extends BaseTsKvQuery implements ReadTsKvQuery { @@ -23,27 +23,27 @@ public class BaseReadTsKvQuery extends BaseTsKvQuery implements ReadTsKvQuery {
23 private final long interval; 23 private final long interval;
24 private final int limit; 24 private final int limit;
25 private final Aggregation aggregation; 25 private final Aggregation aggregation;
26 - private final String orderBy; 26 + private final String order;
27 27
28 public BaseReadTsKvQuery(String key, long startTs, long endTs, long interval, int limit, Aggregation aggregation) { 28 public BaseReadTsKvQuery(String key, long startTs, long endTs, long interval, int limit, Aggregation aggregation) {
29 this(key, startTs, endTs, interval, limit, aggregation, "DESC"); 29 this(key, startTs, endTs, interval, limit, aggregation, "DESC");
30 } 30 }
31 31
32 public BaseReadTsKvQuery(String key, long startTs, long endTs, long interval, int limit, Aggregation aggregation, 32 public BaseReadTsKvQuery(String key, long startTs, long endTs, long interval, int limit, Aggregation aggregation,
33 - String orderBy) { 33 + String order) {
34 super(key, startTs, endTs); 34 super(key, startTs, endTs);
35 this.interval = interval; 35 this.interval = interval;
36 this.limit = limit; 36 this.limit = limit;
37 this.aggregation = aggregation; 37 this.aggregation = aggregation;
38 - this.orderBy = orderBy; 38 + this.order = order;
39 } 39 }
40 40
41 public BaseReadTsKvQuery(String key, long startTs, long endTs) { 41 public BaseReadTsKvQuery(String key, long startTs, long endTs) {
42 this(key, startTs, endTs, endTs - startTs, 1, Aggregation.AVG, "DESC"); 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,6 +23,6 @@ public interface ReadTsKvQuery extends TsKvQuery {
23 23
24 Aggregation getAggregation(); 24 Aggregation getAggregation();
25 25
26 - String getOrderBy(); 26 + String getOrder();
27 27
28 } 28 }
@@ -27,13 +27,17 @@ public class EntityDataPageLink { @@ -27,13 +27,17 @@ public class EntityDataPageLink {
27 private int page; 27 private int page;
28 private String textSearch; 28 private String textSearch;
29 private EntityDataSortOrder sortOrder; 29 private EntityDataSortOrder sortOrder;
  30 + private boolean dynamic = false;
30 31
31 public EntityDataPageLink() { 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 @JsonIgnore 39 @JsonIgnore
35 public EntityDataPageLink nextPageLink() { 40 public EntityDataPageLink nextPageLink() {
36 return new EntityDataPageLink(this.pageSize, this.page+1, this.textSearch, this.sortOrder); 41 return new EntityDataPageLink(this.pageSize, this.page+1, this.textSearch, this.sortOrder);
37 } 42 }
38 -  
39 } 43 }
@@ -451,7 +451,7 @@ public class DefaultEntityQueryRepository implements EntityQueryRepository { @@ -451,7 +451,7 @@ public class DefaultEntityQueryRepository implements EntityQueryRepository {
451 List<String> searchPredicates = selectionMapping.stream().map(mapping -> { 451 List<String> searchPredicates = selectionMapping.stream().map(mapping -> {
452 String paramName = mapping.getValueAlias() + "_lowerSearchText"; 452 String paramName = mapping.getValueAlias() + "_lowerSearchText";
453 ctx.addStringParameter(paramName, lowerSearchText); 453 ctx.addStringParameter(paramName, lowerSearchText);
454 - return String.format("LOWER(%s) LIKE :%s", mapping.getValueAlias(), paramName); 454 + return String.format("LOWER(%s) LIKE concat('%%', :%s, '%%')", mapping.getValueAlias(), paramName);
455 } 455 }
456 ).collect(Collectors.toList()); 456 ).collect(Collectors.toList());
457 return String.format(" WHERE %s", String.join(" or ", searchPredicates)); 457 return String.format(" WHERE %s", String.join(" or ", searchPredicates));
@@ -152,7 +152,7 @@ public abstract class AbstractChunkedAggregationTimeseriesDao extends AbstractSq @@ -152,7 +152,7 @@ public abstract class AbstractChunkedAggregationTimeseriesDao extends AbstractSq
152 query.getEndTs(), 152 query.getEndTs(),
153 PageRequest.of(0, query.getLimit(), 153 PageRequest.of(0, query.getLimit(),
154 Sort.by(Sort.Direction.fromString( 154 Sort.by(Sort.Direction.fromString(
155 - query.getOrderBy()), "ts"))); 155 + query.getOrder()), "ts")));
156 tsKvEntities.forEach(tsKvEntity -> tsKvEntity.setStrKey(query.getKey())); 156 tsKvEntities.forEach(tsKvEntity -> tsKvEntity.setStrKey(query.getKey()));
157 return Futures.immediateFuture(DaoUtil.convertDataList(tsKvEntities)); 157 return Futures.immediateFuture(DaoUtil.convertDataList(tsKvEntities));
158 } 158 }
@@ -110,7 +110,7 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements @@ -110,7 +110,7 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements
110 query.getEndTs(), 110 query.getEndTs(),
111 PageRequest.of(0, query.getLimit(), 111 PageRequest.of(0, query.getLimit(),
112 Sort.by(Sort.Direction.fromString( 112 Sort.by(Sort.Direction.fromString(
113 - query.getOrderBy()), "ts"))); 113 + query.getOrder()), "ts")));
114 timescaleTsKvEntities.forEach(tsKvEntity -> tsKvEntity.setStrKey(strKey)); 114 timescaleTsKvEntities.forEach(tsKvEntity -> tsKvEntity.setStrKey(strKey));
115 return Futures.immediateFuture(DaoUtil.convertDataList(timescaleTsKvEntities)); 115 return Futures.immediateFuture(DaoUtil.convertDataList(timescaleTsKvEntities));
116 } 116 }
@@ -170,7 +170,7 @@ public class BaseTimeseriesService implements TimeseriesService { @@ -170,7 +170,7 @@ public class BaseTimeseriesService implements TimeseriesService {
170 } else { 170 } else {
171 endTs = query.getEndTs(); 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 }).collect(Collectors.toList()); 174 }).collect(Collectors.toList());
175 } 175 }
176 176
@@ -168,7 +168,7 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem @@ -168,7 +168,7 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem
168 while (stepTs < query.getEndTs()) { 168 while (stepTs < query.getEndTs()) {
169 long startTs = stepTs; 169 long startTs = stepTs;
170 long endTs = stepTs + step; 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 futures.add(findAndAggregateAsync(tenantId, entityId, subQuery, toPartitionTs(startTs), toPartitionTs(endTs))); 172 futures.add(findAndAggregateAsync(tenantId, entityId, subQuery, toPartitionTs(startTs), toPartitionTs(endTs)));
173 stepTs = endTs; 173 stepTs = endTs;
174 } 174 }
@@ -40,7 +40,7 @@ public class TsKvQueryCursor extends QueryCursor { @@ -40,7 +40,7 @@ public class TsKvQueryCursor extends QueryCursor {
40 40
41 public TsKvQueryCursor(String entityType, UUID entityId, ReadTsKvQuery baseQuery, List<Long> partitions) { 41 public TsKvQueryCursor(String entityType, UUID entityId, ReadTsKvQuery baseQuery, List<Long> partitions) {
42 super(entityType, entityId, baseQuery, partitions); 42 super(entityType, entityId, baseQuery, partitions);
43 - this.orderBy = baseQuery.getOrderBy(); 43 + this.orderBy = baseQuery.getOrder();
44 this.partitionIndex = isDesc() ? partitions.size() - 1 : 0; 44 this.partitionIndex = isDesc() ? partitions.size() - 1 : 0;
45 this.data = new ArrayList<>(); 45 this.data = new ArrayList<>();
46 this.currentLimit = baseQuery.getLimit(); 46 this.currentLimit = baseQuery.getLimit();