Refactoring for TelemetrySubscriptionUpdate

application/src/main/java/org/thingsboard/server/service/subscription/TbAbstractEntityQuerySubCtx.java

@@ -44,6 +44,7 @@ import org.thingsboard.server.service.ws.WebSocketSessionRef;
 import org.thingsboard.server.service.ws.telemetry.sub.TelemetrySubscriptionUpdate;
 
 import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -154,9 +155,8 @@ public abstract class TbAbstractEntityQuerySubCtx<T extends EntityCountQuery> ex
     private void dynamicValueSubUpdate(String sessionId, TelemetrySubscriptionUpdate subscriptionUpdate,
                                        Map<String, DynamicValueKeySub> dynamicValueKeySubMap) {
         Map<String, TsValue> latestUpdate = new HashMap<>();
-        subscriptionUpdate.getData().forEach((k, v) -> {
+        subscriptionUpdate.getData().forEach((key, values) -> {
-            Object[] data = (Object[]) v.get(0);
+            latestUpdate.put(key, getLatest(values));
-            latestUpdate.put(k, new TsValue((Long) data[0], (String) data[1]));
         });
 
         boolean invalidateFilter = false;
@@ -283,6 +283,12 @@ public abstract class TbAbstractEntityQuerySubCtx<T extends EntityCountQuery> ex
         }
     }
 
+    protected TsValue getLatest(List<TsValue> values) {
+        return values.stream()
+                .max(Comparator.comparing(TsValue::getTs))
+                .orElse(null);
+    }
+
     @Data
     public static class DynamicValueKey {
         @Getter

application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmDataSubCtx.java

@@ -38,7 +38,6 @@ import org.thingsboard.server.common.data.query.TsValue;
 import org.thingsboard.server.dao.alarm.AlarmService;
 import org.thingsboard.server.dao.attributes.AttributesService;
 import org.thingsboard.server.dao.entity.EntityService;
-import org.thingsboard.server.dao.model.ModelConstants;
 import org.thingsboard.server.dao.sql.query.EntityKeyMapping;
 import org.thingsboard.server.service.ws.WebSocketService;
 import org.thingsboard.server.service.ws.WebSocketSessionRef;
@@ -191,9 +190,8 @@ public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx<AlarmDataQuery> {
         EntityId entityId = subToEntityIdMap.get(subscriptionUpdate.getSubscriptionId());
         if (entityId != null) {
             Map<String, TsValue> latestUpdate = new HashMap<>();
-            subscriptionUpdate.getData().forEach((k, v) -> {
+            subscriptionUpdate.getData().forEach((key, values) -> {
-                Object[] data = (Object[]) v.get(0);
+                latestUpdate.put(key, getLatest(values));
-                latestUpdate.put(k, new TsValue((Long) data[0], (String) data[1]));
             });
             EntityData entityData = entitiesMap.get(entityId);
             entityData.getLatest().computeIfAbsent(keyType, tmp -> new HashMap<>()).putAll(latestUpdate);

application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java

@@ -43,7 +43,6 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -94,16 +93,12 @@ public class TbEntityDataSubCtx extends TbAbstractDataSubCtx<EntityDataQuery> {
 
     private void sendLatestWsMsg(EntityId entityId, String sessionId, TelemetrySubscriptionUpdate subscriptionUpdate, EntityKeyType keyType) {
         Map<String, TsValue> latestUpdate = new HashMap<>();
-        subscriptionUpdate.getData().forEach((key, data) -> data.stream()
+        subscriptionUpdate.getData().forEach((key, values) -> {
-                .filter(o -> o instanceof Object[] && ((Object[]) o).length > 0 && ((Object[]) o)[0] instanceof Long)
+            latestUpdate.put(key, getLatest(values));
-                .max(Comparator.comparingLong(o -> (Long) ((Object[]) o)[0]))
+        });
-                .ifPresent(max -> {
-                    Object[] arr = (Object[]) max;
-                    latestUpdate.put(key, new TsValue((Long) arr[0], (String) arr[1]));
-                }));
         EntityData entityData = getDataForEntity(entityId);
         if (entityData != null && entityData.getLatest() != null) {
-            Map<String, TsValue> latestCtxValues = entityData.getLatest().computeIfAbsent(keyType, key -> new HashMap<>());
+            Map<String, TsValue> latestCtxValues = entityData.getLatest().computeIfAbsent(keyType, __ -> new HashMap<>());
             log.trace("[{}][{}][{}] Going to compare update with {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), latestCtxValues);
             latestCtxValues.forEach((k, v) -> {
                 TsValue update = latestUpdate.get(k);
@@ -134,40 +129,39 @@ public class TbEntityDataSubCtx extends TbAbstractDataSubCtx<EntityDataQuery> {
 
     private void sendTsWsMsg(EntityId entityId, String sessionId, TelemetrySubscriptionUpdate subscriptionUpdate, EntityKeyType keyType) {
         Map<String, List<TsValue>> tsUpdate = new HashMap<>();
-        subscriptionUpdate.getData().forEach((k, v) -> {
+        subscriptionUpdate.getData().forEach((key, values) -> {
-            Object[] data = (Object[]) v.get(0);
+            tsUpdate.put(key, new ArrayList<>(values));
-            tsUpdate.computeIfAbsent(k, key -> new ArrayList<>()).add(new TsValue((Long) data[0], (String) data[1]));
         });
         Map<String, TsValue> latestCtxValues = getLatestTsValuesForEntity(entityId);
         log.trace("[{}][{}][{}] Going to compare update with {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), latestCtxValues);
         if (latestCtxValues != null) {
-            latestCtxValues.forEach((k, v) -> {
+            latestCtxValues.forEach((key, latest) -> {
-                List<TsValue> updateList = tsUpdate.get(k);
+                List<TsValue> updateList = tsUpdate.get(key);
                 if (updateList != null) {
                     for (TsValue update : new ArrayList<>(updateList)) {
-                        if (update.getTs() < v.getTs()) {
+                        if (update.getTs() < latest.getTs()) {
-                            log.trace("[{}][{}][{}] Removed stale update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs());
+                            log.trace("[{}][{}][{}] Removed stale update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), key, update.getTs());
                             // Looks like this is redundant feature and our UI is ready to merge the updates.
                             //updateList.remove(update);
-                        } else if ((update.getTs() == v.getTs() && update.getValue().equals(v.getValue()))) {
+                        } else if ((update.getTs() == latest.getTs() && update.getValue().equals(latest.getValue()))) {
-                            log.trace("[{}][{}][{}] Removed duplicate update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs());
+                            log.trace("[{}][{}][{}] Removed duplicate update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), key, update.getTs());
                             updateList.remove(update);
                         }
                         if (updateList.isEmpty()) {
-                            tsUpdate.remove(k);
+                            tsUpdate.remove(key);
                         }
                     }
                 }
             });
             //Setting new values
-            tsUpdate.forEach((k, v) -> {
+            tsUpdate.forEach((key, values) -> {
-                Optional<TsValue> maxValue = v.stream().max(Comparator.comparingLong(TsValue::getTs));
+                values.stream().max(Comparator.comparingLong(TsValue::getTs))
-                maxValue.ifPresent(max -> latestCtxValues.put(k, max));
+                        .ifPresent(latest -> latestCtxValues.put(key, latest));
             });
         }
         if (!tsUpdate.isEmpty()) {
             Map<String, TsValue[]> tsMap = new HashMap<>();
-            tsUpdate.forEach((key, tsValue) -> tsMap.put(key, tsValue.toArray(new TsValue[tsValue.size()])));
+            tsUpdate.forEach((key, values) -> tsMap.put(key, values.toArray(new TsValue[0])));
             EntityData entityData = new EntityData(entityId, null, tsMap);
             sendWsMsg(new EntityDataUpdate(cmdId, null, Collections.singletonList(entityData), maxEntitiesPerDataSubscription));
         }

application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java

@@ -274,7 +274,7 @@ public class DefaultWebSocketService implements WebSocketService {
     @Override
     public void sendUpdate(String sessionId, int cmdId, TelemetrySubscriptionUpdate update) {
         // We substitute the subscriptionId with cmdId for old-style subscriptions.
-        doSendUpdate(sessionId, cmdId, update.copyWithNewSubscriptionId(cmdId));
+        doSendUpdate(sessionId, cmdId, update.withSubscriptionId(cmdId));
     }
 
     @Override

application/src/main/java/org/thingsboard/server/service/ws/telemetry/sub/TelemetrySubscriptionUpdate.java

@@ -16,12 +16,16 @@
 package org.thingsboard.server.service.ws.telemetry.sub;
 
 import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.With;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.common.data.query.TsValue;
 import org.thingsboard.server.service.subscription.SubscriptionErrorCode;
 
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
@@ -29,8 +33,13 @@ import java.util.stream.Collectors;
 
 @AllArgsConstructor
 public class TelemetrySubscriptionUpdate {
+
+    @Getter
+    @With
     private final int subscriptionId;
+    @Getter
     private int errorCode;
+    @Getter
     private String errorMsg;
     private Map<String, List<Object>> data;
 
@@ -66,11 +75,27 @@ public class TelemetrySubscriptionUpdate {
         this.errorMsg = errorMsg != null ? errorMsg : errorCode.getDefaultMsg();
     }
 
-    public int getSubscriptionId() {
+    public Map<String, List<TsValue>> getData() {
-        return subscriptionId;
+        if (data == null || data.isEmpty()) {
+            return Collections.emptyMap();
         }
 
-    public Map<String, List<Object>> getData() {
+        Map<String, List<TsValue>> data = new HashMap<>();
+        this.data.forEach((key, entries) -> {
+            if (entries.isEmpty()) {
+                return;
+            }
+
+            List<TsValue> values = new ArrayList<>(entries.size());
+            entries.forEach(object -> {
+                if (!(object instanceof Object[] entry) || entry.length < 2) {
+                    return;
+                }
+                TsValue tsValue = new TsValue((Long) entry[0], (String) entry[1]);
+                values.add(tsValue);
+            });
+            data.put(key, values);
+        });
         return data;
     }
 
@@ -86,18 +111,6 @@ public class TelemetrySubscriptionUpdate {
         }
     }
 
-    public int getErrorCode() {
-        return errorCode;
-    }
-
-    public String getErrorMsg() {
-        return errorMsg;
-    }
-
-    public TelemetrySubscriptionUpdate copyWithNewSubscriptionId(int subscriptionId){
-        return new TelemetrySubscriptionUpdate(subscriptionId, errorCode, errorMsg, data);
-    }
-
     @Override
     public String toString() {
         StringBuilder result = new StringBuilder("TelemetrySubscriptionUpdate [subscriptionId=" + subscriptionId + ", errorCode=" + errorCode + ", errorMsg=" + errorMsg + ", data=");
@@ -110,4 +123,5 @@ public class TelemetrySubscriptionUpdate {
         });
         return result.toString();
     }
+
 }