From 9e8489628a6fbea4ee454a9c2d59f5a59266ef01 Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Tue, 23 Jun 2020 12:47:19 +0300 Subject: [PATCH] Latest Attributes Subscription Implementation --- .../subscription/TbEntityDataSubCtx.java | 104 ++++--- .../controller/BaseWebsocketApiTest.java | 289 ++++++++++++++++-- .../dao/sql/query/EntityKeyMapping.java | 2 +- 3 files changed, 338 insertions(+), 57 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java index 403c34449e..f00cd19c0e 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java @@ -84,42 +84,52 @@ public class TbEntityDataSubCtx { public List createSubscriptions(List keys) { this.subToEntityIdMap = new HashMap<>(); tbSubs = new ArrayList<>(); - List attrSubKeys = new ArrayList<>(); - List tsSubKeys = new ArrayList<>(); - for (EntityKey key : keys) { - switch (key.getType()) { - case TIME_SERIES: - tsSubKeys.add(key); - break; - case ATTRIBUTE: - case CLIENT_ATTRIBUTE: - case SHARED_ATTRIBUTE: - case SERVER_ATTRIBUTE: - attrSubKeys.add(key); - } - } + Map> keysByType = new HashMap<>(); + keys.forEach(key -> keysByType.computeIfAbsent(key.getType(), k -> new ArrayList<>()).add(key)); for (EntityData entityData : data.getData()) { - if (!tsSubKeys.isEmpty()) { - tbSubs.add(createTsSub(entityData, tsSubKeys)); - } + keysByType.forEach((keysType, keysList) -> { + int subIdx = sessionRef.getSessionSubIdSeq().incrementAndGet(); + subToEntityIdMap.put(subIdx, entityData.getEntityId()); + switch (keysType) { + case TIME_SERIES: + tbSubs.add(createTsSub(entityData, subIdx, keysList)); + break; + case CLIENT_ATTRIBUTE: + tbSubs.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.CLIENT_SCOPE, keysList)); + break; + case SHARED_ATTRIBUTE: + tbSubs.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.SHARED_SCOPE, keysList)); + break; + case SERVER_ATTRIBUTE: + tbSubs.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.SERVER_SCOPE, keysList)); + break; + case ATTRIBUTE: + tbSubs.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.ANY_SCOPE, keysList)); + break; + } + }); } return tbSubs; } - private TbSubscription createTsSub(EntityData entityData, List tsSubKeys) { - int subIdx = sessionRef.getSessionSubIdSeq().incrementAndGet(); - subToEntityIdMap.put(subIdx, entityData.getEntityId()); - Map keyStates = new HashMap<>(); - tsSubKeys.forEach(key -> keyStates.put(key.getKey(), 0L)); - if (entityData.getLatest() != null) { - Map currentValues = entityData.getLatest().get(EntityKeyType.TIME_SERIES); - if (currentValues != null) { - currentValues.forEach((k, v) -> { - log.trace("[{}][{}] Updating key: {} with ts: {}", serviceId, cmdId, k, v.getTs()); - keyStates.put(k, v.getTs()); - }); - } - } + private TbSubscription createAttrSub(EntityData entityData, int subIdx, EntityKeyType keysType, TbAttributeSubscriptionScope scope, List subKeys) { + Map keyStates = buildKeyStats(entityData, keysType, subKeys); + log.trace("[{}][{}][{}] Creating attributes subscription with keys: {}", serviceId, cmdId, subIdx, keyStates); + return TbAttributeSubscription.builder() + .serviceId(serviceId) + .sessionId(sessionRef.getSessionId()) + .subscriptionId(subIdx) + .tenantId(sessionRef.getSecurityCtx().getTenantId()) + .entityId(entityData.getEntityId()) + .updateConsumer((s, subscriptionUpdate) -> sendWsMsg(s, subscriptionUpdate, keysType)) + .allKeys(false) + .keyStates(keyStates) + .scope(scope) + .build(); + } + + private TbSubscription createTsSub(EntityData entityData, int subIdx, List subKeys) { + Map keyStates = buildKeyStats(entityData, EntityKeyType.TIME_SERIES, subKeys); if (entityData.getTimeseries() != null) { entityData.getTimeseries().forEach((k, v) -> { long ts = Arrays.stream(v).map(TsValue::getTs).max(Long::compareTo).orElse(0L); @@ -127,8 +137,7 @@ public class TbEntityDataSubCtx { keyStates.put(k, ts); }); } - - log.trace("[{}][{}][{}] Creating subscription with keys: {}", serviceId, cmdId, subIdx, keyStates); + log.trace("[{}][{}][{}] Creating time-series subscription with keys: {}", serviceId, cmdId, subIdx, keyStates); return TbTimeseriesSubscription.builder() .serviceId(serviceId) .sessionId(sessionRef.getSessionId()) @@ -137,14 +146,33 @@ public class TbEntityDataSubCtx { .entityId(entityData.getEntityId()) .updateConsumer(this::sendTsWsMsg) .allKeys(false) - .keyStates(keyStates).build(); + .keyStates(keyStates) + .build(); } + private Map buildKeyStats(EntityData entityData, EntityKeyType keysType, List subKeys) { + Map keyStates = new HashMap<>(); + subKeys.forEach(key -> keyStates.put(key.getKey(), 0L)); + if (entityData.getLatest() != null) { + Map currentValues = entityData.getLatest().get(keysType); + if (currentValues != null) { + currentValues.forEach((k, v) -> { + log.trace("[{}][{}] Updating key: {} with ts: {}", serviceId, cmdId, k, v.getTs()); + keyStates.put(k, v.getTs()); + }); + } + } + return keyStates; + } private void sendTsWsMsg(String sessionId, SubscriptionUpdate subscriptionUpdate) { + sendWsMsg(sessionId, subscriptionUpdate, EntityKeyType.TIME_SERIES); + } + + private void sendWsMsg(String sessionId, SubscriptionUpdate subscriptionUpdate, EntityKeyType keyType) { EntityId entityId = subToEntityIdMap.get(subscriptionUpdate.getSubscriptionId()); if (entityId != null) { - log.trace("[{}][{}][{}] Received subscription update: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), subscriptionUpdate); + log.trace("[{}][{}][{}][{}] Received subscription update: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), keyType, subscriptionUpdate); Map latestUpdate = new HashMap<>(); subscriptionUpdate.getData().forEach((k, v) -> { Object[] data = (Object[]) v.get(0); @@ -152,7 +180,7 @@ public class TbEntityDataSubCtx { }); EntityData entityData = getDataForEntity(entityId); if (entityData != null && entityData.getLatest() != null) { - Map latestCtxValues = entityData.getLatest().get(EntityKeyType.TIME_SERIES); + Map latestCtxValues = entityData.getLatest().get(keyType); log.trace("[{}][{}][{}] Going to compare update with {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), latestCtxValues); if (latestCtxValues != null) { latestCtxValues.forEach((k, v) -> { @@ -170,12 +198,12 @@ public class TbEntityDataSubCtx { } } if (!latestUpdate.isEmpty()) { - Map> latestMap = Collections.singletonMap(EntityKeyType.TIME_SERIES, latestUpdate); + Map> latestMap = Collections.singletonMap(keyType, latestUpdate); entityData = new EntityData(entityId, latestMap, null); wsService.sendWsMsg(sessionId, new EntityDataUpdate(cmdId, null, Collections.singletonList(entityData))); } } else { - log.trace("[{}][{}][{}] Received stale subscription update: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), subscriptionUpdate); + log.trace("[{}][{}][{}][{}] Received stale subscription update: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), keyType, subscriptionUpdate); } } diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseWebsocketApiTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseWebsocketApiTest.java index 0f0d550836..ec8ca198bb 100644 --- a/application/src/test/java/org/thingsboard/server/controller/BaseWebsocketApiTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/BaseWebsocketApiTest.java @@ -28,6 +28,8 @@ import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.Tenant; import org.thingsboard.server.common.data.User; import org.thingsboard.server.common.data.kv.Aggregation; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; import org.thingsboard.server.common.data.kv.BasicTsKvEntry; import org.thingsboard.server.common.data.kv.LongDataEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; @@ -41,6 +43,7 @@ import org.thingsboard.server.common.data.query.EntityKeyType; import org.thingsboard.server.common.data.query.TsValue; import org.thingsboard.server.common.data.security.Authority; import org.thingsboard.server.dao.timeseries.TimeseriesService; +import org.thingsboard.server.service.subscription.TbAttributeSubscriptionScope; import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; import org.thingsboard.server.service.telemetry.cmd.TelemetryPluginCmdsWrapper; import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd; @@ -48,6 +51,7 @@ import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate; import org.thingsboard.server.service.telemetry.cmd.v2.EntityHistoryCmd; import org.thingsboard.server.service.telemetry.cmd.v2.LatestValueCmd; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; @@ -139,7 +143,7 @@ public class BaseWebsocketApiTest extends AbstractWebsocketTest { List tsData = Arrays.asList(dataPoint1, dataPoint2, dataPoint3); sendTelemetry(device, tsData); - Thread.sleep(1000); + Thread.sleep(100); wsClient.send(mapper.writeValueAsString(wrapper)); msg = wsClient.waitForReply(); @@ -156,24 +160,8 @@ public class BaseWebsocketApiTest extends AbstractWebsocketTest { Assert.assertEquals(new TsValue(dataPoint3.getTs(), dataPoint3.getValueAsString()), tsArray[2]); } - private void sendTelemetry(Device device, List tsData) throws InterruptedException { - CountDownLatch latch = new CountDownLatch(1); - tsService.saveAndNotify(device.getTenantId(), device.getId(), tsData, 0, new FutureCallback() { - @Override - public void onSuccess(@Nullable Void result) { - latch.countDown(); - } - - @Override - public void onFailure(Throwable t) { - latch.countDown(); - } - }); - latch.await(3, TimeUnit.SECONDS); - } - @Test - public void testEntityDataLatestWsCmd() throws Exception { + public void testEntityDataLatestTsWsCmd() throws Exception { Device device = new Device(); device.setName("Device"); device.setType("default"); @@ -210,6 +198,8 @@ public class BaseWebsocketApiTest extends AbstractWebsocketTest { List tsData = Arrays.asList(dataPoint1); sendTelemetry(device, tsData); + Thread.sleep(100); + cmd = new EntityDataCmd(1, edq, null, latestCmd, null); wrapper = new TelemetryPluginCmdsWrapper(); wrapper.setEntityDataCmds(Collections.singletonList(cmd)); @@ -258,4 +248,267 @@ public class BaseWebsocketApiTest extends AbstractWebsocketTest { Assert.assertNull(msg); } + @Test + public void testEntityDataLatestAttrWsCmd() throws Exception { + Device device = new Device(); + device.setName("Device"); + device.setType("default"); + device.setLabel("testLabel" + (int) (Math.random() * 1000)); + device = doPost("/api/device", device, Device.class); + + long now = System.currentTimeMillis(); + + DeviceTypeFilter dtf = new DeviceTypeFilter(); + dtf.setDeviceNameFilter("D"); + dtf.setDeviceType("default"); + EntityDataQuery edq = new EntityDataQuery(dtf, new EntityDataPageLink(1, 0, null, null), + Collections.emptyList(), Collections.emptyList(), Collections.emptyList()); + + LatestValueCmd latestCmd = new LatestValueCmd(); + latestCmd.setKeys(Collections.singletonList(new EntityKey(EntityKeyType.SERVER_ATTRIBUTE, "serverAttributeKey"))); + EntityDataCmd cmd = new EntityDataCmd(1, edq, null, latestCmd, null); + + TelemetryPluginCmdsWrapper wrapper = new TelemetryPluginCmdsWrapper(); + wrapper.setEntityDataCmds(Collections.singletonList(cmd)); + + wsClient.send(mapper.writeValueAsString(wrapper)); + String msg = wsClient.waitForReply(); + EntityDataUpdate update = mapper.readValue(msg, EntityDataUpdate.class); + Assert.assertEquals(1, update.getCmdId()); + PageData pageData = update.getData(); + Assert.assertNotNull(pageData); + Assert.assertEquals(1, pageData.getData().size()); + Assert.assertEquals(device.getId(), pageData.getData().get(0).getEntityId()); + Assert.assertNotNull(pageData.getData().get(0).getLatest().get(EntityKeyType.SERVER_ATTRIBUTE).get("serverAttributeKey")); + Assert.assertEquals(0, pageData.getData().get(0).getLatest().get(EntityKeyType.SERVER_ATTRIBUTE).get("serverAttributeKey").getTs()); + Assert.assertEquals("", pageData.getData().get(0).getLatest().get(EntityKeyType.SERVER_ATTRIBUTE).get("serverAttributeKey").getValue()); + + AttributeKvEntry dataPoint1 = new BaseAttributeKvEntry(now - TimeUnit.MINUTES.toMillis(1), new LongDataEntry("serverAttributeKey", 42L)); + List tsData = Arrays.asList(dataPoint1); + sendAttributes(device, TbAttributeSubscriptionScope.SERVER_SCOPE, tsData); + + Thread.sleep(100); + + cmd = new EntityDataCmd(1, edq, null, latestCmd, null); + wrapper = new TelemetryPluginCmdsWrapper(); + wrapper.setEntityDataCmds(Collections.singletonList(cmd)); + + wsClient.send(mapper.writeValueAsString(wrapper)); + msg = wsClient.waitForReply(); + update = mapper.readValue(msg, EntityDataUpdate.class); + + Assert.assertEquals(1, update.getCmdId()); + + pageData = update.getData(); + Assert.assertNotNull(pageData); + Assert.assertEquals(1, pageData.getData().size()); + Assert.assertEquals(device.getId(), pageData.getData().get(0).getEntityId()); + Assert.assertNotNull(pageData.getData().get(0).getLatest().get(EntityKeyType.SERVER_ATTRIBUTE)); + TsValue tsValue = pageData.getData().get(0).getLatest().get(EntityKeyType.SERVER_ATTRIBUTE).get("serverAttributeKey"); + Assert.assertEquals(new TsValue(dataPoint1.getLastUpdateTs(), dataPoint1.getValueAsString()), tsValue); + + now = System.currentTimeMillis(); + AttributeKvEntry dataPoint2 = new BaseAttributeKvEntry(now, new LongDataEntry("serverAttributeKey", 52L)); + + wsClient.registerWaitForUpdate(); + sendAttributes(device, TbAttributeSubscriptionScope.SERVER_SCOPE, Arrays.asList(dataPoint2)); + msg = wsClient.waitForUpdate(); + + update = mapper.readValue(msg, EntityDataUpdate.class); + Assert.assertEquals(1, update.getCmdId()); + List eData = update.getUpdate(); + Assert.assertNotNull(eData); + Assert.assertEquals(1, eData.size()); + Assert.assertEquals(device.getId(), eData.get(0).getEntityId()); + Assert.assertNotNull(eData.get(0).getLatest().get(EntityKeyType.SERVER_ATTRIBUTE)); + tsValue = eData.get(0).getLatest().get(EntityKeyType.SERVER_ATTRIBUTE).get("serverAttributeKey"); + Assert.assertEquals(new TsValue(dataPoint2.getLastUpdateTs(), dataPoint2.getValueAsString()), tsValue); + + //Sending update from the past, while latest value has new timestamp; + wsClient.registerWaitForUpdate(); + sendAttributes(device, TbAttributeSubscriptionScope.SERVER_SCOPE, Arrays.asList(dataPoint1)); + msg = wsClient.waitForUpdate(TimeUnit.SECONDS.toMillis(1)); + Assert.assertNull(msg); + + //Sending duplicate update again + wsClient.registerWaitForUpdate(); + sendAttributes(device, TbAttributeSubscriptionScope.SERVER_SCOPE, Arrays.asList(dataPoint2)); + msg = wsClient.waitForUpdate(TimeUnit.SECONDS.toMillis(1)); + Assert.assertNull(msg); + } + + @Test + public void testEntityDataLatestAttrTypesWsCmd() throws Exception { + Device device = new Device(); + device.setName("Device"); + device.setType("default"); + device.setLabel("testLabel" + (int) (Math.random() * 1000)); + device = doPost("/api/device", device, Device.class); + + long now = System.currentTimeMillis(); + + DeviceTypeFilter dtf = new DeviceTypeFilter(); + dtf.setDeviceNameFilter("D"); + dtf.setDeviceType("default"); + EntityDataQuery edq = new EntityDataQuery(dtf, new EntityDataPageLink(1, 0, null, null), + Collections.emptyList(), Collections.emptyList(), Collections.emptyList()); + + LatestValueCmd latestCmd = new LatestValueCmd(); + List keys = new ArrayList<>(); + keys.add(new EntityKey(EntityKeyType.SERVER_ATTRIBUTE, "serverAttributeKey")); + keys.add(new EntityKey(EntityKeyType.CLIENT_ATTRIBUTE, "clientAttributeKey")); + keys.add(new EntityKey(EntityKeyType.SHARED_ATTRIBUTE, "sharedAttributeKey")); + keys.add(new EntityKey(EntityKeyType.ATTRIBUTE, "anyAttributeKey")); + latestCmd.setKeys(keys); + EntityDataCmd cmd = new EntityDataCmd(1, edq, null, latestCmd, null); + + TelemetryPluginCmdsWrapper wrapper = new TelemetryPluginCmdsWrapper(); + wrapper.setEntityDataCmds(Collections.singletonList(cmd)); + + wsClient.send(mapper.writeValueAsString(wrapper)); + String msg = wsClient.waitForReply(); + EntityDataUpdate update = mapper.readValue(msg, EntityDataUpdate.class); + Assert.assertEquals(1, update.getCmdId()); + PageData pageData = update.getData(); + Assert.assertNotNull(pageData); + Assert.assertEquals(1, pageData.getData().size()); + Assert.assertEquals(device.getId(), pageData.getData().get(0).getEntityId()); + Assert.assertNotNull(pageData.getData().get(0).getLatest().get(EntityKeyType.SERVER_ATTRIBUTE).get("serverAttributeKey")); + Assert.assertEquals(0, pageData.getData().get(0).getLatest().get(EntityKeyType.SERVER_ATTRIBUTE).get("serverAttributeKey").getTs()); + Assert.assertEquals("", pageData.getData().get(0).getLatest().get(EntityKeyType.SERVER_ATTRIBUTE).get("serverAttributeKey").getValue()); + Assert.assertNotNull(pageData.getData().get(0).getLatest().get(EntityKeyType.CLIENT_ATTRIBUTE).get("clientAttributeKey")); + Assert.assertEquals(0, pageData.getData().get(0).getLatest().get(EntityKeyType.CLIENT_ATTRIBUTE).get("clientAttributeKey").getTs()); + Assert.assertEquals("", pageData.getData().get(0).getLatest().get(EntityKeyType.CLIENT_ATTRIBUTE).get("clientAttributeKey").getValue()); + Assert.assertNotNull(pageData.getData().get(0).getLatest().get(EntityKeyType.SHARED_ATTRIBUTE).get("sharedAttributeKey")); + Assert.assertEquals(0, pageData.getData().get(0).getLatest().get(EntityKeyType.SHARED_ATTRIBUTE).get("sharedAttributeKey").getTs()); + Assert.assertEquals("", pageData.getData().get(0).getLatest().get(EntityKeyType.SHARED_ATTRIBUTE).get("sharedAttributeKey").getValue()); + Assert.assertNotNull(pageData.getData().get(0).getLatest().get(EntityKeyType.ATTRIBUTE).get("anyAttributeKey")); + Assert.assertEquals(0, pageData.getData().get(0).getLatest().get(EntityKeyType.ATTRIBUTE).get("anyAttributeKey").getTs()); + Assert.assertEquals("", pageData.getData().get(0).getLatest().get(EntityKeyType.ATTRIBUTE).get("anyAttributeKey").getValue()); + + + wsClient.registerWaitForUpdate(); + AttributeKvEntry dataPoint1 = new BaseAttributeKvEntry(now - TimeUnit.MINUTES.toMillis(1), new LongDataEntry("serverAttributeKey", 42L)); + List tsData = Arrays.asList(dataPoint1); + sendAttributes(device, TbAttributeSubscriptionScope.SERVER_SCOPE, tsData); + + Thread.sleep(100); + + cmd = new EntityDataCmd(1, edq, null, latestCmd, null); + wrapper = new TelemetryPluginCmdsWrapper(); + wrapper.setEntityDataCmds(Collections.singletonList(cmd)); + + msg = wsClient.waitForUpdate(); + + update = mapper.readValue(msg, EntityDataUpdate.class); + Assert.assertEquals(1, update.getCmdId()); + List eData = update.getUpdate(); + Assert.assertNotNull(eData); + Assert.assertEquals(1, eData.size()); + Assert.assertEquals(device.getId(), eData.get(0).getEntityId()); + Assert.assertNotNull(eData.get(0).getLatest().get(EntityKeyType.SERVER_ATTRIBUTE)); + TsValue attrValue = eData.get(0).getLatest().get(EntityKeyType.SERVER_ATTRIBUTE).get("serverAttributeKey"); + Assert.assertEquals(new TsValue(dataPoint1.getLastUpdateTs(), dataPoint1.getValueAsString()), attrValue); + + //Sending update from the past, while latest value has new timestamp; + wsClient.registerWaitForUpdate(); + sendAttributes(device, TbAttributeSubscriptionScope.SHARED_SCOPE, Arrays.asList(dataPoint1)); + msg = wsClient.waitForUpdate(TimeUnit.SECONDS.toMillis(1)); + Assert.assertNull(msg); + + //Sending duplicate update again + wsClient.registerWaitForUpdate(); + sendAttributes(device, TbAttributeSubscriptionScope.CLIENT_SCOPE, Arrays.asList(dataPoint1)); + msg = wsClient.waitForUpdate(TimeUnit.SECONDS.toMillis(1)); + Assert.assertNull(msg); + + //Sending update from the past, while latest value has new timestamp; + wsClient.registerWaitForUpdate(); + AttributeKvEntry dataPoint2 = new BaseAttributeKvEntry(now, new LongDataEntry("sharedAttributeKey", 42L)); + sendAttributes(device, TbAttributeSubscriptionScope.SHARED_SCOPE, Arrays.asList(dataPoint2)); + msg = wsClient.waitForUpdate(TimeUnit.SECONDS.toMillis(1)); + update = mapper.readValue(msg, EntityDataUpdate.class); + Assert.assertEquals(1, update.getCmdId()); + eData = update.getUpdate(); + Assert.assertNotNull(eData); + Assert.assertEquals(1, eData.size()); + Assert.assertEquals(device.getId(), eData.get(0).getEntityId()); + Assert.assertNotNull(eData.get(0).getLatest().get(EntityKeyType.SHARED_ATTRIBUTE)); + attrValue = eData.get(0).getLatest().get(EntityKeyType.SHARED_ATTRIBUTE).get("sharedAttributeKey"); + Assert.assertEquals(new TsValue(dataPoint2.getLastUpdateTs(), dataPoint2.getValueAsString()), attrValue); + + wsClient.registerWaitForUpdate(); + AttributeKvEntry dataPoint3 = new BaseAttributeKvEntry(now, new LongDataEntry("clientAttributeKey", 42L)); + sendAttributes(device, TbAttributeSubscriptionScope.CLIENT_SCOPE, Arrays.asList(dataPoint3)); + msg = wsClient.waitForUpdate(TimeUnit.SECONDS.toMillis(1)); + update = mapper.readValue(msg, EntityDataUpdate.class); + Assert.assertEquals(1, update.getCmdId()); + eData = update.getUpdate(); + Assert.assertNotNull(eData); + Assert.assertEquals(1, eData.size()); + Assert.assertEquals(device.getId(), eData.get(0).getEntityId()); + Assert.assertNotNull(eData.get(0).getLatest().get(EntityKeyType.CLIENT_ATTRIBUTE)); + attrValue = eData.get(0).getLatest().get(EntityKeyType.CLIENT_ATTRIBUTE).get("clientAttributeKey"); + Assert.assertEquals(new TsValue(dataPoint3.getLastUpdateTs(), dataPoint3.getValueAsString()), attrValue); + + wsClient.registerWaitForUpdate(); + AttributeKvEntry dataPoint4 = new BaseAttributeKvEntry(now, new LongDataEntry("anyAttributeKey", 42L)); + sendAttributes(device, TbAttributeSubscriptionScope.CLIENT_SCOPE, Arrays.asList(dataPoint4)); + msg = wsClient.waitForUpdate(TimeUnit.SECONDS.toMillis(1)); + update = mapper.readValue(msg, EntityDataUpdate.class); + Assert.assertEquals(1, update.getCmdId()); + eData = update.getUpdate(); + Assert.assertNotNull(eData); + Assert.assertEquals(1, eData.size()); + Assert.assertEquals(device.getId(), eData.get(0).getEntityId()); + Assert.assertNotNull(eData.get(0).getLatest().get(EntityKeyType.ATTRIBUTE)); + attrValue = eData.get(0).getLatest().get(EntityKeyType.ATTRIBUTE).get("anyAttributeKey"); + Assert.assertEquals(new TsValue(dataPoint4.getLastUpdateTs(), dataPoint4.getValueAsString()), attrValue); + + wsClient.registerWaitForUpdate(); + AttributeKvEntry dataPoint5 = new BaseAttributeKvEntry(now, new LongDataEntry("anyAttributeKey", 43L)); + sendAttributes(device, TbAttributeSubscriptionScope.SERVER_SCOPE, Arrays.asList(dataPoint5)); + msg = wsClient.waitForUpdate(TimeUnit.SECONDS.toMillis(1)); + update = mapper.readValue(msg, EntityDataUpdate.class); + Assert.assertEquals(1, update.getCmdId()); + eData = update.getUpdate(); + Assert.assertNotNull(eData); + Assert.assertEquals(1, eData.size()); + Assert.assertEquals(device.getId(), eData.get(0).getEntityId()); + Assert.assertNotNull(eData.get(0).getLatest().get(EntityKeyType.ATTRIBUTE)); + attrValue = eData.get(0).getLatest().get(EntityKeyType.ATTRIBUTE).get("anyAttributeKey"); + Assert.assertEquals(new TsValue(dataPoint5.getLastUpdateTs(), dataPoint5.getValueAsString()), attrValue); + } + + private void sendTelemetry(Device device, List tsData) throws InterruptedException { + CountDownLatch latch = new CountDownLatch(1); + tsService.saveAndNotify(device.getTenantId(), device.getId(), tsData, 0, new FutureCallback() { + @Override + public void onSuccess(@Nullable Void result) { + latch.countDown(); + } + + @Override + public void onFailure(Throwable t) { + latch.countDown(); + } + }); + latch.await(3, TimeUnit.SECONDS); + } + + private void sendAttributes(Device device, TbAttributeSubscriptionScope scope, List attrData) throws InterruptedException { + CountDownLatch latch = new CountDownLatch(1); + tsService.saveAndNotify(device.getTenantId(), device.getId(), scope.name(), attrData, new FutureCallback() { + @Override + public void onSuccess(@Nullable Void result) { + latch.countDown(); + } + + @Override + public void onFailure(Throwable t) { + latch.countDown(); + } + }); + latch.await(3, TimeUnit.SECONDS); + } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/query/EntityKeyMapping.java b/dao/src/main/java/org/thingsboard/server/dao/sql/query/EntityKeyMapping.java index b7cb331237..0b2c74a25c 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/query/EntityKeyMapping.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/query/EntityKeyMapping.java @@ -136,7 +136,7 @@ public class EntityKeyMapping { } else { scope = DataConstants.SERVER_SCOPE; } - query = String.format("%s AND %s.attribute_type=%s", query, alias, scope); + query = String.format("%s AND %s.attribute_type='%s'", query, alias, scope); } return query; }