简柏林 / thingskit · Commits

Commit 1e9016cb478c6494344903e6de45f33bb646d32a

Authored by Andrii Shvaika
1 parent 3176c208

Implementation

Inline Side-by-side
Showing 14 changed files with 652 additions and 300 deletions
... ... @@ -54,6 +54,7 @@ import org.thingsboard.server.service.telemetry.TelemetryWebSocketService;
54 54 import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef;
55 55 import org.thingsboard.server.service.telemetry.cmd.v2.AlarmDataCmd;
56 56 import org.thingsboard.server.service.telemetry.cmd.v2.AlarmDataUpdate;
  57 +import org.thingsboard.server.service.telemetry.cmd.v2.EntityCountCmd;
57 58 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd;
58 59 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate;
59 60 import org.thingsboard.server.service.telemetry.cmd.v2.EntityHistoryCmd;
... ... @@ -92,7 +93,7 @@ import java.util.stream.Collectors;
92 93 public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubscriptionService {
93 94
94 95 private static final int DEFAULT_LIMIT = 100;
95   - private final Map<String, Map<Integer, TbAbstractDataSubCtx>> subscriptionsBySessionId = new ConcurrentHashMap<>();
  96 + private final Map<String, Map<Integer, TbAbstractSubCtx>> subscriptionsBySessionId = new ConcurrentHashMap<>();
96 97
97 98 @Autowired
98 99 private TelemetryWebSocketService wsService;
... ... @@ -202,7 +203,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
202 203 //TODO: validate number of dynamic page links against rate limits. Ignore dynamic flag if limit is reached.
203 204 TbEntityDataSubCtx finalCtx = ctx;
204 205 ScheduledFuture<?> task = scheduler.scheduleWithFixedDelay(
205   - () -> refreshDynamicQuery(tenantId, customerId, finalCtx),
  206 + () -> refreshDynamicQuery(finalCtx),
206 207 dynamicPageLinkRefreshInterval, dynamicPageLinkRefreshInterval, TimeUnit.SECONDS);
207 208 finalCtx.setRefreshTask(task);
208 209 }
... ... @@ -236,6 +237,26 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
236 237 }
237 238
238 239 @Override
  240 + public void handleCmd(TelemetryWebSocketSessionRef session, EntityCountCmd cmd) {
  241 + TbEntityCountSubCtx ctx = getSubCtx(session.getSessionId(), cmd.getCmdId());
  242 + if (ctx == null) {
  243 + ctx = createSubCtx(session, cmd);
  244 + long start = System.currentTimeMillis();
  245 + ctx.fetchData();
  246 + long end = System.currentTimeMillis();
  247 + stats.getRegularQueryInvocationCnt().incrementAndGet();
  248 + stats.getRegularQueryTimeSpent().addAndGet(end - start);
  249 + TbEntityCountSubCtx finalCtx = ctx;
  250 + ScheduledFuture<?> task = scheduler.scheduleWithFixedDelay(
  251 + () -> refreshDynamicQuery(finalCtx),
  252 + dynamicPageLinkRefreshInterval, dynamicPageLinkRefreshInterval, TimeUnit.SECONDS);
  253 + finalCtx.setRefreshTask(task);
  254 + } else {
  255 + log.debug("[{}][{}] Received duplicate command: {}", session.getSessionId(), cmd.getCmdId(), cmd);
  256 + }
  257 + }
  258 +
  259 + @Override
