From fabf6f32351193343c20a331f344a147ef56ccc1 Mon Sep 17 00:00:00 2001 From: dashevchenko Date: Tue, 25 Mar 2025 18:28:30 +0200 Subject: [PATCH] fixed telemetry update handling for timeseries subscription --- .../DefaultTbLocalSubscriptionService.java | 8 ++--- .../service/ws/DefaultWebSocketService.java | 3 ++ .../sub/TelemetrySubscriptionUpdate.java | 1 + .../controller/TbTestWebSocketClient.java | 14 +++++++++ .../server/controller/WebsocketApiTest.java | 29 +++++++++++++++++++ 5 files changed, 50 insertions(+), 5 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java index 59681d3daa..165b0824f5 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java @@ -343,11 +343,9 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer s -> { TbTimeSeriesSubscription sub = (TbTimeSeriesSubscription) s; List updateData = null; - if (sub.isAllKeys()) { - updateData = data; - } else { - for (TsKvEntry kv : data) { - if (sub.getKeyStates().containsKey((kv.getKey()))) { + for (TsKvEntry kv : data) { + if (sub.getKeyStates().containsKey((kv.getKey()))) { + if (!sub.isLatestValues() || kv.getTs() > sub.getKeyStates().get(kv.getKey())) { if (updateData == null) { updateData = new ArrayList<>(); } diff --git a/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java b/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java index 7288a8bac9..c52a20754d 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java @@ -123,6 +123,7 @@ public class DefaultWebSocketService implements WebSocketService { private static final String FAILED_TO_FETCH_DATA = "Failed to fetch data!"; private static final String FAILED_TO_FETCH_ATTRIBUTES = "Failed to fetch attributes!"; private static final String SESSION_META_DATA_NOT_FOUND = "Session meta-data not found!"; + private static final String LATEST_TELEMETRY_SCOPE = "LATEST_TELEMETRY"; private final ConcurrentMap wsSessionsMap = new ConcurrentHashMap<>(); @@ -684,6 +685,7 @@ public class DefaultWebSocketService implements WebSocketService { .queryTs(queryTs) .allKeys(true) .keyStates(subState) + .latestValues(LATEST_TELEMETRY_SCOPE.equals(cmd.getScope())) .build(); subLock.lock(); @@ -739,6 +741,7 @@ public class DefaultWebSocketService implements WebSocketService { .queryTs(queryTs) .allKeys(false) .keyStates(subState) + .latestValues(LATEST_TELEMETRY_SCOPE.equals(cmd.getScope())) .build(); subLock.lock(); diff --git a/application/src/main/java/org/thingsboard/server/service/ws/telemetry/sub/TelemetrySubscriptionUpdate.java b/application/src/main/java/org/thingsboard/server/service/ws/telemetry/sub/TelemetrySubscriptionUpdate.java index b22b021a03..1a3046e301 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/telemetry/sub/TelemetrySubscriptionUpdate.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/telemetry/sub/TelemetrySubscriptionUpdate.java @@ -16,6 +16,7 @@ package org.thingsboard.server.service.ws.telemetry.sub; import lombok.AllArgsConstructor; +import net.minidev.json.annotate.JsonIgnore; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.service.subscription.SubscriptionErrorCode; diff --git a/application/src/test/java/org/thingsboard/server/controller/TbTestWebSocketClient.java b/application/src/test/java/org/thingsboard/server/controller/TbTestWebSocketClient.java index 959db43125..1c5b1dd154 100644 --- a/application/src/test/java/org/thingsboard/server/controller/TbTestWebSocketClient.java +++ b/application/src/test/java/org/thingsboard/server/controller/TbTestWebSocketClient.java @@ -31,6 +31,7 @@ import org.thingsboard.server.service.ws.AuthCmd; import org.thingsboard.server.service.ws.WsCmd; import org.thingsboard.server.service.ws.WsCommandsWrapper; import org.thingsboard.server.service.ws.telemetry.cmd.v1.AttributesSubscriptionCmd; +import org.thingsboard.server.service.ws.telemetry.cmd.v1.TimeseriesSubscriptionCmd; import org.thingsboard.server.service.ws.telemetry.cmd.v2.AlarmCountUpdate; import org.thingsboard.server.service.ws.telemetry.cmd.v2.EntityCountUpdate; import org.thingsboard.server.service.ws.telemetry.cmd.v2.EntityDataCmd; @@ -38,6 +39,7 @@ import org.thingsboard.server.service.ws.telemetry.cmd.v2.EntityDataUpdate; import org.thingsboard.server.service.ws.telemetry.cmd.v2.EntityHistoryCmd; import org.thingsboard.server.service.ws.telemetry.cmd.v2.LatestValueCmd; import org.thingsboard.server.service.ws.telemetry.cmd.v2.TimeSeriesCmd; +import org.thingsboard.server.service.ws.telemetry.sub.TelemetrySubscriptionUpdate; import java.net.URI; import java.nio.channels.NotYetConnectedException; @@ -271,6 +273,18 @@ public class TbTestWebSocketClient extends WebSocketClient { return sendEntityDataQuery(edq); } + public JsonNode sendTimeseriesCmd(EntityId entityId, String scope) { + log.warn("sendTimeseriesCmd entityId: {}, scope: {}", entityId, scope); + TimeseriesSubscriptionCmd cmd = new TimeseriesSubscriptionCmd(0, 0, 0, 10, null); + cmd.setEntityId(entityId.getId().toString()); + cmd.setEntityType(entityId.getEntityType().toString()); + cmd.setCmdId(1); + cmd.setScope(scope); + send(cmd); + String msg = this.waitForReply(); + return JacksonUtil.fromString(msg, JsonNode.class); + } + public void send(WsCmd... cmds) { WsCommandsWrapper cmdsWrapper = new WsCommandsWrapper(); cmdsWrapper.setCmds(List.of(cmds)); diff --git a/application/src/test/java/org/thingsboard/server/controller/WebsocketApiTest.java b/application/src/test/java/org/thingsboard/server/controller/WebsocketApiTest.java index 9801907d3b..3842cf917b 100644 --- a/application/src/test/java/org/thingsboard/server/controller/WebsocketApiTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/WebsocketApiTest.java @@ -68,11 +68,13 @@ import org.thingsboard.server.service.ws.telemetry.cmd.v2.AlarmStatusUpdate; import org.thingsboard.server.service.ws.telemetry.cmd.v2.EntityCountCmd; import org.thingsboard.server.service.ws.telemetry.cmd.v2.EntityCountUpdate; import org.thingsboard.server.service.ws.telemetry.cmd.v2.EntityDataUpdate; +import org.thingsboard.server.service.ws.telemetry.sub.TelemetrySubscriptionUpdate; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -584,6 +586,33 @@ public class WebsocketApiTest extends AbstractControllerTest { Assert.assertNull(msg); } + @Test + public void testTimeseriesSubscriptionCmd() throws Exception { + long now = System.currentTimeMillis() - 100; + + long lastTs = now - TimeUnit.MINUTES.toMillis(1); + TsKvEntry dataPoint1 = new BasicTsKvEntry(lastTs, new LongDataEntry("temperature", 42L)); + sendTelemetry(device, List.of(dataPoint1)); + + JsonNode update = getWsClient().sendTimeseriesCmd(device.getId(), "LATEST_TELEMETRY"); + JsonNode data = update.get("data"); + Assert.assertEquals(1, data.size()); + Assert.assertEquals(JacksonUtil.newArrayNode().add(lastTs).add("42"), data.get("temperature").get(0)); + + //Sending update from the past, while latest value has new timestamp; + TsKvEntry dataPoint4 = new BasicTsKvEntry(now - TimeUnit.MINUTES.toMillis(5), new LongDataEntry("temperature", 45L)); + getWsClient().registerWaitForUpdate(); + sendTelemetry(device, List.of(dataPoint4)); + String msg = getWsClient().waitForUpdate(TimeUnit.SECONDS.toMillis(1)); + Assert.assertNull(msg); + + //Sending duplicate update again + getWsClient().registerWaitForUpdate(); + sendTelemetry(device, List.of(dataPoint4)); + msg = getWsClient().waitForUpdate(TimeUnit.SECONDS.toMillis(1)); + Assert.assertNull(msg); + } + @Test public void testEntityDataLatestTsWsCmd() throws Exception { long now = System.currentTimeMillis();