diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java index 97b30b4e92..7c66971d27 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java @@ -201,7 +201,7 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene if (serviceId.equals(targetId)) { localSubscriptionService.onTimeSeriesUpdate(entityId, update, TbCallback.EMPTY); } else { - sendCoreNotification(targetId, entityId, TbSubscriptionUtils.toProto(true, entityId, update)); + sendCoreNotification(targetId, entityId, TbSubscriptionUtils.toProto(entityId, update)); } } @@ -213,7 +213,7 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene @Override public void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes, boolean notifyDevice, TbCallback callback) { getEntityUpdatesInfo(entityId).attributesUpdateTs = System.currentTimeMillis(); - processAttributesUpdate(entityId, attributes); + processAttributesUpdate(entityId, scope, attributes); if (entityId.getEntityType() == EntityType.DEVICE) { if (TbAttributeSubscriptionScope.SERVER_SCOPE.name().equalsIgnoreCase(scope)) { updateDeviceInactivityTimeout(tenantId, entityId, attributes); @@ -228,7 +228,7 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene @Override public void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List keys, boolean notifyDevice, TbCallback callback) { - processAttributesUpdate(entityId, + processAttributesUpdate(entityId, scope, keys.stream().map(key -> new BaseAttributeKvEntry(0, new StringDataEntry(key, ""))).collect(Collectors.toList())); if (entityId.getEntityType() == EntityType.DEVICE) { if (TbAttributeSubscriptionScope.SERVER_SCOPE.name().equalsIgnoreCase(scope) @@ -242,17 +242,17 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene callback.onSuccess(); } - public void processAttributesUpdate(EntityId entityId, List update) { + public void processAttributesUpdate(EntityId entityId, String scope, List update) { TbEntityRemoteSubsInfo subInfo = entitySubscriptions.get(entityId); if (subInfo != null) { log.trace("[{}] Handling attributes update: {}", entityId, update); subInfo.getSubs().forEach((serviceId, sub) -> { if (sub.attrAllKeys) { - processAttributesUpdate(serviceId, entityId, update); + processAttributesUpdate(serviceId, entityId, scope, update); } else if (sub.attrKeys != null) { List tmp = getSubList(update, sub.attrKeys); if (tmp != null) { - processAttributesUpdate(serviceId, entityId, tmp); + processAttributesUpdate(serviceId, entityId, scope, tmp); } } }); @@ -261,12 +261,12 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene } } - private void processAttributesUpdate(String targetId, EntityId entityId, List update) { + private void processAttributesUpdate(String targetId, EntityId entityId, String scope, List update) { List tsKvEntryList = update.stream().map(attr -> new BasicTsKvEntry(attr.getLastUpdateTs(), attr)).collect(Collectors.toList()); if (serviceId.equals(targetId)) { - localSubscriptionService.onAttributesUpdate(entityId, tsKvEntryList, TbCallback.EMPTY); + localSubscriptionService.onAttributesUpdate(entityId, scope, tsKvEntryList, TbCallback.EMPTY); } else { - sendCoreNotification(targetId, entityId, TbSubscriptionUtils.toProto(false, entityId, tsKvEntryList)); + sendCoreNotification(targetId, entityId, TbSubscriptionUtils.toProto(scope, entityId, tsKvEntryList)); } } 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 f15a25fe4d..1bb452f4a9 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 @@ -262,37 +262,39 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer @Override public void onAttributesUpdate(TransportProtos.TbSubUpdateProto proto, TbCallback callback) { - onAttributesUpdate(new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB()), TbSubscriptionUtils.fromProto(proto), callback); + onAttributesUpdate(new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB()), proto.getScope(), TbSubscriptionUtils.fromProto(proto), callback); } @Override - public void onAttributesUpdate(EntityId entityId, List data, TbCallback callback) { - onAttributesUpdate(entityId.getId(), data, callback); + public void onAttributesUpdate(EntityId entityId, String scope, List data, TbCallback callback) { + onAttributesUpdate(entityId.getId(), scope, data, callback); } - private void onAttributesUpdate(UUID entityId, List data, TbCallback callback) { + private void onAttributesUpdate(UUID entityId, String scope, List data, TbCallback callback) { entityUpdates.get(entityId).attributesUpdateTs = System.currentTimeMillis(); processSubscriptionData(entityId, sub -> TbSubscriptionType.ATTRIBUTES.equals(sub.getType()), s -> { TbAttributeSubscription sub = (TbAttributeSubscription) s; - List updateData = null; - if (sub.isAllKeys()) { - updateData = data; - } else { - for (TsKvEntry kv : data) { - if (sub.getKeyStates().containsKey((kv.getKey()))) { - if (updateData == null) { - updateData = new ArrayList<>(); + if (sub.getScope() == null || TbAttributeSubscriptionScope.ANY_SCOPE.equals(sub.getScope()) || sub.getScope().name().equals(scope)) { + List updateData = null; + if (sub.isAllKeys()) { + updateData = data; + } else { + for (TsKvEntry kv : data) { + if (sub.getKeyStates().containsKey((kv.getKey()))) { + if (updateData == null) { + updateData = new ArrayList<>(); + } + updateData.add(kv); } - updateData.add(kv); } } - } - if (updateData != null) { - TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(sub.getSubscriptionId(), updateData); - update.getLatestValues().forEach((key, value) -> sub.getKeyStates().put(key, value)); - subscriptionUpdateExecutor.submit(() -> sub.getUpdateProcessor().accept(sub, update)); + if (updateData != null) { + TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(sub.getSubscriptionId(), updateData); + update.getLatestValues().forEach((key, value) -> sub.getKeyStates().put(key, value)); + subscriptionUpdateExecutor.submit(() -> sub.getUpdateProcessor().accept(sub, update)); + } } }, callback); } @@ -444,7 +446,13 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer return; } final Map keyStates = subscription.getKeyStates(); - DonAsynchron.withCallback(attrService.find(subscription.getTenantId(), subscription.getEntityId(), DataConstants.CLIENT_SCOPE, keyStates.keySet()), values -> { + String scope; + if (subscription.getScope() != null && !TbAttributeSubscriptionScope.ANY_SCOPE.equals(subscription.getScope())) { + scope = subscription.getScope().name(); + } else { + scope = DataConstants.CLIENT_SCOPE; + } + DonAsynchron.withCallback(attrService.find(subscription.getTenantId(), subscription.getEntityId(), scope, keyStates.keySet()), values -> { List updates = new ArrayList<>(); values.forEach(latestEntry -> { if (latestEntry.getLastUpdateTs() > keyStates.get(latestEntry.getKey())) { @@ -453,7 +461,7 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer }); var missedUpdates = updates.stream().filter(u -> u.getValue() != null).collect(Collectors.toList()); if (!missedUpdates.isEmpty()) { - onAttributesUpdate(subscription.getEntityId(), missedUpdates, TbCallback.EMPTY); + onAttributesUpdate(subscription.getEntityId(), scope, missedUpdates, TbCallback.EMPTY); } }, e -> log.error("Failed to fetch missed updates.", e), tsCallBackExecutor); diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbLocalSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbLocalSubscriptionService.java index d9fc87f51b..d968bd515f 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbLocalSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbLocalSubscriptionService.java @@ -18,16 +18,12 @@ package org.thingsboard.server.service.subscription; import org.thingsboard.server.common.data.alarm.AlarmInfo; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.discovery.event.ClusterTopologyChangeEvent; -import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; import org.thingsboard.server.service.ws.notification.sub.NotificationRequestUpdate; import org.thingsboard.server.service.ws.notification.sub.NotificationsSubscriptionUpdate; -import org.thingsboard.server.service.ws.telemetry.sub.AlarmSubscriptionUpdate; -import org.thingsboard.server.service.ws.telemetry.sub.TelemetrySubscriptionUpdate; import java.util.List; @@ -49,7 +45,7 @@ public interface TbLocalSubscriptionService { void onAttributesUpdate(TransportProtos.TbSubUpdateProto attrUpdate, TbCallback callback); - void onAttributesUpdate(EntityId entityId, List update, TbCallback callback); + void onAttributesUpdate(EntityId entityId, String scope, List update, TbCallback callback); void onAlarmUpdate(EntityId entityId, AlarmInfo alarm, boolean deleted, TbCallback callback); diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java index 2fd49f0645..1163e13077 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java @@ -411,7 +411,15 @@ public class TbSubscriptionUtils { return result; } - static ToCoreNotificationMsg toProto(boolean timeSeries, EntityId entityId, List updates) { + static ToCoreNotificationMsg toProto(EntityId entityId, List updates) { + return toProto(true, null, entityId, updates); + } + + static ToCoreNotificationMsg toProto(String scope, EntityId entityId, List updates) { + return toProto(false, scope, entityId, updates); + } + + static ToCoreNotificationMsg toProto(boolean timeSeries, String scope, EntityId entityId, List updates) { TransportProtos.TbSubUpdateProto.Builder builder = TransportProtos.TbSubUpdateProto.newBuilder(); builder.setEntityIdMSB(entityId.getId().getMostSignificantBits()); @@ -434,6 +442,7 @@ public class TbSubscriptionUtils { if (timeSeries) { result.setTsUpdate(builder); } else { + builder.setScope(scope); result.setAttrUpdate(builder); } return ToCoreNotificationMsg.newBuilder().setToLocalSubscriptionServiceMsg(result).build(); 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 a8334fc959..73ee8f9252 100644 --- a/application/src/test/java/org/thingsboard/server/controller/WebsocketApiTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/WebsocketApiTest.java @@ -454,6 +454,12 @@ public class WebsocketApiTest extends AbstractControllerTest { getWsClient().registerWaitForUpdate(); + // Pushing update with wrong scope and make sure it will not arrive. + AttributeKvEntry invalidDataPoint = new BaseAttributeKvEntry(now - TimeUnit.MINUTES.toMillis(1), new LongDataEntry("serverAttributeKey", 55L)); + sendAttributes(device, TbAttributeSubscriptionScope.CLIENT_SCOPE, Arrays.asList(invalidDataPoint)); + + Assert.assertNull(getWsClient().waitForUpdate(3000)); + AttributeKvEntry dataPoint1 = new BaseAttributeKvEntry(now - TimeUnit.MINUTES.toMillis(1), new LongDataEntry("serverAttributeKey", 42L)); List tsData = Arrays.asList(dataPoint1); sendAttributes(device, TbAttributeSubscriptionScope.SERVER_SCOPE, tsData); diff --git a/common/cluster-api/src/main/proto/queue.proto b/common/cluster-api/src/main/proto/queue.proto index 95288c541c..ac373df510 100644 --- a/common/cluster-api/src/main/proto/queue.proto +++ b/common/cluster-api/src/main/proto/queue.proto @@ -619,6 +619,7 @@ message TbSubUpdateProto { int32 errorCode = 3; string errorMsg = 4; repeated TsValueListProto data = 5; + string scope = 6; } message TbAlarmSubUpdateProto {