239 260 public void handleCmd(TelemetryWebSocketSessionRef session, AlarmDataCmd cmd) {
240 261 TbAlarmDataSubCtx ctx = getSubCtx(session.getSessionId(), cmd.getCmdId());
241 262 if (ctx == null) {
... ... @@ -267,7 +288,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
267 288 }
268 289 }
269 290
270   - private void refreshDynamicQuery(TenantId tenantId, CustomerId customerId, TbEntityDataSubCtx finalCtx) {
  291 + private void refreshDynamicQuery(TbAbstractSubCtx finalCtx) {
271 292 try {
272 293 long start = System.currentTimeMillis();
273 294 finalCtx.update();
... ... @@ -299,7 +320,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
299 320 }
300 321
301 322 private TbEntityDataSubCtx createSubCtx(TelemetryWebSocketSessionRef sessionRef, EntityDataCmd cmd) {
302   - Map<Integer, TbAbstractDataSubCtx> sessionSubs = subscriptionsBySessionId.computeIfAbsent(sessionRef.getSessionId(), k -> new HashMap<>());
  323 + Map<Integer, TbAbstractSubCtx> sessionSubs = subscriptionsBySessionId.computeIfAbsent(sessionRef.getSessionId(), k -> new HashMap<>());
303 324 TbEntityDataSubCtx ctx = new TbEntityDataSubCtx(serviceId, wsService, entityService, localSubscriptionService,
304 325 attributesService, stats, sessionRef, cmd.getCmdId(), maxEntitiesPerDataSubscription);
305 326 if (cmd.getQuery() != null) {
... ... @@ -309,8 +330,20 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
309 330 return ctx;
310 331 }
311 332
  333 + private TbEntityCountSubCtx createSubCtx(TelemetryWebSocketSessionRef sessionRef, EntityCountCmd cmd) {
  334 + Map<Integer, TbAbstractSubCtx> sessionSubs = subscriptionsBySessionId.computeIfAbsent(sessionRef.getSessionId(), k -> new HashMap<>());
  335 + TbEntityCountSubCtx ctx = new TbEntityCountSubCtx(serviceId, wsService, entityService, localSubscriptionService,
  336 + attributesService, stats, sessionRef, cmd.getCmdId());
  337 + if (cmd.getQuery() != null) {
  338 + ctx.setAndResolveQuery(cmd.getQuery());
  339 + }
  340 + sessionSubs.put(cmd.getCmdId(), ctx);
  341 + return ctx;
  342 + }
  343 +
  344 +
312 345 private TbAlarmDataSubCtx createSubCtx(TelemetryWebSocketSessionRef sessionRef, AlarmDataCmd cmd) {
313   - Map<Integer, TbAbstractDataSubCtx> sessionSubs = subscriptionsBySessionId.computeIfAbsent(sessionRef.getSessionId(), k -> new HashMap<>());
  346 + Map<Integer, TbAbstractSubCtx> sessionSubs = subscriptionsBySessionId.computeIfAbsent(sessionRef.getSessionId(), k -> new HashMap<>());
314 347 TbAlarmDataSubCtx ctx = new TbAlarmDataSubCtx(serviceId, wsService, entityService, localSubscriptionService,
315 348 attributesService, stats, alarmService, sessionRef, cmd.getCmdId(), maxEntitiesPerAlarmSubscription);
316 349 ctx.setAndResolveQuery(cmd.getQuery());
... ... @@ -319,8 +352,8 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
319 352 }
320 353
321 354 @SuppressWarnings("unchecked")
322   - private <T extends TbAbstractDataSubCtx> T getSubCtx(String sessionId, int cmdId) {
323   - Map<Integer, TbAbstractDataSubCtx> sessionSubs = subscriptionsBySessionId.get(sessionId);
  355 + private <T extends TbAbstractSubCtx> T getSubCtx(String sessionId, int cmdId) {
  356 + Map<Integer, TbAbstractSubCtx> sessionSubs = subscriptionsBySessionId.get(sessionId);
324 357 if (sessionSubs != null) {
325 358 return (T) sessionSubs.get(cmdId);
326 359 } else {
... ... @@ -464,17 +497,16 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
464 497 cleanupAndCancel(getSubCtx(sessionId, cmd.getCmdId()));
465 498 }
466 499
467   - private void cleanupAndCancel(TbAbstractDataSubCtx ctx) {
  500 + private void cleanupAndCancel(TbAbstractSubCtx ctx) {
468 501 if (ctx != null) {
469 502 ctx.cancelTasks();
470   - ctx.clearEntitySubscriptions();
471   - ctx.clearDynamicValueSubscriptions();
  503 + ctx.clearSubscriptions();
472 504 }
473 505 }
474 506
475 507 @Override
476 508 public void cancelAllSessionSubscriptions(String sessionId) {
477   - Map<Integer, TbAbstractDataSubCtx> sessionSubs = subscriptionsBySessionId.remove(sessionId);
  509 + Map<Integer, TbAbstractSubCtx> sessionSubs = subscriptionsBySessionId.remove(sessionId);
478 510 if (sessionSubs != null) {
479 511 sessionSubs.values().forEach(this::cleanupAndCancel);
480 512 }
... ...
... ... @@ -15,32 +15,16 @@
15 15 */
16 16 package org.thingsboard.server.service.subscription;
17 17
18   -import com.google.common.util.concurrent.Futures;
19   -import com.google.common.util.concurrent.ListenableFuture;
20   -import com.google.common.util.concurrent.MoreExecutors;
21   -import lombok.Data;
22 18 import lombok.Getter;
23   -import lombok.Setter;
24 19 import lombok.extern.slf4j.Slf4j;
25   -import org.thingsboard.server.common.data.id.CustomerId;
26 20 import org.thingsboard.server.common.data.id.EntityId;
27   -import org.thingsboard.server.common.data.id.TenantId;
28   -import org.thingsboard.server.common.data.id.UserId;
29   -import org.thingsboard.server.common.data.kv.AttributeKvEntry;
30 21 import org.thingsboard.server.common.data.page.PageData;
31 22 import org.thingsboard.server.common.data.query.AbstractDataQuery;
32   -import org.thingsboard.server.common.data.query.ComplexFilterPredicate;
33   -import org.thingsboard.server.common.data.query.DynamicValue;
34   -import org.thingsboard.server.common.data.query.DynamicValueSourceType;
35 23 import org.thingsboard.server.common.data.query.EntityData;
36 24 import org.thingsboard.server.common.data.query.EntityDataPageLink;
37 25 import org.thingsboard.server.common.data.query.EntityDataQuery;
38 26 import org.thingsboard.server.common.data.query.EntityKey;
39 27 import org.thingsboard.server.common.data.query.EntityKeyType;
40   -import org.thingsboard.server.common.data.query.FilterPredicateType;
41   -import org.thingsboard.server.common.data.query.KeyFilter;
42   -import org.thingsboard.server.common.data.query.KeyFilterPredicate;
43   -import org.thingsboard.server.common.data.query.SimpleKeyFilterPredicate;
44 28 import org.thingsboard.server.common.data.query.TsValue;
45 29 import org.thingsboard.server.dao.attributes.AttributesService;
46 30 import org.thingsboard.server.dao.entity.EntityService;
... ... @@ -52,140 +36,25 @@ import java.util.ArrayList;
52 36 import java.util.Arrays;
53 37 import java.util.Collections;
54 38 import java.util.HashMap;
55   -import java.util.HashSet;
56 39 import java.util.List;
57 40 import java.util.Map;
58   -import java.util.Optional;
59   -import java.util.Set;
60 41 import java.util.concurrent.ConcurrentHashMap;
61   -import java.util.concurrent.ExecutionException;
62   -import java.util.concurrent.ScheduledFuture;
63 42 import java.util.function.Function;
64 43 import java.util.stream.Collectors;
65 44
66 45 @Slf4j
67   -@Data
68   -public abstract class TbAbstractDataSubCtx<T extends AbstractDataQuery<? extends EntityDataPageLink>> {
  46 +public abstract class TbAbstractDataSubCtx<T extends AbstractDataQuery<? extends EntityDataPageLink>> extends TbAbstractSubCtx<T> {
69 47
70   - protected final String serviceId;
71   - protected final SubscriptionServiceStatistics stats;
72   - protected final TelemetryWebSocketService wsService;
73   - protected final EntityService entityService;
74   - protected final TbLocalSubscriptionService localSubscriptionService;
75   - protected final AttributesService attributesService;
76   - protected final TelemetryWebSocketSessionRef sessionRef;
77   - protected final int cmdId;
78 48 protected final Map<Integer, EntityId> subToEntityIdMap;
79   - protected final Set<Integer> subToDynamicValueKeySet;
80   - @Getter
81   - protected final Map<DynamicValueKey, List<DynamicValue>> dynamicValues;
82 49 @Getter
83 50 protected PageData<EntityData> data;
84   - @Getter
85   - @Setter
86   - protected T query;
87   - @Setter
88   - protected volatile ScheduledFuture<?> refreshTask;
89 51
90 52 public TbAbstractDataSubCtx(String serviceId, TelemetryWebSocketService wsService,
91 53 EntityService entityService, TbLocalSubscriptionService localSubscriptionService,
92 54 AttributesService attributesService, SubscriptionServiceStatistics stats,
93 55 TelemetryWebSocketSessionRef sessionRef, int cmdId) {
94   - this.serviceId = serviceId;
95   - this.wsService = wsService;
96   - this.entityService = entityService;
97   - this.localSubscriptionService = localSubscriptionService;
98   - this.attributesService = attributesService;
99   - this.stats = stats;
100   - this.sessionRef = sessionRef;
101   - this.cmdId = cmdId;
  56 + super(serviceId, wsService, entityService, localSubscriptionService, attributesService, stats, sessionRef, cmdId);
102 57 this.subToEntityIdMap = new ConcurrentHashMap<>();
103   - this.subToDynamicValueKeySet = ConcurrentHashMap.newKeySet();
104   - this.dynamicValues = new ConcurrentHashMap<>();
105   - }
106   -
107   - public void setAndResolveQuery(T query) {
108   - dynamicValues.clear();
109   - this.query = query;
110   - if (query != null && query.getKeyFilters() != null) {
111   - for (KeyFilter filter : query.getKeyFilters()) {
112   - registerDynamicValues(filter.getPredicate());
113   - }
114   - }
115   - resolve(getTenantId(), getCustomerId(), getUserId());
116   - }
117   -
118   - public void resolve(TenantId tenantId, CustomerId customerId, UserId userId) {
119   - List<ListenableFuture<DynamicValueKeySub>> futures = new ArrayList<>();
120   - for (DynamicValueKey key : dynamicValues.keySet()) {
121   - switch (key.getSourceType()) {
122   - case CURRENT_TENANT:
123   - futures.add(resolveEntityValue(tenantId, tenantId, key));
124   - break;
125   - case CURRENT_CUSTOMER:
126   - if (customerId != null && !customerId.isNullUid()) {
127   - futures.add(resolveEntityValue(tenantId, customerId, key));
128   - }
129   - break;
130   - case CURRENT_USER:
131   - if (userId != null && !userId.isNullUid()) {
132   - futures.add(resolveEntityValue(tenantId, userId, key));
133   - }
134   - break;
135   - }
136   - }
137   - try {
138   - Map<EntityId, Map<String, DynamicValueKeySub>> tmpSubMap = new HashMap<>();
139   - for (DynamicValueKeySub sub : Futures.successfulAsList(futures).get()) {
140   - tmpSubMap.computeIfAbsent(sub.getEntityId(), tmp -> new HashMap<>()).put(sub.getKey().getSourceAttribute(), sub);
141   - }
142   - for (EntityId entityId : tmpSubMap.keySet()) {
143   - Map<String, Long> keyStates = new HashMap<>();
144   - Map<String, DynamicValueKeySub> dynamicValueKeySubMap = tmpSubMap.get(entityId);
145   - dynamicValueKeySubMap.forEach((k, v) -> keyStates.put(k, v.getLastUpdateTs()));
146   - int subIdx = sessionRef.getSessionSubIdSeq().incrementAndGet();
147   - TbAttributeSubscription sub = TbAttributeSubscription.builder()
148   - .serviceId(serviceId)
149   - .sessionId(sessionRef.getSessionId())
150   - .subscriptionId(subIdx)
151   - .tenantId(sessionRef.getSecurityCtx().getTenantId())
152   - .entityId(entityId)
153   - .updateConsumer((s, subscriptionUpdate) -> dynamicValueSubUpdate(s, subscriptionUpdate, dynamicValueKeySubMap))
154   - .allKeys(false)
155   - .keyStates(keyStates)
156   - .scope(TbAttributeSubscriptionScope.SERVER_SCOPE)
157   - .build();
158   - subToDynamicValueKeySet.add(subIdx);
159   - localSubscriptionService.addSubscription(sub);
160   - }
161   - } catch (InterruptedException | ExecutionException e) {
162   - log.info("[{}][{}][{}] Failed to resolve dynamic values: {}", tenantId, customerId, userId, dynamicValues.keySet());
163   - }
164   -
165   - }
166   -
167   - private void dynamicValueSubUpdate(String sessionId, TelemetrySubscriptionUpdate subscriptionUpdate,
168   - Map<String, DynamicValueKeySub> dynamicValueKeySubMap) {
169   - Map<String, TsValue> latestUpdate = new HashMap<>();
170   - subscriptionUpdate.getData().forEach((k, v) -> {
171   - Object[] data = (Object[]) v.get(0);
172   - latestUpdate.put(k, new TsValue((Long) data[0], (String) data[1]));
173   - });
174   -
175   - boolean invalidateFilter = false;
176   - for (Map.Entry<String, TsValue> entry : latestUpdate.entrySet()) {
177   - String k = entry.getKey();
178   - TsValue tsValue = entry.getValue();
179   - DynamicValueKeySub sub = dynamicValueKeySubMap.get(k);
180   - if (sub.updateValue(tsValue)) {
181   - invalidateFilter = true;
182   - updateDynamicValuesByKey(sub, tsValue);
183   - }
184   - }
185   -
186   - if (invalidateFilter) {
187   - update();
188   - }
189 58 }
190 59
191 60 public void fetchData() {
... ... @@ -231,104 +100,10 @@ public abstract class TbAbstractDataSubCtx<T extends AbstractDataQuery<? extends
231 100 return data.getData();
232 101 }
233 102
234   - @Data
235   - private static class DynamicValueKeySub {
236   - private final DynamicValueKey key;
237   - private final EntityId entityId;
238   - private long lastUpdateTs;
239   - private String lastUpdateValue;
240   -
241   - boolean updateValue(TsValue value) {
242   - if (value.getTs() > lastUpdateTs && (lastUpdateValue == null || !lastUpdateValue.equals(value.getValue()))) {
243   - this.lastUpdateTs = value.getTs();
244   - this.lastUpdateValue = value.getValue();
245   - return true;
246   - } else {
247   - return false;
248   - }
249   - }
250   - }
251   -
252   - private ListenableFuture<DynamicValueKeySub> resolveEntityValue(TenantId tenantId, EntityId entityId, DynamicValueKey key) {
253   - ListenableFuture<Optional<AttributeKvEntry>> entry = attributesService.find(tenantId, entityId,
254   - TbAttributeSubscriptionScope.SERVER_SCOPE.name(), key.getSourceAttribute());
255   - return Futures.transform(entry, attributeOpt -> {
256   - DynamicValueKeySub sub = new DynamicValueKeySub(key, entityId);
257   - if (attributeOpt.isPresent()) {
258   - AttributeKvEntry attribute = attributeOpt.get();
259   - sub.setLastUpdateTs(attribute.getLastUpdateTs());
260   - sub.setLastUpdateValue(attribute.getValueAsString());
261   - updateDynamicValuesByKey(sub, new TsValue(attribute.getLastUpdateTs(), attribute.getValueAsString()));
262   - }
263   - return sub;
264   - }, MoreExecutors.directExecutor());
265   - }
266   -
267   - @SuppressWarnings("unchecked")
268   - private void updateDynamicValuesByKey(DynamicValueKeySub sub, TsValue tsValue) {
269   - DynamicValueKey dvk = sub.getKey();
270   - switch (dvk.getPredicateType()) {
271   - case STRING:
272   - dynamicValues.get(dvk).forEach(dynamicValue -> dynamicValue.setResolvedValue(tsValue.getValue()));
273   - break;
274   - case NUMERIC:
275   - try {
276   - Double dValue = Double.parseDouble(tsValue.getValue());
277   - dynamicValues.get(dvk).forEach(dynamicValue -> dynamicValue.setResolvedValue(dValue));
278   - } catch (NumberFormatException e) {
279   - dynamicValues.get(dvk).forEach(dynamicValue -> dynamicValue.setResolvedValue(null));
280   - }
281   - break;
282   - case BOOLEAN:
283   - Boolean bValue = Boolean.parseBoolean(tsValue.getValue());
284   - dynamicValues.get(dvk).forEach(dynamicValue -> dynamicValue.setResolvedValue(bValue));
285   - break;
286   - }
287   - }
288   -
289   - @SuppressWarnings("unchecked")
290   - private void registerDynamicValues(KeyFilterPredicate predicate) {
291   - switch (predicate.getType()) {
292   - case STRING:
293   - case NUMERIC:
294   - case BOOLEAN:
295   - Optional<DynamicValue> value = getDynamicValueFromSimplePredicate((SimpleKeyFilterPredicate) predicate);
296   - if (value.isPresent()) {
297   - DynamicValue dynamicValue = value.get();
298   - DynamicValueKey key = new DynamicValueKey(
299   - predicate.getType(),
300   - dynamicValue.getSourceType(),
301   - dynamicValue.getSourceAttribute());
302   - dynamicValues.computeIfAbsent(key, tmp -> new ArrayList<>()).add(dynamicValue);
303   - }
304   - break;
305   - case COMPLEX:
306   - ((ComplexFilterPredicate) predicate).getPredicates().forEach(this::registerDynamicValues);
307   - }
308   - }
309   -
310   - private Optional<DynamicValue<T>> getDynamicValueFromSimplePredicate(SimpleKeyFilterPredicate<T> predicate) {
311   - if (predicate.getValue().getUserValue() == null) {
312   - return Optional.ofNullable(predicate.getValue().getDynamicValue());
313   - } else {
314   - return Optional.empty();
315   - }
316   - }
317   -
318   - public String getSessionId() {
319   - return sessionRef.getSessionId();
320   - }
321   -
322   - public TenantId getTenantId() {
323   - return sessionRef.getSecurityCtx().getTenantId();
324   - }
325   -
326   - public CustomerId getCustomerId() {
327   - return sessionRef.getSecurityCtx().getCustomerId();
328   - }
329   -
330   - public UserId getUserId() {
331   - return sessionRef.getSecurityCtx().getId();
  103 + @Override
  104 + public void clearSubscriptions() {
  105 + clearEntitySubscriptions();
  106 + super.clearSubscriptions();
332 107 }
333 108
334 109 public void clearEntitySubscriptions() {
... ... @@ -340,26 +115,6 @@ public abstract class TbAbstractDataSubCtx<T extends AbstractDataQuery<? extends
340 115 }
341 116 }
342 117
343   - public void clearDynamicValueSubscriptions() {
344   - if (subToDynamicValueKeySet != null) {
345   - for (Integer subId : subToDynamicValueKeySet) {
346   - localSubscriptionService.cancelSubscription(sessionRef.getSessionId(), subId);
347   - }
348   - subToDynamicValueKeySet.clear();
349   - }
350   - }
351   -
352   - public void setRefreshTask(ScheduledFuture<?> task) {
353   - this.refreshTask = task;
354   - }
355   -
356   - public void cancelTasks() {
357   - if (this.refreshTask != null) {
358   - log.trace("[{}][{}] Canceling old refresh task", sessionRef.getSessionId(), cmdId);
359   - this.refreshTask.cancel(true);
360   - }
361   - }
362   -
363 118 public void createSubscriptions(List<EntityKey> keys, boolean resultToLatestValues) {
364 119 Map<EntityKeyType, List<EntityKey>> keysByType = getEntityKeyByTypeMap(keys);
365 120 for (EntityData entityData : data.getData()) {
... ... @@ -459,14 +214,4 @@ public abstract class TbAbstractDataSubCtx<T extends AbstractDataQuery<? extends
459 214
460 215 abstract void sendWsMsg(String sessionId, TelemetrySubscriptionUpdate subscriptionUpdate, EntityKeyType keyType, boolean resultToLatestValues);
461 216
462   - @Data
463   - private static class DynamicValueKey {
464   - @Getter
465   - private final FilterPredicateType predicateType;
466   - @Getter
467   - private final DynamicValueSourceType sourceType;
468   - @Getter
469   - private final String sourceAttribute;
470   - }
471   -
472 217 }
... ...
  1 +/**
  2 + * Copyright © 2016-2021 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.subscription;
  17 +
  18 +import com.google.common.util.concurrent.Futures;
  19 +import com.google.common.util.concurrent.ListenableFuture;
  20 +import com.google.common.util.concurrent.MoreExecutors;
  21 +import lombok.Data;
  22 +import lombok.Getter;
  23 +import lombok.Setter;
  24 +import lombok.extern.slf4j.Slf4j;
  25 +import org.thingsboard.server.common.data.id.CustomerId;
  26 +import org.thingsboard.server.common.data.id.EntityId;
  27 +import org.thingsboard.server.common.data.id.TenantId;
  28 +import org.thingsboard.server.common.data.id.UserId;
  29 +import org.thingsboard.server.common.data.kv.AttributeKvEntry;
  30 +import org.thingsboard.server.common.data.query.ComplexFilterPredicate;
  31 +import org.thingsboard.server.common.data.query.DynamicValue;
  32 +import org.thingsboard.server.common.data.query.DynamicValueSourceType;
  33 +import org.thingsboard.server.common.data.query.EntityCountQuery;
  34 +import org.thingsboard.server.common.data.query.EntityKeyType;
  35 +import org.thingsboard.server.common.data.query.FilterPredicateType;
  36 +import org.thingsboard.server.common.data.query.KeyFilter;
  37 +import org.thingsboard.server.common.data.query.KeyFilterPredicate;
  38 +import org.thingsboard.server.common.data.query.SimpleKeyFilterPredicate;
  39 +import org.thingsboard.server.common.data.query.TsValue;
  40 +import org.thingsboard.server.dao.attributes.AttributesService;
  41 +import org.thingsboard.server.dao.entity.EntityService;
  42 +import org.thingsboard.server.service.telemetry.TelemetryWebSocketService;
  43 +import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef;
  44 +import org.thingsboard.server.service.telemetry.sub.TelemetrySubscriptionUpdate;
  45 +
  46 +import java.util.ArrayList;
  47 +import java.util.HashMap;
  48 +import java.util.List;
  49 +import java.util.Map;
  50 +import java.util.Optional;
  51 +import java.util.Set;
  52 +import java.util.concurrent.ConcurrentHashMap;
  53 +import java.util.concurrent.ExecutionException;
  54 +import java.util.concurrent.ScheduledFuture;
  55 +
  56 +@Slf4j
  57 +@Data
  58 +public abstract class TbAbstractSubCtx<T extends EntityCountQuery> {
  59 +
  60 + protected final String serviceId;
  61 + protected final SubscriptionServiceStatistics stats;
  62 + protected final TelemetryWebSocketService wsService;
  63 + protected final EntityService entityService;
  64 + protected final TbLocalSubscriptionService localSubscriptionService;
  65 + protected final AttributesService attributesService;
  66 + protected final TelemetryWebSocketSessionRef sessionRef;
  67 + protected final int cmdId;
  68 + protected final Set<Integer> subToDynamicValueKeySet;
  69 + @Getter
  70 + protected final Map<DynamicValueKey, List<DynamicValue>> dynamicValues;
  71 + @Getter
  72 + @Setter
  73 + protected T query;
  74 + @Setter
  75 + protected volatile ScheduledFuture<?> refreshTask;
  76 +
  77 + public TbAbstractSubCtx(String serviceId, TelemetryWebSocketService wsService,
  78 + EntityService entityService, TbLocalSubscriptionService localSubscriptionService,
  79 + AttributesService attributesService, SubscriptionServiceStatistics stats,
  80 + TelemetryWebSocketSessionRef sessionRef, int cmdId) {
  81 + this.serviceId = serviceId;
  82 + this.wsService = wsService;
  83 + this.entityService = entityService;
  84 + this.localSubscriptionService = localSubscriptionService;
  85 + this.attributesService = attributesService;
  86 + this.stats = stats;
  87 + this.sessionRef = sessionRef;
  88 + this.cmdId = cmdId;
  89 + this.subToDynamicValueKeySet = ConcurrentHashMap.newKeySet();
  90 + this.dynamicValues = new ConcurrentHashMap<>();
  91 + }
  92 +
  93 + public void setAndResolveQuery(T query) {
  94 + dynamicValues.clear();
  95 + this.query = query;
  96 + if (query != null && query.getKeyFilters() != null) {
  97 + for (KeyFilter filter : query.getKeyFilters()) {
  98 + registerDynamicValues(filter.getPredicate());
  99 + }
  100 + }
  101 + resolve(getTenantId(), getCustomerId(), getUserId());
  102 + }
  103 +
  104 + public void resolve(TenantId tenantId, CustomerId customerId, UserId userId) {
  105 + List<ListenableFuture<DynamicValueKeySub>> futures = new ArrayList<>();
  106 + for (DynamicValueKey key : dynamicValues.keySet()) {
  107 + switch (key.getSourceType()) {
  108 + case CURRENT_TENANT:
  109 + futures.add(resolveEntityValue(tenantId, tenantId, key));
  110 + break;
  111 + case CURRENT_CUSTOMER:
  112 + if (customerId != null && !customerId.isNullUid()) {
  113 + futures.add(resolveEntityValue(tenantId, customerId, key));
  114 + }
  115 + break;
  116 + case CURRENT_USER:
  117 + if (userId != null && !userId.isNullUid()) {
  118 + futures.add(resolveEntityValue(tenantId, userId, key));
  119 + }
  120 + break;
  121 + }
  122 + }
  123 + try {
  124 + Map<EntityId, Map<String, DynamicValueKeySub>> tmpSubMap = new HashMap<>();
  125 + for (DynamicValueKeySub sub : Futures.successfulAsList(futures).get()) {
  126 + tmpSubMap.computeIfAbsent(sub.getEntityId(), tmp -> new HashMap<>()).put(sub.getKey().getSourceAttribute(), sub);
  127 + }
  128 + for (EntityId entityId : tmpSubMap.keySet()) {
  129 + Map<String, Long> keyStates = new HashMap<>();
  130 + Map<String, DynamicValueKeySub> dynamicValueKeySubMap = tmpSubMap.get(entityId);
  131 + dynamicValueKeySubMap.forEach((k, v) -> keyStates.put(k, v.getLastUpdateTs()));
  132 + int subIdx = sessionRef.getSessionSubIdSeq().incrementAndGet();
  133 + TbAttributeSubscription sub = TbAttributeSubscription.builder()
  134 + .serviceId(serviceId)
  135 + .sessionId(sessionRef.getSessionId())
  136 + .subscriptionId(subIdx)
  137 + .tenantId(sessionRef.getSecurityCtx().getTenantId())
  138 + .entityId(entityId)
  139 + .updateConsumer((s, subscriptionUpdate) -> dynamicValueSubUpdate(s, subscriptionUpdate, dynamicValueKeySubMap))
  140 + .allKeys(false)
  141 + .keyStates(keyStates)
  142 + .scope(TbAttributeSubscriptionScope.SERVER_SCOPE)
  143 + .build();
  144 + subToDynamicValueKeySet.add(subIdx);
  145 + localSubscriptionService.addSubscription(sub);
  146 + }
  147 + } catch (InterruptedException | ExecutionException e) {
  148 + log.info("[{}][{}][{}] Failed to resolve dynamic values: {}", tenantId, customerId, userId, dynamicValues.keySet());
  149 + }
  150 +
  151 + }
  152 +
  153 + private void dynamicValueSubUpdate(String sessionId, TelemetrySubscriptionUpdate subscriptionUpdate,
  154 + Map<String, DynamicValueKeySub> dynamicValueKeySubMap) {
  155 + Map<String, TsValue> latestUpdate = new HashMap<>();
  156 + subscriptionUpdate.getData().forEach((k, v) -> {
  157 + Object[] data = (Object[]) v.get(0);
  158 + latestUpdate.put(k, new TsValue((Long) data[0], (String) data[1]));
  159 + });
  160 +
  161 + boolean invalidateFilter = false;
  162 + for (Map.Entry<String, TsValue> entry : latestUpdate.entrySet()) {
  163 + String k = entry.getKey();
  164 + TsValue tsValue = entry.getValue();
  165 + DynamicValueKeySub sub = dynamicValueKeySubMap.get(k);
  166 + if (sub.updateValue(tsValue)) {
  167 + invalidateFilter = true;
  168 + updateDynamicValuesByKey(sub, tsValue);
  169 + }
  170 + }
  171 +
  172 + if (invalidateFilter) {
  173 + update();
  174 + }
  175 + }
  176 +
  177 + public abstract void fetchData();
  178 +
  179 + protected abstract void update();
  180 +
  181 + public void clearSubscriptions() {
  182 + clearDynamicValueSubscriptions();
  183 + }
  184 +
  185 + @Data
  186 + private static class DynamicValueKeySub {
  187 + private final DynamicValueKey key;
  188 + private final EntityId entityId;
  189 + private long lastUpdateTs;
  190 + private String lastUpdateValue;
  191 +
  192 + boolean updateValue(TsValue value) {
  193 + if (value.getTs() > lastUpdateTs && (lastUpdateValue == null || !lastUpdateValue.equals(value.getValue()))) {
  194 + this.lastUpdateTs = value.getTs();
  195 + this.lastUpdateValue = value.getValue();
  196 + return true;
  197 + } else {
  198 + return false;
  199 + }
  200 + }
  201 + }
  202 +
  203 + private ListenableFuture<DynamicValueKeySub> resolveEntityValue(TenantId tenantId, EntityId entityId, DynamicValueKey key) {
  204 + ListenableFuture<Optional<AttributeKvEntry>> entry = attributesService.find(tenantId, entityId,
  205 + TbAttributeSubscriptionScope.SERVER_SCOPE.name(), key.getSourceAttribute());
  206 + return Futures.transform(entry, attributeOpt -> {
  207 + DynamicValueKeySub sub = new DynamicValueKeySub(key, entityId);
  208 + if (attributeOpt.isPresent()) {
  209 + AttributeKvEntry attribute = attributeOpt.get();
  210 + sub.setLastUpdateTs(attribute.getLastUpdateTs());
  211 + sub.setLastUpdateValue(attribute.getValueAsString());
  212 + updateDynamicValuesByKey(sub, new TsValue(attribute.getLastUpdateTs(), attribute.getValueAsString()));
  213 + }
  214 + return sub;
  215 + }, MoreExecutors.directExecutor());
  216 + }
  217 +
  218 + @SuppressWarnings("unchecked")
  219 + protected void updateDynamicValuesByKey(DynamicValueKeySub sub, TsValue tsValue) {
  220 + DynamicValueKey dvk = sub.getKey();
  221 + switch (dvk.getPredicateType()) {
  222 + case STRING:
  223 + dynamicValues.get(dvk).forEach(dynamicValue -> dynamicValue.setResolvedValue(tsValue.getValue()));
  224 + break;
  225 + case NUMERIC:
  226 + try {
  227 + Double dValue = Double.parseDouble(tsValue.getValue());
  228 + dynamicValues.get(dvk).forEach(dynamicValue -> dynamicValue.setResolvedValue(dValue));
  229 + } catch (NumberFormatException e) {
  230 + dynamicValues.get(dvk).forEach(dynamicValue -> dynamicValue.setResolvedValue(null));
  231 + }
  232 + break;
  233 + case BOOLEAN:
  234 + Boolean bValue = Boolean.parseBoolean(tsValue.getValue());
  235 + dynamicValues.get(dvk).forEach(dynamicValue -> dynamicValue.setResolvedValue(bValue));
  236 + break;
  237 + }
  238 + }
  239 +
  240 + @SuppressWarnings("unchecked")
  241 + private void registerDynamicValues(KeyFilterPredicate predicate) {
  242 + switch (predicate.getType()) {
  243 + case STRING:
  244 + case NUMERIC:
  245 + case BOOLEAN:
  246 + Optional<DynamicValue> value = getDynamicValueFromSimplePredicate((SimpleKeyFilterPredicate) predicate);
  247 + if (value.isPresent()) {
  248 + DynamicValue dynamicValue = value.get();
  249 + DynamicValueKey key = new DynamicValueKey(
  250 + predicate.getType(),
  251 + dynamicValue.getSourceType(),
  252 + dynamicValue.getSourceAttribute());
  253 + dynamicValues.computeIfAbsent(key, tmp -> new ArrayList<>()).add(dynamicValue);
  254 + }
  255 + break;
  256 + case COMPLEX:
  257 + ((ComplexFilterPredicate) predicate).getPredicates().forEach(this::registerDynamicValues);
  258 + }
  259 + }
  260 +
  261 + private Optional<DynamicValue<T>> getDynamicValueFromSimplePredicate(SimpleKeyFilterPredicate<T> predicate) {
  262 + if (predicate.getValue().getUserValue() == null) {
  263 + return Optional.ofNullable(predicate.getValue().getDynamicValue());
  264 + } else {
  265 + return Optional.empty();
  266 + }
  267 + }
  268 +
  269 + public String getSessionId() {
  270 + return sessionRef.getSessionId();
  271 + }
  272 +
  273 + public TenantId getTenantId() {
  274 + return sessionRef.getSecurityCtx().getTenantId();
  275 + }
  276 +
  277 + public CustomerId getCustomerId() {
  278 + return sessionRef.getSecurityCtx().getCustomerId();
  279 + }
  280 +
  281 + public UserId getUserId() {
  282 + return sessionRef.getSecurityCtx().getId();
  283 + }
  284 +
  285 + protected void clearDynamicValueSubscriptions() {
  286 + if (subToDynamicValueKeySet != null) {
  287 + for (Integer subId : subToDynamicValueKeySet) {
  288 + localSubscriptionService.cancelSubscription(sessionRef.getSessionId(), subId);
  289 + }
  290 + subToDynamicValueKeySet.clear();
  291 + }
  292 + }
  293 +
  294 + public void setRefreshTask(ScheduledFuture<?> task) {
  295 + this.refreshTask = task;
  296 + }
  297 +
  298 + public void cancelTasks() {
  299 + if (this.refreshTask != null) {
  300 + log.trace("[{}][{}] Canceling old refresh task", sessionRef.getSessionId(), cmdId);
  301 + this.refreshTask.cancel(true);
  302 + }
  303 + }
  304 +
  305 + @Data
  306 + public static class DynamicValueKey {
  307 + @Getter
  308 + private final FilterPredicateType predicateType;
  309 + @Getter
  310 + private final DynamicValueSourceType sourceType;
  311 + @Getter
  312 + private final String sourceAttribute;
  313 + }
  314 +
  315 +}
... ...
... ... @@ -90,8 +90,7 @@ public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx<AlarmDataQuery> {
90 90 AlarmDataUpdate update;
91 91 if (!entitiesMap.isEmpty()) {
92 92 long start = System.currentTimeMillis();
93   - PageData<AlarmData> alarms = alarmService.findAlarmDataByQueryForEntities(getTenantId(), getCustomerId(),
94   - query, getOrderedEntityIds());
  93 + PageData<AlarmData> alarms = alarmService.findAlarmDataByQueryForEntities(getTenantId(), getCustomerId(), query, getOrderedEntityIds());
95 94 long end = System.currentTimeMillis();
96 95 stats.getAlarmQueryInvocationCnt().incrementAndGet();
97 96 stats.getAlarmQueryTimeSpent().addAndGet(end - start);
... ...
  1 +/**
  2 + * Copyright © 2016-2021 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.subscription;
  17 +
  18 +import lombok.extern.slf4j.Slf4j;
  19 +import org.thingsboard.server.common.data.query.EntityCountQuery;
  20 +import org.thingsboard.server.common.data.query.EntityKeyType;
  21 +import org.thingsboard.server.dao.attributes.AttributesService;
  22 +import org.thingsboard.server.dao.entity.EntityService;
  23 +import org.thingsboard.server.service.telemetry.TelemetryWebSocketService;
  24 +import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef;
  25 +import org.thingsboard.server.service.telemetry.cmd.v2.EntityCountUpdate;
  26 +import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate;
  27 +import org.thingsboard.server.service.telemetry.sub.TelemetrySubscriptionUpdate;
  28 +
  29 +@Slf4j
  30 +public class TbEntityCountSubCtx extends TbAbstractSubCtx<EntityCountQuery> {
  31 +
  32 + private volatile int result;
  33 +
  34 + public TbEntityCountSubCtx(String serviceId, TelemetryWebSocketService wsService, EntityService entityService,
  35 + TbLocalSubscriptionService localSubscriptionService, AttributesService attributesService,
  36 + SubscriptionServiceStatistics stats, TelemetryWebSocketSessionRef sessionRef, int cmdId) {
  37 + super(serviceId, wsService, entityService, localSubscriptionService, attributesService, stats, sessionRef, cmdId);
  38 + }
  39 +
  40 + @Override
  41 + public void fetchData() {
  42 + result = (int) entityService.countEntitiesByQuery(getTenantId(), getCustomerId(), query);
  43 + wsService.sendWsMsg(sessionRef.getSessionId(), new EntityCountUpdate(cmdId, result));
  44 + }
  45 +
  46 + @Override
  47 + protected void update() {
  48 + int newCount = (int) entityService.countEntitiesByQuery(getTenantId(), getCustomerId(), query);
  49 + if (newCount != result) {
  50 + result = newCount;
  51 + wsService.sendWsMsg(sessionRef.getSessionId(), new EntityCountUpdate(cmdId, result));
  52 + }
  53 + }
  54 +
  55 +}
... ...
... ... @@ -17,6 +17,7 @@ package org.thingsboard.server.service.subscription;
17 17
18 18 import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef;
19 19 import org.thingsboard.server.service.telemetry.cmd.v2.AlarmDataCmd;
  20 +import org.thingsboard.server.service.telemetry.cmd.v2.EntityCountCmd;
20 21 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd;
21 22 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUnsubscribeCmd;
22 23 import org.thingsboard.server.service.telemetry.cmd.v2.UnsubscribeCmd;
... ... @@ -25,6 +26,8 @@ public interface TbEntityDataSubscriptionService {
25 26
26 27 void handleCmd(TelemetryWebSocketSessionRef sessionId, EntityDataCmd cmd);
27 28
  29 + void handleCmd(TelemetryWebSocketSessionRef sessionId, EntityCountCmd cmd);
  30 +
28 31 void handleCmd(TelemetryWebSocketSessionRef sessionId, AlarmDataCmd cmd);
29 32
30 33 void cancelSubscription(String sessionId, UnsubscribeCmd subscriptionId);
... ...
... ... @@ -51,22 +51,22 @@ import org.thingsboard.server.service.security.ValidationResult;
51 51 import org.thingsboard.server.service.security.ValidationResultCode;
52 52 import org.thingsboard.server.service.security.model.UserPrincipal;
53 53 import org.thingsboard.server.service.security.permission.Operation;
  54 +import org.thingsboard.server.service.subscription.TbAttributeSubscription;
  55 +import org.thingsboard.server.service.subscription.TbAttributeSubscriptionScope;
54 56 import org.thingsboard.server.service.subscription.TbEntityDataSubscriptionService;
55 57 import org.thingsboard.server.service.subscription.TbLocalSubscriptionService;
56   -import org.thingsboard.server.service.subscription.TbAttributeSubscriptionScope;
57   -import org.thingsboard.server.service.subscription.TbAttributeSubscription;
58 58 import org.thingsboard.server.service.subscription.TbTimeseriesSubscription;
  59 +import org.thingsboard.server.service.telemetry.cmd.TelemetryPluginCmdsWrapper;
59 60 import org.thingsboard.server.service.telemetry.cmd.v1.AttributesSubscriptionCmd;
60 61 import org.thingsboard.server.service.telemetry.cmd.v1.GetHistoryCmd;
61 62 import org.thingsboard.server.service.telemetry.cmd.v1.SubscriptionCmd;
62 63 import org.thingsboard.server.service.telemetry.cmd.v1.TelemetryPluginCmd;
63   -import org.thingsboard.server.service.telemetry.cmd.TelemetryPluginCmdsWrapper;
64 64 import org.thingsboard.server.service.telemetry.cmd.v1.TimeseriesSubscriptionCmd;
65 65 import org.thingsboard.server.service.telemetry.cmd.v2.AlarmDataCmd;
66   -import org.thingsboard.server.service.telemetry.cmd.v2.AlarmDataUnsubscribeCmd;
  66 +import org.thingsboard.server.service.telemetry.cmd.v2.CmdUpdate;
67 67 import org.thingsboard.server.service.telemetry.cmd.v2.DataUpdate;
  68 +import org.thingsboard.server.service.telemetry.cmd.v2.EntityCountCmd;
68 69 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd;
69   -import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUnsubscribeCmd;
70 70 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate;
71 71 import org.thingsboard.server.service.telemetry.cmd.v2.UnsubscribeCmd;
72 72 import org.thingsboard.server.service.telemetry.exception.UnauthorizedException;
... ... @@ -216,12 +216,18 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
216 216 if (cmdsWrapper.getAlarmDataCmds() != null) {
217 217 cmdsWrapper.getAlarmDataCmds().forEach(cmd -> handleWsAlarmDataCmd(sessionRef, cmd));
218 218 }
  219 + if (cmdsWrapper.getEntityCountCmds() != null) {
  220 + cmdsWrapper.getEntityCountCmds().forEach(cmd -> handleWsEntityCountCmd(sessionRef, cmd));
  221 + }
219 222 if (cmdsWrapper.getEntityDataUnsubscribeCmds() != null) {
220 223 cmdsWrapper.getEntityDataUnsubscribeCmds().forEach(cmd -> handleWsDataUnsubscribeCmd(sessionRef, cmd));
221 224 }
222 225 if (cmdsWrapper.getAlarmDataUnsubscribeCmds() != null) {
223 226 cmdsWrapper.getAlarmDataUnsubscribeCmds().forEach(cmd -> handleWsDataUnsubscribeCmd(sessionRef, cmd));
224 227 }
  228 + if (cmdsWrapper.getEntityCountUnsubscribeCmds() != null) {
  229 + cmdsWrapper.getEntityCountUnsubscribeCmds().forEach(cmd -> handleWsDataUnsubscribeCmd(sessionRef, cmd));
  230 + }
225 231 }
226 232 } catch (IOException e) {
227 233 log.warn("Failed to decode subscription cmd: {}", e.getMessage(), e);
... ... @@ -239,6 +245,16 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
239 245 }
240 246 }
241 247
  248 + private void handleWsEntityCountCmd(TelemetryWebSocketSessionRef sessionRef, EntityCountCmd cmd) {
  249 + String sessionId = sessionRef.getSessionId();
  250 + log.debug("[{}] Processing: {}", sessionId, cmd);
  251 +
  252 + if (validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId)
  253 + && validateSubscriptionCmd(sessionRef, cmd)) {
  254 + entityDataSubService.handleCmd(sessionRef, cmd);
  255 + }
  256 + }
  257 +
242 258 private void handleWsAlarmDataCmd(TelemetryWebSocketSessionRef sessionRef, AlarmDataCmd cmd) {
243 259 String sessionId = sessionRef.getSessionId();
244 260 log.debug("[{}] Processing: {}", sessionId, cmd);
... ... @@ -264,7 +280,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
264 280 }
265 281
266 282 @Override
267   - public void sendWsMsg(String sessionId, DataUpdate update) {
  283 + public void sendWsMsg(String sessionId, CmdUpdate update) {
268 284 sendWsMsg(sessionId, update.getCmdId(), update);
269 285 }
270 286
... ... @@ -679,6 +695,20 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
679 695 return true;
680 696 }
681 697
  698 + private boolean validateSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, EntityCountCmd cmd) {
  699 + if (cmd.getCmdId() < 0) {
  700 + TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
  701 + "Cmd id is negative value!");
  702 + sendWsMsg(sessionRef, update);
  703 + return false;
  704 + } else if (cmd.getQuery() == null) {
  705 + TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, "Query is empty!");
  706 + sendWsMsg(sessionRef, update);
  707 + return false;
  708 + }
  709 + return true;
  710 + }
  711 +
682 712 private boolean validateSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, AlarmDataCmd cmd) {
683 713 if (cmd.getCmdId() < 0) {
684 714 TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
... ...
... ... @@ -15,6 +15,7 @@
15 15 */
16 16 package org.thingsboard.server.service.telemetry;
17 17
  18 +import org.thingsboard.server.service.telemetry.cmd.v2.CmdUpdate;
18 19 import org.thingsboard.server.service.telemetry.cmd.v2.DataUpdate;
19 20 import org.thingsboard.server.service.telemetry.sub.TelemetrySubscriptionUpdate;
20 21
... ... @@ -29,6 +30,6 @@ public interface TelemetryWebSocketService {
29 30
30 31 void sendWsMsg(String sessionId, TelemetrySubscriptionUpdate update);
31 32
32   - void sendWsMsg(String sessionId, DataUpdate update);
  33 + void sendWsMsg(String sessionId, CmdUpdate update);
33 34
34 35 }
... ...
... ... @@ -15,11 +15,13 @@
15 15 */
16 16 package org.thingsboard.server.service.telemetry.cmd.v2;
17 17
  18 +import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
18 19 import lombok.AllArgsConstructor;
19 20 import lombok.Data;
20 21
21 22 @Data
22 23 @AllArgsConstructor
  24 +@JsonIgnoreProperties(ignoreUnknown = true)
23 25 public abstract class CmdUpdate {
24 26
25 27 private final int cmdId;
... ...
... ... @@ -15,18 +15,12 @@
15 15 */
16 16 package org.thingsboard.server.service.telemetry.cmd.v2;
17 17
18   -import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
19   -import lombok.AllArgsConstructor;
20   -import lombok.Data;
21   -import lombok.EqualsAndHashCode;
22 18 import lombok.Getter;
23   -import lombok.ToString;
24 19 import org.thingsboard.server.common.data.page.PageData;
25 20 import org.thingsboard.server.service.telemetry.sub.SubscriptionErrorCode;
26 21
27 22 import java.util.List;
28 23
29   -@JsonIgnoreProperties(ignoreUnknown = true)
30 24 public abstract class DataUpdate<T> extends CmdUpdate {
31 25
32 26 @Getter
... ...
... ... @@ -35,16 +35,23 @@ import org.thingsboard.server.common.data.kv.LongDataEntry;
35 35 import org.thingsboard.server.common.data.kv.TsKvEntry;
36 36 import org.thingsboard.server.common.data.page.PageData;
37 37 import org.thingsboard.server.common.data.query.DeviceTypeFilter;
  38 +import org.thingsboard.server.common.data.query.EntityCountQuery;
38 39 import org.thingsboard.server.common.data.query.EntityData;
39 40 import org.thingsboard.server.common.data.query.EntityDataPageLink;
40 41 import org.thingsboard.server.common.data.query.EntityDataQuery;
41 42 import org.thingsboard.server.common.data.query.EntityKey;
42 43 import org.thingsboard.server.common.data.query.EntityKeyType;
  44 +import org.thingsboard.server.common.data.query.EntityKeyValueType;
  45 +import org.thingsboard.server.common.data.query.FilterPredicateValue;
  46 +import org.thingsboard.server.common.data.query.KeyFilter;
  47 +import org.thingsboard.server.common.data.query.NumericFilterPredicate;
43 48 import org.thingsboard.server.common.data.query.TsValue;
44 49 import org.thingsboard.server.common.data.security.Authority;
45 50 import org.thingsboard.server.service.subscription.TbAttributeSubscriptionScope;
46 51 import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
47 52 import org.thingsboard.server.service.telemetry.cmd.TelemetryPluginCmdsWrapper;
  53 +import org.thingsboard.server.service.telemetry.cmd.v2.EntityCountCmd;
  54 +import org.thingsboard.server.service.telemetry.cmd.v2.EntityCountUpdate;
48 55 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd;
49 56 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate;
50 57 import org.thingsboard.server.service.telemetry.cmd.v2.EntityHistoryCmd;
... ... @@ -244,6 +251,98 @@ public class BaseWebsocketApiTest extends AbstractWebsocketTest {
244 251 }
245 252
246 253 @Test
  254 + public void testEntityCountWsCmd() throws Exception {
  255 + Device device = new Device();
  256 + device.setName("Device");
  257 + device.setType("default");
  258 + device.setLabel("testLabel" + (int) (Math.random() * 1000));
  259 + device = doPost("/api/device", device, Device.class);
  260 +
  261 + AttributeKvEntry dataPoint1 = new BaseAttributeKvEntry(System.currentTimeMillis(), new LongDataEntry("temperature", 42L));
  262 + sendAttributes(device, TbAttributeSubscriptionScope.SERVER_SCOPE, Collections.singletonList(dataPoint1));
  263 +
  264 + DeviceTypeFilter dtf1 = new DeviceTypeFilter();
  265 + dtf1.setDeviceNameFilter("D");
  266 + dtf1.setDeviceType("default");
  267 + EntityCountQuery edq1 = new EntityCountQuery(dtf1, Collections.emptyList());
  268 +
  269 + EntityCountCmd cmd1 = new EntityCountCmd(1, edq1);
  270 +
  271 + TelemetryPluginCmdsWrapper wrapper1 = new TelemetryPluginCmdsWrapper();
  272 + wrapper1.setEntityCountCmds(Collections.singletonList(cmd1));
  273 +
  274 + wsClient.send(mapper.writeValueAsString(wrapper1));
  275 + String msg1 = wsClient.waitForReply();
  276 + EntityCountUpdate update1 = mapper.readValue(msg1, EntityCountUpdate.class);
  277 + Assert.assertEquals(1, update1.getCmdId());
  278 + Assert.assertEquals(1, update1.getCount());
  279 +
  280 + DeviceTypeFilter dtf2 = new DeviceTypeFilter();
  281 + dtf2.setDeviceNameFilter("D");
  282 + dtf2.setDeviceType("non-existing-device-type");
  283 + EntityCountQuery edq2 = new EntityCountQuery(dtf2, Collections.emptyList());
  284 +
  285 + EntityCountCmd cmd2 = new EntityCountCmd(2, edq2);
  286 +
  287 + TelemetryPluginCmdsWrapper wrapper2 = new TelemetryPluginCmdsWrapper();
  288 + wrapper2.setEntityCountCmds(Collections.singletonList(cmd2));
  289 + wsClient.send(mapper.writeValueAsString(wrapper2));
  290 +
  291 + String msg2 = wsClient.waitForReply();
  292 + EntityCountUpdate update2 = mapper.readValue(msg2, EntityCountUpdate.class);
  293 + Assert.assertEquals(2, update2.getCmdId());
  294 + Assert.assertEquals(0, update2.getCount());
  295 +
  296 + KeyFilter highTemperatureFilter = new KeyFilter();
  297 + highTemperatureFilter.setKey(new EntityKey(EntityKeyType.ATTRIBUTE, "temperature"));
  298 + NumericFilterPredicate predicate = new NumericFilterPredicate();
  299 + predicate.setValue(FilterPredicateValue.fromDouble(40));
  300 + predicate.setOperation(NumericFilterPredicate.NumericOperation.GREATER);
  301 + highTemperatureFilter.setPredicate(predicate);
  302 + highTemperatureFilter.setValueType(EntityKeyValueType.NUMERIC);
  303 +
  304 + DeviceTypeFilter dtf3 = new DeviceTypeFilter();
  305 + dtf3.setDeviceNameFilter("D");
  306 + dtf3.setDeviceType("default");
  307 + EntityCountQuery edq3 = new EntityCountQuery(dtf3, Collections.singletonList(highTemperatureFilter));
  308 +
  309 + EntityCountCmd cmd3 = new EntityCountCmd(3, edq3);
  310 +
  311 + TelemetryPluginCmdsWrapper wrapper3 = new TelemetryPluginCmdsWrapper();
  312 + wrapper3.setEntityCountCmds(Collections.singletonList(cmd3));
  313 + wsClient.send(mapper.writeValueAsString(wrapper3));
  314 +
  315 + String msg3 = wsClient.waitForReply();
  316 + EntityCountUpdate update3 = mapper.readValue(msg3, EntityCountUpdate.class);
  317 + Assert.assertEquals(3, update3.getCmdId());
  318 + Assert.assertEquals(1, update3.getCount());
  319 +
  320 + KeyFilter highTemperatureFilter2 = new KeyFilter();
  321 + highTemperatureFilter2.setKey(new EntityKey(EntityKeyType.ATTRIBUTE, "temperature"));
  322 + NumericFilterPredicate predicate2 = new NumericFilterPredicate();
  323 + predicate2.setValue(FilterPredicateValue.fromDouble(50));
  324 + predicate2.setOperation(NumericFilterPredicate.NumericOperation.GREATER);
  325 + highTemperatureFilter2.setPredicate(predicate2);
  326 + highTemperatureFilter2.setValueType(EntityKeyValueType.NUMERIC);
  327 +
  328 + DeviceTypeFilter dtf4 = new DeviceTypeFilter();
  329 + dtf4.setDeviceNameFilter("D");
  330 + dtf4.setDeviceType("default");
  331 + EntityCountQuery edq4 = new EntityCountQuery(dtf4, Collections.singletonList(highTemperatureFilter2));
  332 +
  333 + EntityCountCmd cmd4 = new EntityCountCmd(4, edq4);
  334 +
  335 + TelemetryPluginCmdsWrapper wrapper4 = new TelemetryPluginCmdsWrapper();
  336 + wrapper4.setEntityCountCmds(Collections.singletonList(cmd4));
  337 + wsClient.send(mapper.writeValueAsString(wrapper4));
  338 +
  339 + String msg4 = wsClient.waitForReply();
  340 + EntityCountUpdate update4 = mapper.readValue(msg4, EntityCountUpdate.class);
  341 + Assert.assertEquals(4, update4.getCmdId());
  342 + Assert.assertEquals(0, update4.getCount());
  343 + }
  344 +
  345 + @Test
247 346 public void testEntityDataLatestWidgetFlow() throws Exception {
248 347 Device device = new Device();
249 348 device.setName("Device");
... ...
... ... @@ -26,9 +26,9 @@ import java.util.Arrays;
26 26
27 27 @RunWith(ClasspathSuite.class)
28 28 @ClasspathSuite.ClassnameFilters({
29   -// "org.thingsboard.server.controller.sql.WebsocketApiSqlTest",
  29 + "org.thingsboard.server.controller.sql.WebsocketApiSqlTest",
30 30 // "org.thingsboard.server.controller.sql.TenantProfileControllerSqlTest",
31   - "org.thingsboard.server.controller.sql.*Test",
  31 +// "org.thingsboard.server.controller.sql.*Test",
32 32 })
33 33 public class ControllerSqlTestSuite {
34 34
... ...
... ... @@ -249,18 +249,70 @@ public class DefaultEntityQueryRepository implements EntityQueryRepository {
249 249 public long countEntitiesByQuery(TenantId tenantId, CustomerId customerId, EntityCountQuery query) {
250 250 EntityType entityType = resolveEntityType(query.getEntityFilter());
251 251 QueryContext ctx = new QueryContext(new QuerySecurityContext(tenantId, customerId, entityType));
252   - ctx.append("select count(e.id) from ");
253   - ctx.append(addEntityTableQuery(ctx, query.getEntityFilter()));
254   - ctx.append(" e where ");
255   - ctx.append(buildEntityWhere(ctx, query.getEntityFilter(), Collections.emptyList()));
256   - return transactionTemplate.execute(status -> {
257   - long startTs = System.currentTimeMillis();
258   - try {
259   - return jdbcTemplate.queryForObject(ctx.getQuery(), ctx, Long.class);
260   - } finally {
261   - queryLog.logQuery(ctx, ctx.getQuery(), System.currentTimeMillis() - startTs);
  252 + if (query.getKeyFilters() == null || query.getKeyFilters().isEmpty()) {
  253 + ctx.append("select count(e.id) from ");
  254 + ctx.append(addEntityTableQuery(ctx, query.getEntityFilter()));
  255 + ctx.append(" e where ");
  256 + ctx.append(buildEntityWhere(ctx, query.getEntityFilter(), Collections.emptyList()));
  257 + return transactionTemplate.execute(status -> {
  258 + long startTs = System.currentTimeMillis();
  259 + try {
  260 + return jdbcTemplate.queryForObject(ctx.getQuery(), ctx, Long.class);
  261 + } finally {
  262 + queryLog.logQuery(ctx, ctx.getQuery(), System.currentTimeMillis() - startTs);
  263 + }
  264 + });
  265 + } else {
  266 + List<EntityKeyMapping> mappings = EntityKeyMapping.prepareEntityCountKeyMapping(query);
  267 +
  268 + List<EntityKeyMapping> selectionMapping = mappings.stream().filter(EntityKeyMapping::isSelection)
  269 + .collect(Collectors.toList());
  270 + List<EntityKeyMapping> entityFieldsSelectionMapping = selectionMapping.stream().filter(mapping -> !mapping.isLatest())
  271 + .collect(Collectors.toList());
  272 +
  273 + List<EntityKeyMapping> filterMapping = mappings.stream().filter(EntityKeyMapping::hasFilter)
  274 + .collect(Collectors.toList());
  275 + List<EntityKeyMapping> entityFieldsFiltersMapping = filterMapping.stream().filter(mapping -> !mapping.isLatest())
  276 + .collect(Collectors.toList());
  277 +
  278 + List<EntityKeyMapping> allLatestMappings = mappings.stream().filter(EntityKeyMapping::isLatest)
  279 + .collect(Collectors.toList());
  280 +
  281 +
  282 + String entityWhereClause = DefaultEntityQueryRepository.this.buildEntityWhere(ctx, query.getEntityFilter(), entityFieldsFiltersMapping);
  283 + String latestJoinsCnt = EntityKeyMapping.buildLatestJoins(ctx, query.getEntityFilter(), entityType, allLatestMappings, true);
  284 + String entityFieldsSelection = EntityKeyMapping.buildSelections(entityFieldsSelectionMapping, query.getEntityFilter().getType(), entityType);
  285 + String entityTypeStr;
  286 + if (query.getEntityFilter().getType().equals(EntityFilterType.RELATIONS_QUERY)) {
  287 + entityTypeStr = "e.entity_type";
  288 + } else {
  289 + entityTypeStr = "'" + entityType.name() + "'";
262 290 }
263   - });
  291 + if (!StringUtils.isEmpty(entityFieldsSelection)) {
  292 + entityFieldsSelection = String.format("e.id id, %s entity_type, %s", entityTypeStr, entityFieldsSelection);
  293 + } else {
  294 + entityFieldsSelection = String.format("e.id id, %s entity_type", entityTypeStr);
  295 + }
  296 +
  297 + String fromClauseCount = String.format("from (select %s from (select %s from %s e where %s) entities %s ) result %s",
  298 + "entities.*",
  299 + entityFieldsSelection,
  300 + addEntityTableQuery(ctx, query.getEntityFilter()),
  301 + entityWhereClause,
  302 + latestJoinsCnt,
  303 + "");
  304 +
  305 + String countQuery = String.format("select count(id) %s", fromClauseCount);
  306 +
  307 + return transactionTemplate.execute(status -> {
  308 + long startTs = System.currentTimeMillis();
  309 + try {
  310 + return jdbcTemplate.queryForObject(countQuery, ctx, Long.class);
  311 + } finally {
  312 + queryLog.logQuery(ctx, ctx.getQuery(), System.currentTimeMillis() - startTs);
  313 + }
  314 + });
  315 + }
264 316 }
265 317
266 318 @Override
... ...
... ... @@ -21,6 +21,7 @@ import org.thingsboard.server.common.data.DataConstants;
21 21 import org.thingsboard.server.common.data.EntityType;
22 22 import org.thingsboard.server.common.data.query.BooleanFilterPredicate;
23 23 import org.thingsboard.server.common.data.query.ComplexFilterPredicate;
  24 +import org.thingsboard.server.common.data.query.EntityCountQuery;
24 25 import org.thingsboard.server.common.data.query.EntityDataQuery;
25 26 import org.thingsboard.server.common.data.query.EntityDataSortOrder;
26 27 import org.thingsboard.server.common.data.query.EntityFilter;
... ... @@ -380,6 +381,30 @@ public class EntityKeyMapping {
380 381 return mappings;
381 382 }
382 383
  384 + public static List<EntityKeyMapping> prepareEntityCountKeyMapping(EntityCountQuery query) {
  385 + Map<EntityKey, List<KeyFilter>> filters =
  386 + query.getKeyFilters() != null ?
  387 + query.getKeyFilters().stream().collect(Collectors.groupingBy(KeyFilter::getKey)) : Collections.emptyMap();
  388 + int index = 2;
  389 + List<EntityKeyMapping> mappings = new ArrayList<>();
  390 + if (!filters.isEmpty()) {
  391 + for (EntityKey filterField : filters.keySet()) {
  392 + EntityKeyMapping mapping = new EntityKeyMapping();
  393 + mapping.setIndex(index);
  394 + mapping.setAlias(String.format("alias%s", index));
  395 + mapping.setKeyFilters(filters.get(filterField));
  396 + mapping.setLatest(!filterField.getType().equals(EntityKeyType.ENTITY_FIELD));
  397 + mapping.setSelection(false);
  398 + mapping.setEntityKey(filterField);
  399 + mappings.add(mapping);
  400 + index += 1;
  401 + }
  402 + }
  403 +
  404 + return mappings;
  405 + }
  406 +
  407 +
383 408 private String buildAttributeSelection() {
384 409 return buildTimeSeriesOrAttrSelection(true);
385 410 }
... ...