dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #13465 from thingsboard/fix/telemetry-bug

Browse Source 
Fix WS updates when updating older timeseries
This commit is contained in:
Viacheslav Klimov 2025-05-29 11:51:18 +03:00 committed by GitHub
commit ec9c3d5614
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 272 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java
View File

@ -348,7 +348,7 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
if (sub.isLatestValues()) {
for (TsKvEntry kv : data) {
Long stateTs = keyStates.get(kv.getKey());
if (stateTs == null || kv.getTs() > stateTs) {
if (stateTs == null || kv.getTs() >= stateTs) {
if (updateData == null) {
updateData = new ArrayList<>();
}
@ -362,7 +362,7 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
for (TsKvEntry kv : data) {
Long stateTs = keyStates.get(kv.getKey());
if (stateTs != null) {
if (!sub.isLatestValues() || kv.getTs() > stateTs) {
if (!sub.isLatestValues() || kv.getTs() >= stateTs) {
if (updateData == null) {
updateData = new ArrayList<>();
}

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

@ -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) -> {
Object[] data = (Object[]) v.get(0);
latestUpdate.put(k, new TsValue((Long) data[0], (String) data[1]));
subscriptionUpdate.getValues().forEach((key, values) -> {
latestUpdate.put(key, getLatest(values));
});
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

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

@ -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) -> {
Object[] data = (Object[]) v.get(0);
latestUpdate.put(k, new TsValue((Long) data[0], (String) data[1]));
subscriptionUpdate.getValues().forEach((key, values) -> {
latestUpdate.put(key, getLatest(values));
});
EntityData entityData = entitiesMap.get(entityId);
entityData.getLatest().computeIfAbsent(keyType, tmp -> new HashMap<>()).putAll(latestUpdate);

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

@ -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,35 +93,32 @@ 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((k, v) -> {
Object[] data = (Object[]) v.get(0);
latestUpdate.put(k, new TsValue((Long) data[0], (String) data[1]));
subscriptionUpdate.getValues().forEach((key, values) -> {
latestUpdate.put(key, getLatest(values));
});
EntityData entityData = getDataForEntity(entityId);
if (entityData != null && entityData.getLatest() != null) {
Map<String, TsValue> latestCtxValues = entityData.getLatest().get(keyType);
Map<String, TsValue> latestCtxValues = entityData.getLatest().computeIfAbsent(keyType, __ -> new HashMap<>());
log.trace("[{}][{}][{}] Going to compare update with {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), latestCtxValues);
if (latestCtxValues != null) {
latestCtxValues.forEach((k, v) -> {
TsValue update = latestUpdate.get(k);
if (update != null) {
//Ignore notifications about deleted keys
if (!(update.getTs() == 0 && (update.getValue() == null || update.getValue().isEmpty()))) {
if (update.getTs() < v.getTs()) {
log.trace("[{}][{}][{}] Removed stale update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs());
latestUpdate.remove(k);
} else if ((update.getTs() == v.getTs() && update.getValue().equals(v.getValue()))) {
log.trace("[{}][{}][{}] Removed duplicate update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs());
latestUpdate.remove(k);
}
} else {
log.trace("[{}][{}][{}] Received deleted notification for: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k);
latestCtxValues.forEach((k, v) -> {
TsValue update = latestUpdate.get(k);
if (update != null) {
//Ignore notifications about deleted keys
if (!(update.getTs() == 0 && (update.getValue() == null || update.getValue().isEmpty()))) {
if (update.getTs() < v.getTs()) {
log.trace("[{}][{}][{}] Removed stale update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs());
latestUpdate.remove(k);
} else if ((update.getTs() == v.getTs() && update.getValue().equals(v.getValue()))) {
log.trace("[{}][{}][{}] Removed duplicate update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs());
latestUpdate.remove(k);
}
} else {
log.trace("[{}][{}][{}] Received deleted notification for: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k);
}
});
//Setting new values
latestUpdate.forEach(latestCtxValues::put);
}
}
});
//Setting new values
latestCtxValues.putAll(latestUpdate);
}
if (!latestUpdate.isEmpty()) {
Map<EntityKeyType, Map<String, TsValue>> latestMap = Collections.singletonMap(keyType, latestUpdate);
@ -133,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) -> {
Object[] data = (Object[]) v.get(0);
tsUpdate.computeIfAbsent(k, key -> new ArrayList<>()).add(new TsValue((Long) data[0], (String) data[1]));
subscriptionUpdate.getValues().forEach((key, values) -> {
tsUpdate.put(key, new ArrayList<>(values));
});
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) -> {
List<TsValue> updateList = tsUpdate.get(k);
latestCtxValues.forEach((key, latest) -> {
List<TsValue> updateList = tsUpdate.get(key);
if (updateList != null) {
for (TsValue update : new ArrayList<>(updateList)) {
if (update.getTs() < v.getTs()) {
log.trace("[{}][{}][{}] Removed stale update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs());
if (update.getTs() < latest.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()))) {
log.trace("[{}][{}][{}] Removed duplicate update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs());
} else if ((update.getTs() == latest.getTs() && update.getValue().equals(latest.getValue()))) {
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) -> {
Optional<TsValue> maxValue = v.stream().max(Comparator.comparingLong(TsValue::getTs));
maxValue.ifPresent(max -> latestCtxValues.put(k, max));
tsUpdate.forEach((key, values) -> {
values.stream().max(Comparator.comparingLong(TsValue::getTs))
.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));
}

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

@ -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

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

@ -16,19 +16,26 @@
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;
import java.util.stream.Collectors;
@Getter
@AllArgsConstructor
public class TelemetrySubscriptionUpdate {
@With
private final int subscriptionId;
private int errorCode;
private String errorMsg;
@ -66,11 +73,27 @@ public class TelemetrySubscriptionUpdate {
this.errorMsg = errorMsg != null ? errorMsg : errorCode.getDefaultMsg();
}
public int getSubscriptionId() {
return subscriptionId;
}
public Map<String, List<TsValue>> getValues() {
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,28 +109,17 @@ 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=");
data.forEach((k, v) -> {
result.append(k).append("=[");
for(Object a : v){
result.append(Arrays.toString((Object[])a)).append("|");
for (Object a : v) {
result.append(Arrays.toString((Object[]) a)).append("|");
}
result.append("]");
});
return result.toString();
}
}

79
application/src/test/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionServiceTest.java
View File

@ -32,11 +32,16 @@ import org.thingsboard.server.cache.limits.RateLimitService;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.StringDataEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.data.limit.LimitedApi;
import org.thingsboard.server.common.msg.queue.TbCallback;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.queue.discovery.PartitionService;
import org.thingsboard.server.service.ws.WebSocketSessionRef;
import org.thingsboard.server.service.ws.telemetry.sub.TelemetrySubscriptionUpdate;
import java.util.ArrayList;
import java.util.HashMap;
@ -44,8 +49,13 @@ import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.nullable;
@ -54,11 +64,17 @@ import static org.mockito.Mockito.when;
public class DefaultTbLocalSubscriptionServiceTest {
ListAppender<ILoggingEvent> testLogAppender;
TbLocalSubscriptionService subscriptionService;
private final TenantId tenantId = TenantId.fromUUID(UUID.randomUUID());
private final DeviceId deviceId = new DeviceId(UUID.randomUUID());
private ListAppender<ILoggingEvent> testLogAppender;
private TbLocalSubscriptionService subscriptionService;
private ListeningExecutorService executorService;
@BeforeEach
public void setUp() throws Exception {
this.executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));
Logger logger = (Logger) LoggerFactory.getLogger(DefaultTbLocalSubscriptionService.class);
testLogAppender = new ListAppender<>();
testLogAppender.start();
@ -69,7 +85,9 @@ public class DefaultTbLocalSubscriptionServiceTest {
PartitionService partitionService = mock();
when(partitionService.resolve(any(), any(), any())).thenReturn(TopicPartitionInfo.builder().build());
subscriptionService = new DefaultTbLocalSubscriptionService(mock(), mock(), mock(), partitionService, mock(), mock(), mock(), rateLimitService);
ReflectionTestUtils.setField(subscriptionService, "serviceId", "serviceId");
ReflectionTestUtils.setField(subscriptionService, "subscriptionUpdateExecutor", executorService);
}
@AfterEach
@ -79,27 +97,22 @@ public class DefaultTbLocalSubscriptionServiceTest {
Logger logger = (Logger) LoggerFactory.getLogger(DefaultTbLocalSubscriptionService.class);
logger.detachAppender(testLogAppender);
}
if (executorService != null) {
executorService.shutdownNow();
}
}
@Test
public void addSubscriptionConcurrentModificationTest() throws Exception {
ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));
TenantId tenantId = new TenantId(UUID.randomUUID());
DeviceId deviceId = new DeviceId(UUID.randomUUID());
WebSocketSessionRef sessionRef = mock();
ReflectionTestUtils.setField(subscriptionService, "subscriptionUpdateExecutor", executorService);
List<ListenableFuture<?>> futures = new ArrayList<>();
try {
subscriptionService.onCoreStartupMsg(TransportProtos.CoreStartupMsg.newBuilder().addAllPartitions(List.of(0)).getDefaultInstanceForType());
for (int i = 0; i < 50; i++) {
futures.add(executorService.submit(() -> subscriptionService.addSubscription(createSubscription(tenantId, deviceId), sessionRef)));
}
Futures.allAsList(futures).get();
} finally {
executorService.shutdownNow();
subscriptionService.onCoreStartupMsg(TransportProtos.CoreStartupMsg.newBuilder().addAllPartitions(List.of(0)).getDefaultInstanceForType());
for (int i = 0; i < 50; i++) {
futures.add(executorService.submit(() -> subscriptionService.addSubscription(createSubscription(tenantId, deviceId), sessionRef)));
}
Futures.allAsList(futures).get();
List<ILoggingEvent> logs = testLogAppender.list;
boolean exceptionLogged = logs.stream()
@ -123,4 +136,42 @@ public class DefaultTbLocalSubscriptionServiceTest {
.keyStates(keys)
.build();
}
@Test
public void updateOldTelemetryTest() {
WebSocketSessionRef sessionRef = mock();
String key = "temperature";
long ts = 1237465456L;
Map<String, Long> keyStates = new HashMap<>();
keyStates.put(key, ts);
AtomicReference<TelemetrySubscriptionUpdate> capturedUpdate = new AtomicReference<>();
TbTimeSeriesSubscription tsSubscription = TbTimeSeriesSubscription.builder()
.tenantId(tenantId)
.entityId(deviceId)
.subscriptionId(2)
.sessionId(RandomStringUtils.randomAlphanumeric(5))
.keyStates(keyStates)
.allKeys(true)
.latestValues(true)
.updateProcessor((subscription, update) -> capturedUpdate.set(update))
.build();
subscriptionService.addSubscription(tsSubscription, sessionRef);
// Send telemetry with ts == stateTs
TsKvEntry kv = new BasicTsKvEntry(ts, new StringDataEntry(key, "42"));
subscriptionService.onTimeSeriesUpdate(deviceId, List.of(kv), mock(TbCallback.class));
await().atMost(5, TimeUnit.SECONDS)
.pollInterval(5, TimeUnit.MILLISECONDS)
.untilAsserted(() -> {
TelemetrySubscriptionUpdate update = capturedUpdate.get();
assertNotNull(update);
assertTrue(update.getLatestValues().containsKey(key));
});
}
}

130
application/src/test/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtxTest.java Normal file
View File

@ -0,0 +1,130 @@
/**
* Copyright © 2016-2025 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.service.subscription;
import org.apache.commons.lang3.RandomStringUtils;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.test.util.ReflectionTestUtils;
import org.testcontainers.shaded.org.apache.commons.lang3.RandomUtils;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.LongDataEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.query.EntityData;
import org.thingsboard.server.common.data.query.EntityKeyType;
import org.thingsboard.server.common.data.query.TsValue;
import org.thingsboard.server.service.ws.WebSocketService;
import org.thingsboard.server.service.ws.WebSocketSessionRef;
import org.thingsboard.server.service.ws.telemetry.cmd.v2.CmdUpdate;
import org.thingsboard.server.service.ws.telemetry.cmd.v2.EntityDataUpdate;
import org.thingsboard.server.service.ws.telemetry.sub.TelemetrySubscriptionUpdate;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.BDDMockito.then;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
public class TbEntityDataSubCtxTest {
private final DeviceId deviceId = new DeviceId(UUID.randomUUID());
private final Integer cmdId = RandomUtils.nextInt();
private final Integer subscriptionId = RandomUtils.nextInt();
private final String serviceId = RandomStringUtils.randomAlphanumeric(10);
private final String sessionId = RandomStringUtils.randomAlphanumeric(10);
private final int maxEntitiesPerDataSubscription = 100;
private TbEntityDataSubCtx subCtx;
@Mock
private WebSocketService webSocketService;
@Mock
private WebSocketSessionRef webSocketSessionRef;
@BeforeEach
public void setUp() {
when(webSocketSessionRef.getSessionId()).thenReturn(sessionId);
subCtx = new TbEntityDataSubCtx(serviceId, webSocketService, mock(), mock(), mock(), mock(), webSocketSessionRef, cmdId, maxEntitiesPerDataSubscription);
Map<Integer, EntityId> subToEntityIdMap = new HashMap<>();
subToEntityIdMap.put(subscriptionId, deviceId);
ReflectionTestUtils.setField(subCtx, "subToEntityIdMap", subToEntityIdMap);
long now = System.currentTimeMillis();
long oldTs = now - 1_000_000;
Map<String, TsValue> latestCtxValues = new HashMap<>();
latestCtxValues.put("key", new TsValue(oldTs, "15"));
Map<EntityKeyType, Map<String, TsValue>> latest = new HashMap<>();
latest.put(EntityKeyType.TIME_SERIES, latestCtxValues);
EntityData entityData = new EntityData();
entityData.setEntityId(deviceId);
entityData.setLatest(latest);
PageData<EntityData> data = new PageData<>(List.of(entityData), 1, 1, true);
ReflectionTestUtils.setField(subCtx, "data", data);
}
@Test
public void testSendLatestWsMsg() {
long ts = System.currentTimeMillis();
List<TsKvEntry> telemetry = List.of(
new BasicTsKvEntry(ts - 50000, new LongDataEntry("key", 42L), 34L),
new BasicTsKvEntry(ts - 20000, new LongDataEntry("key", 17L), 78L)
);
TelemetrySubscriptionUpdate subUpdate = new TelemetrySubscriptionUpdate(subscriptionId, telemetry);
subCtx.sendWsMsg(sessionId, subUpdate, EntityKeyType.TIME_SERIES, true);
Map<EntityKeyType, Map<String, TsValue>> expectedLatest = new HashMap<>();
Map<String, TsValue> expectedLatestCtxValues = new HashMap<>();
expectedLatestCtxValues.put("key", new TsValue(ts - 20000, "17")); // use latest telemetry
expectedLatest.put(EntityKeyType.TIME_SERIES, expectedLatestCtxValues);
EntityData expectedEntityData = new EntityData();
expectedEntityData.setEntityId(deviceId);
expectedEntityData.setLatest(expectedLatest);
List<EntityData> expected = List.of(expectedEntityData);
ArgumentCaptor<CmdUpdate> cmdUpdateCaptor = ArgumentCaptor.forClass(CmdUpdate.class);
then(webSocketService).should().sendUpdate(eq(sessionId), cmdUpdateCaptor.capture());
CmdUpdate cmdUpdate = cmdUpdateCaptor.getValue();
assertThat(cmdUpdate).isInstanceOf(EntityDataUpdate.class);
EntityDataUpdate entityDataUpdate = (EntityDataUpdate) cmdUpdate;
assertThat(entityDataUpdate.getCmdId()).isEqualTo(cmdId);
assertThat(entityDataUpdate.getData()).isNull();
assertThat(entityDataUpdate.getUpdate()).isEqualTo(expected);
assertThat(entityDataUpdate.getAllowedEntities()).isEqualTo(maxEntitiesPerDataSubscription);
}
}