diff --git a/application/src/main/data/json/edge/rule_chains/edge_root_rule_chain.json b/application/src/main/data/json/edge/rule_chains/edge_root_rule_chain.json index ef0bf2698f..81f9e6a14d 100644 --- a/application/src/main/data/json/edge/rule_chains/edge_root_rule_chain.json +++ b/application/src/main/data/json/edge/rule_chains/edge_root_rule_chain.json @@ -50,8 +50,11 @@ }, "type": "org.thingsboard.rule.engine.telemetry.TbMsgAttributesNode", "name": "Save Client Attributes", - "configurationVersion": 2, + "configurationVersion": 3, "configuration": { + "processingSettings": { + "type": "ON_EVERY_MESSAGE" + }, "scope": "CLIENT_SCOPE", "notifyDevice": false, "sendAttributesUpdatedNotification": false, diff --git a/application/src/main/data/json/tenant/device_profile/rule_chain_template.json b/application/src/main/data/json/tenant/device_profile/rule_chain_template.json index 0f2473cde6..305dc04961 100644 --- a/application/src/main/data/json/tenant/device_profile/rule_chain_template.json +++ b/application/src/main/data/json/tenant/device_profile/rule_chain_template.json @@ -35,8 +35,11 @@ }, "type": "org.thingsboard.rule.engine.telemetry.TbMsgAttributesNode", "name": "Save Client Attributes", - "configurationVersion": 2, + "configurationVersion": 3, "configuration": { + "processingSettings": { + "type": "ON_EVERY_MESSAGE" + }, "scope": "CLIENT_SCOPE", "notifyDevice": false, "sendAttributesUpdatedNotification": false, diff --git a/application/src/main/data/json/tenant/rule_chains/root_rule_chain.json b/application/src/main/data/json/tenant/rule_chains/root_rule_chain.json index 8efda98c5b..a988c9d5eb 100644 --- a/application/src/main/data/json/tenant/rule_chains/root_rule_chain.json +++ b/application/src/main/data/json/tenant/rule_chains/root_rule_chain.json @@ -34,8 +34,11 @@ }, "type": "org.thingsboard.rule.engine.telemetry.TbMsgAttributesNode", "name": "Save Client Attributes", - "configurationVersion": 2, + "configurationVersion": 3, "configuration": { + "processingSettings": { + "type": "ON_EVERY_MESSAGE" + }, "scope": "CLIENT_SCOPE", "notifyDevice": false, "sendAttributesUpdatedNotification": false, diff --git a/application/src/main/data/upgrade/basic/schema_update.sql b/application/src/main/data/upgrade/basic/schema_update.sql index e9cbf8ef8e..29c7a084f4 100644 --- a/application/src/main/data/upgrade/basic/schema_update.sql +++ b/application/src/main/data/upgrade/basic/schema_update.sql @@ -16,52 +16,54 @@ -- UPDATE SAVE TIME SERIES NODES START -DO $$ - BEGIN - -- Check if the rule_node table exists - IF EXISTS ( - SELECT 1 - FROM information_schema.tables - WHERE table_name = 'rule_node' - ) THEN +UPDATE rule_node +SET configuration = ( + (configuration::jsonb - 'skipLatestPersistence') + || jsonb_build_object( + 'processingSettings', jsonb_build_object( + 'type', 'ADVANCED', + 'timeseries', jsonb_build_object('type', 'ON_EVERY_MESSAGE'), + 'latest', jsonb_build_object('type', 'SKIP'), + 'webSockets', jsonb_build_object('type', 'ON_EVERY_MESSAGE'), + 'calculatedFields', jsonb_build_object('type', 'ON_EVERY_MESSAGE') + ) + ) + )::text, + configuration_version = 1 +WHERE type = 'org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNode' + AND configuration_version = 0 + AND configuration::jsonb ->> 'skipLatestPersistence' = 'true'; - UPDATE rule_node - SET configuration = ( - (configuration::jsonb - 'skipLatestPersistence') - || jsonb_build_object( - 'processingSettings', jsonb_build_object( - 'type', 'ADVANCED', - 'timeseries', jsonb_build_object('type', 'ON_EVERY_MESSAGE'), - 'latest', jsonb_build_object('type', 'SKIP'), - 'webSockets', jsonb_build_object('type', 'ON_EVERY_MESSAGE'), - 'calculatedFields', jsonb_build_object('type', 'ON_EVERY_MESSAGE') - ) - ) - )::text, - configuration_version = 1 - WHERE type = 'org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNode' - AND configuration_version = 0 - AND configuration::jsonb ->> 'skipLatestPersistence' = 'true'; +UPDATE rule_node +SET configuration = ( + (configuration::jsonb - 'skipLatestPersistence') + || jsonb_build_object( + 'processingSettings', jsonb_build_object( + 'type', 'ON_EVERY_MESSAGE' + ) + ) + )::text, + configuration_version = 1 +WHERE type = 'org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNode' + AND configuration_version = 0 + AND (configuration::jsonb ->> 'skipLatestPersistence' != 'true' OR configuration::jsonb ->> 'skipLatestPersistence' IS NULL); - UPDATE rule_node - SET configuration = ( - (configuration::jsonb - 'skipLatestPersistence') - || jsonb_build_object( - 'processingSettings', jsonb_build_object( - 'type', 'ON_EVERY_MESSAGE' - ) - ) - )::text, - configuration_version = 1 - WHERE type = 'org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNode' - AND configuration_version = 0 - AND (configuration::jsonb ->> 'skipLatestPersistence' != 'true' OR configuration::jsonb ->> 'skipLatestPersistence' IS NULL); +-- UPDATE SAVE TIME SERIES NODES END - END IF; - END; -$$; +-- UPDATE SAVE ATTRIBUTES NODES START --- UPDATE SAVE TIME SERIES NODES END +UPDATE rule_node +SET configuration = ( + configuration::jsonb + || jsonb_build_object( + 'processingSettings', jsonb_build_object('type', 'ON_EVERY_MESSAGE') + ) + )::text, + configuration_version = 3 +WHERE type = 'org.thingsboard.rule.engine.telemetry.TbMsgAttributesNode' + AND configuration_version = 2; + +-- UPDATE SAVE ATTRIBUTES NODES END ALTER TABLE api_usage_state ADD COLUMN IF NOT EXISTS version BIGINT DEFAULT 1; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index 50989ea5a0..a3003ba6ff 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -552,10 +552,19 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService update) { + private void onTimeSeriesUpdate(EntityId entityId, List update) { getEntityUpdatesInfo(entityId).timeSeriesUpdateTs = System.currentTimeMillis(); TbEntityRemoteSubsInfo subInfo = entitySubscriptions.get(entityId); if (subInfo != null) { @@ -201,42 +197,27 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene @Override public void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes, TbCallback callback) { - onAttributesUpdate(tenantId, entityId, scope, attributes, true, callback); - } - - @Override - public void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes, boolean notifyDevice, TbCallback callback) { getEntityUpdatesInfo(entityId).attributesUpdateTs = System.currentTimeMillis(); processAttributesUpdate(entityId, scope, attributes); - if (entityId.getEntityType() == EntityType.DEVICE) { - if (TbAttributeSubscriptionScope.SERVER_SCOPE.name().equalsIgnoreCase(scope)) { - updateDeviceInactivityTimeout(tenantId, entityId, attributes); - } else if (TbAttributeSubscriptionScope.SHARED_SCOPE.name().equalsIgnoreCase(scope) && notifyDevice) { - clusterService.pushMsgToCore(DeviceAttributesEventNotificationMsg.onUpdate(tenantId, - new DeviceId(entityId.getId()), DataConstants.SHARED_SCOPE, new ArrayList<>(attributes)) - , null); - } - } callback.onSuccess(); } + @Override + public void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List keys, TbCallback callback) { + onAttributesDelete(tenantId, entityId, scope, keys, false, callback); + } + @Override public void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List keys, boolean notifyDevice, TbCallback callback) { 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) - || TbAttributeSubscriptionScope.ANY_SCOPE.name().equalsIgnoreCase(scope)) { - deleteDeviceInactivityTimeout(tenantId, entityId, keys); - } else if (TbAttributeSubscriptionScope.SHARED_SCOPE.name().equalsIgnoreCase(scope) && notifyDevice) { - clusterService.pushMsgToCore(DeviceAttributesEventNotificationMsg.onDelete(tenantId, - new DeviceId(entityId.getId()), scope, keys), null); - } + if (entityId.getEntityType() == EntityType.DEVICE && TbAttributeSubscriptionScope.SHARED_SCOPE.name().equalsIgnoreCase(scope) && notifyDevice) { + clusterService.pushMsgToCore(DeviceAttributesEventNotificationMsg.onDelete(tenantId, new DeviceId(entityId.getId()), scope, keys), null); } callback.onSuccess(); } - public void processAttributesUpdate(EntityId entityId, String scope, List update) { + private void processAttributesUpdate(EntityId entityId, String scope, List update) { TbEntityRemoteSubsInfo subInfo = entitySubscriptions.get(entityId); if (subInfo != null) { log.trace("[{}] Handling attributes update: {}", entityId, update); @@ -264,22 +245,6 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene } } - private void updateDeviceInactivityTimeout(TenantId tenantId, EntityId entityId, List kvEntries) { - for (KvEntry kvEntry : kvEntries) { - if (kvEntry.getKey().equals(DefaultDeviceStateService.INACTIVITY_TIMEOUT)) { - deviceStateService.onDeviceInactivityTimeoutUpdate(tenantId, new DeviceId(entityId.getId()), getLongValue(kvEntry)); - } - } - } - - private void deleteDeviceInactivityTimeout(TenantId tenantId, EntityId entityId, List keys) { - for (String key : keys) { - if (key.equals(DefaultDeviceStateService.INACTIVITY_TIMEOUT)) { - deviceStateService.onDeviceInactivityTimeoutUpdate(tenantId, new DeviceId(entityId.getId()), 0); - } - } - } - @Override public void onAlarmUpdate(TenantId tenantId, EntityId entityId, AlarmInfo alarm, TbCallback callback) { onAlarmSubUpdate(tenantId, entityId, alarm, false, callback); @@ -349,29 +314,6 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene } } - private static long getLongValue(KvEntry kve) { - switch (kve.getDataType()) { - case LONG: - return kve.getLongValue().orElse(0L); - case DOUBLE: - return kve.getDoubleValue().orElse(0.0).longValue(); - case STRING: - try { - return Long.parseLong(kve.getStrValue().orElse("0")); - } catch (NumberFormatException e) { - return 0L; - } - case JSON: - try { - return Long.parseLong(kve.getJsonValue().orElse("0")); - } catch (NumberFormatException e) { - return 0L; - } - default: - return 0L; - } - } - private static List getSubList(List ts, Set keys) { List update = null; for (T entry : ts) { diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java b/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java index d199f18b75..57d4fda5f8 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java @@ -39,8 +39,15 @@ public interface SubscriptionManagerService extends ApplicationListener attributes, TbCallback callback); - void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes, boolean notifyDevice, TbCallback callback); - + void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List keys, TbCallback empty); + + /** + * This method is retained solely for backwards compatibility, specifically to handle + * legacy proto messages that include the notifyDevice field. + * + * @deprecated as of 4.0, this method will be removed in future releases. + */ + @Deprecated(forRemoval = true, since = "4.0") void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List keys, boolean notifyDevice, TbCallback empty); void onTimeSeriesDelete(TenantId tenantId, EntityId entityId, List keys, 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 68c0f52f71..1d5e85cc22 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 @@ -209,7 +209,7 @@ public class TbSubscriptionUtils { return ToCoreMsg.newBuilder().setToSubscriptionMgrMsg(msgBuilder.build()).build(); } - public static ToCoreMsg toAttributesDeleteProto(TenantId tenantId, EntityId entityId, String scope, List keys, boolean notifyDevice) { + public static ToCoreMsg toAttributesDeleteProto(TenantId tenantId, EntityId entityId, String scope, List keys) { TbAttributeDeleteProto.Builder builder = TbAttributeDeleteProto.newBuilder(); builder.setEntityType(entityId.getEntityType().name()); builder.setEntityIdMSB(entityId.getId().getMostSignificantBits()); @@ -218,7 +218,6 @@ public class TbSubscriptionUtils { builder.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()); builder.setScope(scope); builder.addAllKeys(keys); - builder.setNotifyDevice(notifyDevice); SubscriptionMgrMsgProto.Builder msgBuilder = SubscriptionMgrMsgProto.newBuilder(); msgBuilder.setAttrDelete(builder); diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java index 7225156008..d6fa89a7a2 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java @@ -111,8 +111,7 @@ public abstract class AbstractSubscriptionService extends TbApplicationEventList } @Override - public void onFailure(Throwable t) { - } + public void onFailure(Throwable t) {} }, executor); } diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java index 36cd5516a8..2302446b6b 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java @@ -31,20 +31,26 @@ import org.thingsboard.common.util.DonAsynchron; import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.rule.engine.api.AttributesDeleteRequest; import org.thingsboard.rule.engine.api.AttributesSaveRequest; +import org.thingsboard.rule.engine.api.DeviceStateManager; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.TimeseriesDeleteRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.common.data.ApiUsageRecordKey; +import org.thingsboard.server.common.data.AttributeScope; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityView; import org.thingsboard.server.common.data.id.CustomerId; +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.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.TimeseriesSaveResult; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.kv.TsKvLatestRemovingResult; import org.thingsboard.server.common.msg.queue.TbCallback; +import org.thingsboard.server.common.msg.rule.engine.DeviceAttributesEventNotificationMsg; import org.thingsboard.server.common.stats.TbApiUsageReportClient; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.timeseries.TimeseriesService; @@ -52,10 +58,11 @@ import org.thingsboard.server.dao.util.KvUtils; import org.thingsboard.server.service.apiusage.TbApiUsageStateService; import org.thingsboard.server.service.cf.CalculatedFieldQueueService; import org.thingsboard.server.service.entitiy.entityview.TbEntityViewService; +import org.thingsboard.server.service.state.DefaultDeviceStateService; import org.thingsboard.server.service.subscription.TbSubscriptionUtils; import java.util.ArrayList; -import java.util.Comparator; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -65,6 +72,11 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.function.Consumer; +import static java.util.Comparator.comparing; +import static java.util.Comparator.comparingLong; +import static java.util.Comparator.naturalOrder; +import static java.util.Comparator.nullsFirst; + /** * Created by ashvayka on 27.03.18. */ @@ -78,6 +90,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer private final TbApiUsageReportClient apiUsageClient; private final TbApiUsageStateService apiUsageStateService; private final CalculatedFieldQueueService calculatedFieldQueueService; + private final DeviceStateManager deviceStateManager; private ExecutorService tsCallBackExecutor; @@ -89,13 +102,15 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer @Lazy TbEntityViewService tbEntityViewService, TbApiUsageReportClient apiUsageClient, TbApiUsageStateService apiUsageStateService, - CalculatedFieldQueueService calculatedFieldQueueService) { + CalculatedFieldQueueService calculatedFieldQueueService, + DeviceStateManager deviceStateManager) { this.attrService = attrService; this.tsService = tsService; this.tbEntityViewService = tbEntityViewService; this.apiUsageClient = apiUsageClient; this.apiUsageStateService = apiUsageStateService; this.calculatedFieldQueueService = calculatedFieldQueueService; + this.deviceStateManager = deviceStateManager; } @PostConstruct @@ -140,6 +155,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer EntityId entityId = request.getEntityId(); TimeseriesSaveRequest.Strategy strategy = request.getStrategy(); ListenableFuture resultFuture; + if (strategy.saveTimeseries() && strategy.saveLatest()) { resultFuture = tsService.save(tenantId, entityId, request.getEntries(), request.getTtl()); } else if (strategy.saveLatest()) { @@ -176,11 +192,68 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer @Override public void saveAttributesInternal(AttributesSaveRequest request) { log.trace("Executing saveInternal [{}]", request); - ListenableFuture> saveFuture = attrService.save(request.getTenantId(), request.getEntityId(), request.getScope(), request.getEntries()); - DonAsynchron.withCallback(saveFuture, result -> { - calculatedFieldQueueService.pushRequestToQueue(request, result, request.getCallback()); - }, safeCallback(request.getCallback()), tsCallBackExecutor); - addWsCallback(saveFuture, success -> onAttributesUpdate(request.getTenantId(), request.getEntityId(), request.getScope().name(), request.getEntries(), request.isNotifyDevice())); + TenantId tenantId = request.getTenantId(); + EntityId entityId = request.getEntityId(); + AttributesSaveRequest.Strategy strategy = request.getStrategy(); + ListenableFuture> resultFuture; + + if (strategy.saveAttributes()) { + resultFuture = attrService.save(tenantId, entityId, request.getScope(), request.getEntries()); + } else { + resultFuture = Futures.immediateFuture(Collections.emptyList()); + } + + addMainCallback(resultFuture, result -> { + if (strategy.processCalculatedFields()) { + calculatedFieldQueueService.pushRequestToQueue(request, result, request.getCallback()); + } else { + request.getCallback().onSuccess(null); + } + }, t -> request.getCallback().onFailure(t)); + + if (shouldSendSharedAttributesUpdatedNotification(request)) { + addMainCallback(resultFuture, success -> clusterService.pushMsgToCore( + DeviceAttributesEventNotificationMsg.onUpdate(tenantId, new DeviceId(entityId.getId()), DataConstants.SHARED_SCOPE, request.getEntries()), null + )); + } + + if (shouldCheckForInactivityTimeoutUpdates(request)) { + findNewInactivityTimeout(request.getEntries()).ifPresent(newInactivityTimeout -> + addMainCallback(resultFuture, success -> deviceStateManager.onDeviceInactivityTimeoutUpdate( + tenantId, new DeviceId(entityId.getId()), newInactivityTimeout, TbCallback.EMPTY) + ) + ); + } + + if (strategy.sendWsUpdate()) { + addWsCallback(resultFuture, success -> onAttributesUpdate(tenantId, entityId, request.getScope().name(), request.getEntries())); + } + } + + private static boolean shouldSendSharedAttributesUpdatedNotification(AttributesSaveRequest request) { + return request.getStrategy().saveAttributes() && shouldSendSharedAttributesNotification(request.getEntityId(), request.getScope(), request.isNotifyDevice()); + } + + private static boolean shouldCheckForInactivityTimeoutUpdates(AttributesSaveRequest request) { + return request.getStrategy().saveAttributes() + && request.getEntityId().getEntityType() == EntityType.DEVICE + && request.getScope() == AttributeScope.SERVER_SCOPE; + } + + private static Optional findNewInactivityTimeout(List entries) { + return entries.stream() + .filter(entry -> Objects.equals(DefaultDeviceStateService.INACTIVITY_TIMEOUT, entry.getKey())) + // Select the entry with the highest version, or if the versions are equal, the one with the most recent update timestamp + .max(comparing(AttributeKvEntry::getVersion, nullsFirst(naturalOrder())).thenComparingLong(AttributeKvEntry::getLastUpdateTs)) + .map(DefaultTelemetrySubscriptionService::parseAsLong); + } + + private static long parseAsLong(KvEntry kve) { + try { + return Long.parseLong(kve.getValueAsString()); + } catch (NumberFormatException e) { + return 0L; + } } @Override @@ -191,11 +264,45 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer @Override public void deleteAttributesInternal(AttributesDeleteRequest request) { - ListenableFuture> deleteFuture = attrService.removeAll(request.getTenantId(), request.getEntityId(), request.getScope(), request.getKeys()); - DonAsynchron.withCallback(deleteFuture, result -> { - calculatedFieldQueueService.pushRequestToQueue(request, result, request.getCallback()); - }, safeCallback(request.getCallback()), tsCallBackExecutor); - addWsCallback(deleteFuture, success -> onAttributesDelete(request.getTenantId(), request.getEntityId(), request.getScope().name(), request.getKeys(), request.isNotifyDevice())); + TenantId tenantId = request.getTenantId(); + EntityId entityId = request.getEntityId(); + + ListenableFuture> deleteFuture = attrService.removeAll(tenantId, entityId, request.getScope(), request.getKeys()); + + addMainCallback(deleteFuture, + result -> calculatedFieldQueueService.pushRequestToQueue(request, result, request.getCallback()), + t -> request.getCallback().onFailure(t) + ); + + if (shouldSendSharedAttributesDeletedNotification(request)) { + addMainCallback(deleteFuture, success -> clusterService.pushMsgToCore( + DeviceAttributesEventNotificationMsg.onDelete(tenantId, new DeviceId(entityId.getId()), DataConstants.SHARED_SCOPE, request.getKeys()), null + )); + } + + if (inactivityTimeoutDeleted(request)) { + addMainCallback(deleteFuture, success -> deviceStateManager.onDeviceInactivityTimeoutUpdate( + tenantId, new DeviceId(entityId.getId()), 0L, TbCallback.EMPTY) + ); + } + + addWsCallback(deleteFuture, success -> onAttributesDelete(tenantId, entityId, request.getScope().name(), request.getKeys())); + } + + private static boolean shouldSendSharedAttributesDeletedNotification(AttributesDeleteRequest request) { + return shouldSendSharedAttributesNotification(request.getEntityId(), request.getScope(), request.isNotifyDevice()); + } + + private static boolean shouldSendSharedAttributesNotification(EntityId entityId, AttributeScope scope, boolean notifyDevice) { + return entityId.getEntityType() == EntityType.DEVICE + && scope == AttributeScope.SHARED_SCOPE + && notifyDevice; + } + + private static boolean inactivityTimeoutDeleted(AttributesDeleteRequest request) { + return request.getEntityId().getEntityType() == EntityType.DEVICE + && request.getScope() == AttributeScope.SERVER_SCOPE + && request.getKeys().stream().anyMatch(key -> Objects.equals(DefaultDeviceStateService.INACTIVITY_TIMEOUT, key)); } @Override @@ -247,7 +354,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer if (entries != null) { Optional tsKvEntry = entries.stream() .filter(entry -> entry.getTs() > startTs && entry.getTs() <= endTs) - .max(Comparator.comparingLong(TsKvEntry::getTs)); + .max(comparingLong(TsKvEntry::getTs)); tsKvEntry.ifPresent(entityViewLatest::add); } } @@ -280,16 +387,16 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer } } - private void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes, boolean notifyDevice) { + private void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes) { forwardToSubscriptionManagerService(tenantId, entityId, - subscriptionManagerService -> subscriptionManagerService.onAttributesUpdate(tenantId, entityId, scope, attributes, notifyDevice, TbCallback.EMPTY), + subscriptionManagerService -> subscriptionManagerService.onAttributesUpdate(tenantId, entityId, scope, attributes, TbCallback.EMPTY), () -> TbSubscriptionUtils.toAttributesUpdateProto(tenantId, entityId, scope, attributes)); } - private void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List keys, boolean notifyDevice) { + private void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List keys) { forwardToSubscriptionManagerService(tenantId, entityId, - subscriptionManagerService -> subscriptionManagerService.onAttributesDelete(tenantId, entityId, scope, keys, notifyDevice, TbCallback.EMPTY), - () -> TbSubscriptionUtils.toAttributesDeleteProto(tenantId, entityId, scope, keys, notifyDevice)); + subscriptionManagerService -> subscriptionManagerService.onAttributesDelete(tenantId, entityId, scope, keys, TbCallback.EMPTY), + () -> TbSubscriptionUtils.toAttributesDeleteProto(tenantId, entityId, scope, keys)); } private void onTimeSeriesUpdate(TenantId tenantId, EntityId entityId, List ts) { diff --git a/application/src/test/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionServiceTest.java b/application/src/test/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionServiceTest.java index 6bb4f0e2fe..64845a12cd 100644 --- a/application/src/test/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionServiceTest.java @@ -24,25 +24,36 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.EnumSource; import org.junit.jupiter.params.provider.MethodSource; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import org.springframework.test.util.ReflectionTestUtils; +import org.thingsboard.rule.engine.api.AttributesDeleteRequest; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; +import org.thingsboard.rule.engine.api.DeviceStateManager; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.ApiUsageRecordKey; import org.thingsboard.server.common.data.ApiUsageState; import org.thingsboard.server.common.data.ApiUsageStateValue; +import org.thingsboard.server.common.data.AttributeScope; +import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityView; import org.thingsboard.server.common.data.id.ApiUsageStateId; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.id.EntityViewId; import org.thingsboard.server.common.data.id.TenantId; +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.DoubleDataEntry; import org.thingsboard.server.common.data.kv.KvEntry; +import org.thingsboard.server.common.data.kv.LongDataEntry; +import org.thingsboard.server.common.data.kv.StringDataEntry; import org.thingsboard.server.common.data.kv.TimeseriesSaveResult; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.objects.AttributesEntityView; @@ -50,6 +61,7 @@ import org.thingsboard.server.common.data.objects.TelemetryEntityView; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.common.msg.rule.engine.DeviceAttributesEventNotificationMsg; import org.thingsboard.server.common.stats.TbApiUsageReportClient; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.timeseries.TimeseriesService; @@ -73,14 +85,17 @@ import java.util.concurrent.ExecutorService; import java.util.stream.LongStream; import java.util.stream.Stream; +import static com.google.common.util.concurrent.Futures.immediateFailedFuture; import static com.google.common.util.concurrent.Futures.immediateFuture; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.then; import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.never; @ExtendWith(MockitoExtension.class) class DefaultTelemetrySubscriptionServiceTest { @@ -91,7 +106,7 @@ class DefaultTelemetrySubscriptionServiceTest { final long sampleTtl = 10_000L; - final List sampleTelemetry = List.of( + final List sampleTimeseries = List.of( new BasicTsKvEntry(100L, new DoubleDataEntry("temperature", 65.2)), new BasicTsKvEntry(100L, new DoubleDataEntry("humidity", 33.1)) ); @@ -124,12 +139,14 @@ class DefaultTelemetrySubscriptionServiceTest { TbApiUsageStateService apiUsageStateService; @Mock CalculatedFieldQueueService calculatedFieldQueueService; + @Mock + DeviceStateManager deviceStateManager; DefaultTelemetrySubscriptionService telemetryService; @BeforeEach void setup() { - telemetryService = new DefaultTelemetrySubscriptionService(attrService, tsService, tbEntityViewService, apiUsageClient, apiUsageStateService, calculatedFieldQueueService); + telemetryService = new DefaultTelemetrySubscriptionService(attrService, tsService, tbEntityViewService, apiUsageClient, apiUsageStateService, calculatedFieldQueueService, deviceStateManager); ReflectionTestUtils.setField(telemetryService, "clusterService", clusterService); ReflectionTestUtils.setField(telemetryService, "partitionService", partitionService); ReflectionTestUtils.setField(telemetryService, "subscriptionManagerService", Optional.of(subscriptionManagerService)); @@ -146,9 +163,9 @@ class DefaultTelemetrySubscriptionServiceTest { lenient().when(partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId)).thenReturn(tpi); - lenient().when(tsService.save(tenantId, entityId, sampleTelemetry, sampleTtl)).thenReturn(immediateFuture(TimeseriesSaveResult.of(sampleTelemetry.size(), listOfNNumbers(sampleTelemetry.size())))); - lenient().when(tsService.saveWithoutLatest(tenantId, entityId, sampleTelemetry, sampleTtl)).thenReturn(immediateFuture(TimeseriesSaveResult.of(sampleTelemetry.size(), null))); - lenient().when(tsService.saveLatest(tenantId, entityId, sampleTelemetry)).thenReturn(immediateFuture(TimeseriesSaveResult.of(sampleTelemetry.size(), listOfNNumbers(sampleTelemetry.size())))); + lenient().when(tsService.save(tenantId, entityId, sampleTimeseries, sampleTtl)).thenReturn(immediateFuture(TimeseriesSaveResult.of(sampleTimeseries.size(), listOfNNumbers(sampleTimeseries.size())))); + lenient().when(tsService.saveWithoutLatest(tenantId, entityId, sampleTimeseries, sampleTtl)).thenReturn(immediateFuture(TimeseriesSaveResult.of(sampleTimeseries.size(), null))); + lenient().when(tsService.saveLatest(tenantId, entityId, sampleTimeseries)).thenReturn(immediateFuture(TimeseriesSaveResult.of(sampleTimeseries.size(), listOfNNumbers(sampleTimeseries.size())))); // mock no entity views lenient().when(tbEntityViewService.findEntityViewsByTenantIdAndEntityIdAsync(tenantId, entityId)).thenReturn(immediateFuture(Collections.emptyList())); @@ -179,7 +196,7 @@ class DefaultTelemetrySubscriptionServiceTest { .tenantId(tenantId) .customerId(customerId) .entityId(new ApiUsageStateId(UUID.randomUUID())) - .entries(sampleTelemetry) + .entries(sampleTimeseries) .strategy(TimeseriesSaveRequest.Strategy.PROCESS_ALL) .build(); @@ -199,7 +216,7 @@ class DefaultTelemetrySubscriptionServiceTest { .tenantId(tenantId) .customerId(customerId) .entityId(entityId) - .entries(sampleTelemetry) + .entries(sampleTimeseries) .ttl(sampleTtl) .strategy(new TimeseriesSaveRequest.Strategy(true, false, false, false)) .build(); @@ -208,7 +225,7 @@ class DefaultTelemetrySubscriptionServiceTest { telemetryService.saveTimeseries(request); // THEN - then(apiUsageClient).should().report(tenantId, customerId, ApiUsageRecordKey.STORAGE_DP_COUNT, sampleTelemetry.size()); + then(apiUsageClient).should().report(tenantId, customerId, ApiUsageRecordKey.STORAGE_DP_COUNT, sampleTimeseries.size()); } @Test @@ -218,7 +235,7 @@ class DefaultTelemetrySubscriptionServiceTest { .tenantId(tenantId) .customerId(customerId) .entityId(entityId) - .entries(sampleTelemetry) + .entries(sampleTimeseries) .ttl(sampleTtl) .strategy(TimeseriesSaveRequest.Strategy.LATEST_AND_WS) .build(); @@ -240,7 +257,7 @@ class DefaultTelemetrySubscriptionServiceTest { .tenantId(tenantId) .customerId(customerId) .entityId(entityId) - .entries(sampleTelemetry) + .entries(sampleTimeseries) .ttl(sampleTtl) .strategy(TimeseriesSaveRequest.Strategy.PROCESS_ALL) .future(future) @@ -266,7 +283,7 @@ class DefaultTelemetrySubscriptionServiceTest { .tenantId(tenantId) .customerId(customerId) .entityId(entityId) - .entries(sampleTelemetry) + .entries(sampleTimeseries) .ttl(sampleTtl) .strategy(TimeseriesSaveRequest.Strategy.LATEST_AND_WS) .future(future) @@ -286,12 +303,12 @@ class DefaultTelemetrySubscriptionServiceTest { entityView.setTenantId(tenantId); entityView.setCustomerId(customerId); entityView.setEntityId(entityId); - entityView.setKeys(new TelemetryEntityView(sampleTelemetry.stream().map(KvEntry::getKey).toList(), new AttributesEntityView())); + entityView.setKeys(new TelemetryEntityView(sampleTimeseries.stream().map(KvEntry::getKey).toList(), new AttributesEntityView())); // mock that there is one entity view given(tbEntityViewService.findEntityViewsByTenantIdAndEntityIdAsync(tenantId, entityId)).willReturn(immediateFuture(List.of(entityView))); // mock that save latest call for entity view is successful - given(tsService.saveLatest(tenantId, entityView.getId(), sampleTelemetry)).willReturn(immediateFuture(TimeseriesSaveResult.of(sampleTelemetry.size(), listOfNNumbers(sampleTelemetry.size())))); + given(tsService.saveLatest(tenantId, entityView.getId(), sampleTimeseries)).willReturn(immediateFuture(TimeseriesSaveResult.of(sampleTimeseries.size(), listOfNNumbers(sampleTimeseries.size())))); // mock TPI for entity view given(partitionService.resolve(ServiceType.TB_CORE, tenantId, entityView.getId())).willReturn(tpi); @@ -299,7 +316,7 @@ class DefaultTelemetrySubscriptionServiceTest { .tenantId(tenantId) .customerId(customerId) .entityId(entityId) - .entries(sampleTelemetry) + .entries(sampleTimeseries) .ttl(sampleTtl) .strategy(new TimeseriesSaveRequest.Strategy(false, true, false, false)) .build(); @@ -309,12 +326,12 @@ class DefaultTelemetrySubscriptionServiceTest { // THEN // should save latest to both the main entity and it's entity view - then(tsService).should().saveLatest(tenantId, entityId, sampleTelemetry); - then(tsService).should().saveLatest(tenantId, entityView.getId(), sampleTelemetry); + then(tsService).should().saveLatest(tenantId, entityId, sampleTimeseries); + then(tsService).should().saveLatest(tenantId, entityView.getId(), sampleTimeseries); then(tsService).shouldHaveNoMoreInteractions(); // should send WS update only for entity view (WS update for the main entity is disabled in the save request) - then(subscriptionManagerService).should().onTimeSeriesUpdate(tenantId, entityView.getId(), sampleTelemetry, TbCallback.EMPTY); + then(subscriptionManagerService).should().onTimeSeriesUpdate(tenantId, entityView.getId(), sampleTimeseries, TbCallback.EMPTY); then(subscriptionManagerService).shouldHaveNoMoreInteractions(); } @@ -325,7 +342,7 @@ class DefaultTelemetrySubscriptionServiceTest { .tenantId(tenantId) .customerId(customerId) .entityId(entityId) - .entries(sampleTelemetry) + .entries(sampleTimeseries) .ttl(sampleTtl) .strategy(new TimeseriesSaveRequest.Strategy(true, false, false, false)) .build(); @@ -335,7 +352,7 @@ class DefaultTelemetrySubscriptionServiceTest { // THEN // should save only time series for the main entity - then(tsService).should().saveWithoutLatest(tenantId, entityId, sampleTelemetry, sampleTtl); + then(tsService).should().saveWithoutLatest(tenantId, entityId, sampleTimeseries, sampleTtl); then(tsService).shouldHaveNoMoreInteractions(); // should not send any WS updates @@ -343,14 +360,14 @@ class DefaultTelemetrySubscriptionServiceTest { } @ParameterizedTest - @MethodSource("booleanCombinations") - void shouldCallCorrectApiBasedOnBooleanFlagsInTheSaveRequest(boolean saveTimeseries, boolean saveLatest, boolean sendWsUpdate, boolean processCalculatedFields) { + @MethodSource("allCombinationsOfFourBooleans") + void shouldCallCorrectSaveTimeseriesApiBasedOnBooleanFlagsInTheSaveRequest(boolean saveTimeseries, boolean saveLatest, boolean sendWsUpdate, boolean processCalculatedFields) { // GIVEN var request = TimeseriesSaveRequest.builder() .tenantId(tenantId) .customerId(customerId) .entityId(entityId) - .entries(sampleTelemetry) + .entries(sampleTimeseries) .ttl(sampleTtl) .strategy(new TimeseriesSaveRequest.Strategy(saveTimeseries, saveLatest, sendWsUpdate, processCalculatedFields)) .build(); @@ -360,11 +377,11 @@ class DefaultTelemetrySubscriptionServiceTest { // THEN if (saveTimeseries && saveLatest) { - then(tsService).should().save(tenantId, entityId, sampleTelemetry, sampleTtl); + then(tsService).should().save(tenantId, entityId, sampleTimeseries, sampleTtl); } else if (saveLatest) { - then(tsService).should().saveLatest(tenantId, entityId, sampleTelemetry); + then(tsService).should().saveLatest(tenantId, entityId, sampleTimeseries); } else if (saveTimeseries) { - then(tsService).should().saveWithoutLatest(tenantId, entityId, sampleTelemetry, sampleTtl); + then(tsService).should().saveWithoutLatest(tenantId, entityId, sampleTimeseries, sampleTtl); } if (processCalculatedFields) { @@ -374,13 +391,13 @@ class DefaultTelemetrySubscriptionServiceTest { then(tsService).shouldHaveNoMoreInteractions(); if (sendWsUpdate) { - then(subscriptionManagerService).should().onTimeSeriesUpdate(tenantId, entityId, sampleTelemetry, TbCallback.EMPTY); + then(subscriptionManagerService).should().onTimeSeriesUpdate(tenantId, entityId, sampleTimeseries, TbCallback.EMPTY); } else { then(subscriptionManagerService).shouldHaveNoInteractions(); } } - private static Stream booleanCombinations() { + private static Stream allCombinationsOfFourBooleans() { return Stream.of( Arguments.of(true, true, true, true), Arguments.of(true, true, true, false), @@ -401,7 +418,693 @@ class DefaultTelemetrySubscriptionServiceTest { ); } - // used to emulate sequence numbers returned by save latest API + /* --- Save attributes API --- */ + + @ParameterizedTest + @MethodSource("allCombinationsOfThreeBooleans") + void shouldCallCorrectSaveAttributesApiBasedOnBooleanFlagsInTheSaveRequest(boolean saveAttributes, boolean sendWsUpdate, boolean processCalculatedFields) { + // GIVEN + var request = AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(entityId) + .scope(AttributeScope.SERVER_SCOPE) + .entry(new DoubleDataEntry("temperature", 65.2)) + .notifyDevice(false) + .strategy(new AttributesSaveRequest.Strategy(saveAttributes, sendWsUpdate, processCalculatedFields)) + .build(); + + lenient().when(attrService.save(tenantId, entityId, request.getScope(), request.getEntries())).thenReturn(immediateFuture(listOfNNumbers(request.getEntries().size()))); + + // WHEN + telemetryService.saveAttributes(request); + + // THEN + if (saveAttributes) { + then(attrService).should().save(tenantId, entityId, request.getScope(), request.getEntries()); + } else { + then(attrService).shouldHaveNoInteractions(); + } + + if (processCalculatedFields) { + then(calculatedFieldQueueService).should().pushRequestToQueue(eq(request), any(), eq(request.getCallback())); + } + + if (sendWsUpdate) { + then(subscriptionManagerService).should().onAttributesUpdate(tenantId, entityId, request.getScope().name(), request.getEntries(), TbCallback.EMPTY); + } else { + then(subscriptionManagerService).shouldHaveNoInteractions(); + } + } + + static Stream allCombinationsOfThreeBooleans() { + return Stream.of( + Arguments.of(true, true, true), + Arguments.of(true, true, false), + Arguments.of(true, false, true), + Arguments.of(true, false, false), + Arguments.of(false, true, true), + Arguments.of(false, true, false), + Arguments.of(false, false, true), + Arguments.of(false, false, false) + ); + } + + @Test + void shouldThrowErrorWhenTryingToSaveAttributesForApiUsageState() { + // GIVEN + var request = AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(new ApiUsageStateId(UUID.randomUUID())) + .scope(AttributeScope.SHARED_SCOPE) + .entry(new DoubleDataEntry("temperature", 65.2)) + .notifyDevice(true) + .strategy(new AttributesSaveRequest.Strategy(true, false, false)) + .build(); + + // WHEN + assertThatThrownBy(() -> telemetryService.saveAttributes(request)) + .isInstanceOf(RuntimeException.class) + .hasMessage("Can't update API Usage State!"); + + // THEN + then(attrService).shouldHaveNoInteractions(); + } + + @Test + void shouldSendAttributesUpdateNotificationWhenDeviceSharedAttributesAreSavedAndNotifyDeviceIsTrue() { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + List entries = List.of( + new BaseAttributeKvEntry(123L, new DoubleDataEntry("shared1", 65.2)), + new BaseAttributeKvEntry(456L, new StringDataEntry("shared2", "test")) + ); + + var request = AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(AttributeScope.SHARED_SCOPE) + .entries(entries) + .notifyDevice(true) + .strategy(new AttributesSaveRequest.Strategy(true, false, false)) + .build(); + + given(attrService.save(tenantId, deviceId, request.getScope(), entries)).willReturn(immediateFuture(listOfNNumbers(entries.size()))); + + // WHEN + telemetryService.saveAttributes(request); + + // THEN + var expectedAttributesUpdateMsg = DeviceAttributesEventNotificationMsg.onUpdate(tenantId, deviceId, "SHARED_SCOPE", entries); + + then(clusterService).should().pushMsgToCore(eq(expectedAttributesUpdateMsg), isNull()); + } + + @ParameterizedTest + @EnumSource( + value = EntityType.class, + names = {"DEVICE", "API_USAGE_STATE"}, // API usage state excluded due to coverage in another test + mode = EnumSource.Mode.EXCLUDE + ) + void shouldNotSendAttributesUpdateNotificationWhenEntityIsNotDevice(EntityType entityType) { + // GIVEN + var nonDeviceId = EntityIdFactory.getByTypeAndUuid(entityType, "cc51e450-53e1-11ee-883e-e56b48fd2088"); + List entries = List.of( + new BaseAttributeKvEntry(123L, new DoubleDataEntry("shared1", 65.2)), + new BaseAttributeKvEntry(456L, new StringDataEntry("shared2", "test")) + ); + + var request = AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(nonDeviceId) + .scope(AttributeScope.SHARED_SCOPE) + .entries(entries) + .notifyDevice(true) + .strategy(new AttributesSaveRequest.Strategy(true, false, false)) + .build(); + + given(attrService.save(tenantId, nonDeviceId, request.getScope(), entries)).willReturn(immediateFuture(listOfNNumbers(entries.size()))); + + // WHEN + telemetryService.saveAttributes(request); + + // THEN + then(clusterService).should(never()).pushMsgToCore(any(), any()); + } + + @ParameterizedTest + @EnumSource( + value = AttributeScope.class, + names = "SHARED_SCOPE", + mode = EnumSource.Mode.EXCLUDE + ) + void shouldNotSendAttributesUpdateNotificationWhenAttributesAreNotShared(AttributeScope notSharedScope) { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + List entries = List.of( + new BaseAttributeKvEntry(123L, new DoubleDataEntry("shared1", 65.2)), + new BaseAttributeKvEntry(456L, new StringDataEntry("shared2", "test")) + ); + + var request = AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(notSharedScope) + .entries(entries) + .notifyDevice(true) + .strategy(new AttributesSaveRequest.Strategy(true, false, false)) + .build(); + + given(attrService.save(tenantId, deviceId, request.getScope(), entries)).willReturn(immediateFuture(listOfNNumbers(entries.size()))); + + // WHEN + telemetryService.saveAttributes(request); + + // THEN + then(clusterService).should(never()).pushMsgToCore(any(), any()); + } + + @Test + void shouldNotSendAttributesUpdateNotificationWhenNotifyDeviceIsFalse() { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + List entries = List.of( + new BaseAttributeKvEntry(123L, new DoubleDataEntry("shared1", 65.2)), + new BaseAttributeKvEntry(456L, new StringDataEntry("shared2", "test")) + ); + + var request = AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(AttributeScope.SHARED_SCOPE) + .entries(entries) + .notifyDevice(false) + .strategy(new AttributesSaveRequest.Strategy(true, false, false)) + .build(); + + given(attrService.save(tenantId, deviceId, request.getScope(), entries)).willReturn(immediateFuture(listOfNNumbers(entries.size()))); + + // WHEN + telemetryService.saveAttributes(request); + + // THEN + then(clusterService).should(never()).pushMsgToCore(any(), any()); + } + + @Test + void shouldNotSendAttributesUpdateNotificationWhenAttributesSaveWasSkipped() { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + List entries = List.of( + new BaseAttributeKvEntry(123L, new DoubleDataEntry("shared1", 65.2)), + new BaseAttributeKvEntry(456L, new StringDataEntry("shared2", "test")) + ); + + var request = AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(AttributeScope.SHARED_SCOPE) + .entries(entries) + .notifyDevice(true) + .strategy(new AttributesSaveRequest.Strategy(false, false, false)) + .build(); + + // WHEN + telemetryService.saveAttributes(request); + + // THEN + then(clusterService).should(never()).pushMsgToCore(any(), any()); + } + + @Test + void shouldNotSendAttributesUpdateNotificationWhenAttributesSaveFailed() { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + List entries = List.of( + new BaseAttributeKvEntry(123L, new DoubleDataEntry("shared1", 65.2)), + new BaseAttributeKvEntry(456L, new StringDataEntry("shared2", "test")) + ); + + var request = AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(AttributeScope.SHARED_SCOPE) + .entries(entries) + .notifyDevice(true) + .strategy(new AttributesSaveRequest.Strategy(true, false, false)) + .build(); + + given(attrService.save(tenantId, deviceId, request.getScope(), entries)).willReturn(immediateFailedFuture(new RuntimeException("failed to save"))); + + // WHEN + telemetryService.saveAttributes(request); + + // THEN + then(clusterService).should(never()).pushMsgToCore(any(), any()); + } + + @Test + void shouldNotifyDeviceStateManagerWhenDeviceInactivityTimeoutWasUpdated() { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + var inactivityTimeout = new BaseAttributeKvEntry(123L, new LongDataEntry("inactivityTimeout", 5000L)); + + var request = AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(AttributeScope.SERVER_SCOPE) + .entry(inactivityTimeout) + .strategy(new AttributesSaveRequest.Strategy(true, false, false)) + .build(); + + given(attrService.save(tenantId, deviceId, request.getScope(), request.getEntries())).willReturn(immediateFuture(listOfNNumbers(request.getEntries().size()))); + + // WHEN + telemetryService.saveAttributes(request); + + // THEN + then(deviceStateManager).should().onDeviceInactivityTimeoutUpdate(tenantId, deviceId, 5000L, TbCallback.EMPTY); + } + + @Test + void shouldNotNotifyDeviceStateManagerWhenDeviceInactivityTimeoutSaveWasSkipped() { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + var inactivityTimeout = new BaseAttributeKvEntry(123L, new LongDataEntry("inactivityTimeout", 5000L)); + + var request = AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(AttributeScope.SERVER_SCOPE) + .entry(inactivityTimeout) + .strategy(new AttributesSaveRequest.Strategy(false, true, true)) + .build(); + + // WHEN + telemetryService.saveAttributes(request); + + // THEN + then(deviceStateManager).shouldHaveNoInteractions(); + } + + @ParameterizedTest + @EnumSource( + value = EntityType.class, + names = {"DEVICE", "API_USAGE_STATE"}, // API usage state excluded due to coverage in another test + mode = EnumSource.Mode.EXCLUDE + ) + void shouldNotNotifyDeviceStateManagerWhenInactivityTimeoutWasUpdatedButEntityTypeIsNotDevice(EntityType entityType) { + // GIVEN + var nonDeviceId = EntityIdFactory.getByTypeAndUuid(entityType, "cc51e450-53e1-11ee-883e-e56b48fd2088"); + var inactivityTimeout = new BaseAttributeKvEntry(123L, new LongDataEntry("inactivityTimeout", 5000L)); + + var request = AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(nonDeviceId) + .scope(AttributeScope.SERVER_SCOPE) + .entry(inactivityTimeout) + .strategy(new AttributesSaveRequest.Strategy(true, false, false)) + .build(); + + given(attrService.save(tenantId, nonDeviceId, request.getScope(), request.getEntries())).willReturn(immediateFuture(listOfNNumbers(request.getEntries().size()))); + + // WHEN + telemetryService.saveAttributes(request); + + // THEN + then(deviceStateManager).shouldHaveNoInteractions(); + } + + @ParameterizedTest + @EnumSource( + value = AttributeScope.class, + names = {"SERVER_SCOPE"}, + mode = EnumSource.Mode.EXCLUDE + ) + void shouldNotNotifyDeviceStateManagerWhenInactivityTimeoutWasUpdatedButAttributeScopeIsNotServer(AttributeScope nonServerScope) { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + var inactivityTimeout = new BaseAttributeKvEntry(123L, new LongDataEntry("inactivityTimeout", 5000L)); + + var request = AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(nonServerScope) + .entry(inactivityTimeout) + .strategy(new AttributesSaveRequest.Strategy(true, false, false)) + .build(); + + given(attrService.save(tenantId, deviceId, request.getScope(), request.getEntries())).willReturn(immediateFuture(listOfNNumbers(request.getEntries().size()))); + + // WHEN + telemetryService.saveAttributes(request); + + // THEN + then(deviceStateManager).shouldHaveNoInteractions(); + } + + @Test + void shouldNotNotifyDeviceStateManagerWhenUpdatedAttributesDoNotContainInactivityTimeout() { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + var inactivityTimeout = new BaseAttributeKvEntry(123L, new LongDataEntry("notInactivityTimeout", 5000L)); + + var request = AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(AttributeScope.SERVER_SCOPE) + .entry(inactivityTimeout) + .strategy(new AttributesSaveRequest.Strategy(true, false, false)) + .build(); + + given(attrService.save(tenantId, deviceId, request.getScope(), request.getEntries())).willReturn(immediateFuture(listOfNNumbers(request.getEntries().size()))); + + // WHEN + telemetryService.saveAttributes(request); + + // THEN + then(deviceStateManager).shouldHaveNoInteractions(); + } + + @Test + void shouldUseInactivityTimeoutEntryWithTheGreatestVersion() { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + List entries = List.of( + new BaseAttributeKvEntry(new LongDataEntry("inactivityTimeout", 0L), 0L, null), + new BaseAttributeKvEntry(new LongDataEntry("inactivityTimeout", 1000L), 3L, 1L), + new BaseAttributeKvEntry(new LongDataEntry("inactivityTimeout", 2000L), 2L, 2L), + new BaseAttributeKvEntry(new LongDataEntry("inactivityTimeout", 3000L), 1L, 3L) + ); + + var request = AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(AttributeScope.SERVER_SCOPE) + .entries(entries) + .strategy(new AttributesSaveRequest.Strategy(true, false, false)) + .build(); + + given(attrService.save(tenantId, deviceId, request.getScope(), request.getEntries())).willReturn(immediateFuture(listOfNNumbers(request.getEntries().size()))); + + // WHEN + telemetryService.saveAttributes(request); + + // THEN + then(deviceStateManager).should().onDeviceInactivityTimeoutUpdate(tenantId, deviceId, 3000L, TbCallback.EMPTY); + } + + @Test + void shouldUseInactivityTimeoutEntryWithTheGreatestLastUpdateTsWhenVersionsAreTheSame() { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + List entries = List.of( + new BaseAttributeKvEntry(new LongDataEntry("inactivityTimeout", 1000L), 1L, 1L), + new BaseAttributeKvEntry(new LongDataEntry("inactivityTimeout", 2000L), 2L, 1L), + new BaseAttributeKvEntry(new LongDataEntry("inactivityTimeout", 3000L), 3L, 1L) + ); + + var request = AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(AttributeScope.SERVER_SCOPE) + .entries(entries) + .strategy(new AttributesSaveRequest.Strategy(true, false, false)) + .build(); + + given(attrService.save(tenantId, deviceId, request.getScope(), request.getEntries())).willReturn(immediateFuture(listOfNNumbers(request.getEntries().size()))); + + // WHEN + telemetryService.saveAttributes(request); + + // THEN + then(deviceStateManager).should().onDeviceInactivityTimeoutUpdate(tenantId, deviceId, 3000L, TbCallback.EMPTY); + } + + /* --- Delete attributes API --- */ + + @Test + void shouldThrowErrorWhenTryingToDeleteAttributesForApiUsageState() { + // GIVEN + var request = AttributesDeleteRequest.builder() + .tenantId(tenantId) + .entityId(new ApiUsageStateId(UUID.randomUUID())) + .scope(AttributeScope.SHARED_SCOPE) + .keys(List.of("attributeKeyToDelete1", "attributeKeyToDelete2")) + .notifyDevice(true) + .build(); + + // WHEN + assertThatThrownBy(() -> telemetryService.deleteAttributes(request)) + .isInstanceOf(RuntimeException.class) + .hasMessage("Can't update API Usage State!"); + + // THEN + then(attrService).shouldHaveNoInteractions(); + } + + @Test + void shouldSendAttributesDeletedNotificationWhenDeviceSharedAttributesAreDeletedAndNotifyDeviceIsTrue() { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + List keys = List.of("attributeKeyToDelete1", "attributeKeyToDelete2"); + + var request = AttributesDeleteRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(AttributeScope.SHARED_SCOPE) + .keys(keys) + .notifyDevice(true) + .build(); + + given(attrService.removeAll(tenantId, deviceId, request.getScope(), keys)).willReturn(immediateFuture(keys)); + + // WHEN + telemetryService.deleteAttributes(request); + + // THEN + var expectedAttributesDeletedMsg = DeviceAttributesEventNotificationMsg.onDelete(tenantId, deviceId, "SHARED_SCOPE", List.of("attributeKeyToDelete1", "attributeKeyToDelete2")); + + then(clusterService).should().pushMsgToCore(eq(expectedAttributesDeletedMsg), isNull()); + } + + @ParameterizedTest + @EnumSource( + value = EntityType.class, + names = {"DEVICE", "API_USAGE_STATE"}, // API usage state excluded due to coverage in another test + mode = EnumSource.Mode.EXCLUDE + ) + void shouldNotSendAttributesDeletedNotificationWhenEntityIsNotDevice(EntityType entityType) { + // GIVEN + var nonDeviceId = EntityIdFactory.getByTypeAndUuid(entityType, "cc51e450-53e1-11ee-883e-e56b48fd2088"); + List keys = List.of("attributeKeyToDelete1", "attributeKeyToDelete2"); + + var request = AttributesDeleteRequest.builder() + .tenantId(tenantId) + .entityId(nonDeviceId) + .scope(AttributeScope.SHARED_SCOPE) + .keys(keys) + .notifyDevice(true) + .build(); + + given(attrService.removeAll(tenantId, nonDeviceId, request.getScope(), keys)).willReturn(immediateFuture(keys)); + + // WHEN + telemetryService.deleteAttributes(request); + + // THEN + then(clusterService).should(never()).pushMsgToCore(any(), any()); + } + + @ParameterizedTest + @EnumSource( + value = AttributeScope.class, + names = "SHARED_SCOPE", + mode = EnumSource.Mode.EXCLUDE + ) + void shouldNotSendAttributesDeletedNotificationWhenAttributesAreNotShared(AttributeScope notSharedScope) { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + List keys = List.of("attributeKeyToDelete1", "attributeKeyToDelete2"); + + var request = AttributesDeleteRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(notSharedScope) + .keys(keys) + .notifyDevice(true) + .build(); + + given(attrService.removeAll(tenantId, deviceId, request.getScope(), keys)).willReturn(immediateFuture(keys)); + + // WHEN + telemetryService.deleteAttributes(request); + + // THEN + then(clusterService).should(never()).pushMsgToCore(any(), any()); + } + + @Test + void shouldNotSendAttributesDeletedNotificationWhenNotifyDeviceIsFalse() { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + List keys = List.of("attributeKeyToDelete1", "attributeKeyToDelete2"); + + var request = AttributesDeleteRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(AttributeScope.SHARED_SCOPE) + .keys(keys) + .notifyDevice(false) + .build(); + + given(attrService.removeAll(tenantId, deviceId, request.getScope(), keys)).willReturn(immediateFuture(keys)); + + // WHEN + telemetryService.deleteAttributes(request); + + // THEN + then(clusterService).should(never()).pushMsgToCore(any(), any()); + } + + @Test + void shouldNotSendAttributesDeletedNotificationWhenAttributesDeleteFailed() { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + List keys = List.of("attributeKeyToDelete1", "attributeKeyToDelete2"); + + var request = AttributesDeleteRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(AttributeScope.SHARED_SCOPE) + .keys(keys) + .notifyDevice(true) + .build(); + + given(attrService.removeAll(tenantId, deviceId, request.getScope(), keys)).willReturn(immediateFailedFuture(new RuntimeException("failed to delete"))); + + // WHEN + telemetryService.deleteAttributes(request); + + // THEN + then(clusterService).should(never()).pushMsgToCore(any(), any()); + } + + @Test + void shouldNotifyDeviceStateManagerWhenDeviceInactivityTimeoutWasDeleted() { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + + var request = AttributesDeleteRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(AttributeScope.SERVER_SCOPE) + .keys(List.of("inactivityTimeout", "someOtherDeletedAttribute")) + .build(); + + given(attrService.removeAll(tenantId, deviceId, request.getScope(), request.getKeys())).willReturn(immediateFuture(request.getKeys())); + + // WHEN + telemetryService.deleteAttributes(request); + + // THEN + then(deviceStateManager).should().onDeviceInactivityTimeoutUpdate(tenantId, deviceId, 0L, TbCallback.EMPTY); + } + + @ParameterizedTest + @EnumSource( + value = EntityType.class, + names = {"DEVICE", "API_USAGE_STATE"}, // API usage state excluded due to coverage in another test + mode = EnumSource.Mode.EXCLUDE + ) + void shouldNotNotifyDeviceStateManagerWhenInactivityTimeoutWasDeletedButEntityTypeIsNotDevice(EntityType entityType) { + // GIVEN + var nonDeviceId = EntityIdFactory.getByTypeAndUuid(entityType, "cc51e450-53e1-11ee-883e-e56b48fd2088"); + + var request = AttributesDeleteRequest.builder() + .tenantId(tenantId) + .entityId(nonDeviceId) + .scope(AttributeScope.SERVER_SCOPE) + .keys(List.of("inactivityTimeout", "someOtherDeletedAttribute")) + .build(); + + given(attrService.removeAll(tenantId, nonDeviceId, request.getScope(), request.getKeys())).willReturn(immediateFuture(request.getKeys())); + + // WHEN + telemetryService.deleteAttributes(request); + + // THEN + then(deviceStateManager).shouldHaveNoInteractions(); + } + + @ParameterizedTest + @EnumSource( + value = AttributeScope.class, + names = {"SERVER_SCOPE"}, + mode = EnumSource.Mode.EXCLUDE + ) + void shouldNotNotifyDeviceStateManagerWhenInactivityTimeoutWasDeletedButAttributeScopeIsNotServer(AttributeScope nonServerScope) { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + + var request = AttributesDeleteRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(nonServerScope) + .keys(List.of("inactivityTimeout", "someOtherDeletedAttribute")) + .build(); + + given(attrService.removeAll(tenantId, deviceId, request.getScope(), request.getKeys())).willReturn(immediateFuture(request.getKeys())); + + // WHEN + telemetryService.deleteAttributes(request); + + // THEN + then(deviceStateManager).shouldHaveNoInteractions(); + } + + @Test + void shouldNotNotifyDeviceStateManagerWhenInactivityTimeoutWasNotDeleted() { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + + var request = AttributesDeleteRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(AttributeScope.SERVER_SCOPE) + .keys(List.of("someOtherDeletedAttribute")) + .build(); + + given(attrService.removeAll(tenantId, deviceId, request.getScope(), request.getKeys())).willReturn(immediateFuture(request.getKeys())); + + // WHEN + telemetryService.deleteAttributes(request); + + // THEN + then(deviceStateManager).shouldHaveNoInteractions(); + } + + @Test + void shouldNotNotifyDeviceStateManagerWhenDeviceInactivityTimeoutDeleteFailed() { + // GIVEN + var deviceId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); + + var request = AttributesDeleteRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(AttributeScope.SERVER_SCOPE) + .keys(List.of("inactivityTimeout", "someOtherDeletedAttribute")) + .build(); + + given(attrService.removeAll(tenantId, deviceId, request.getScope(), request.getKeys())).willReturn(immediateFailedFuture(new RuntimeException("failed to delete"))); + + // WHEN + telemetryService.deleteAttributes(request); + + // THEN + then(deviceStateManager).shouldHaveNoInteractions(); + } + + // used to emulate versions returned by save APIs private static List listOfNNumbers(int N) { return LongStream.range(0, N).boxed().toList(); } diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 38f86bf9db..6ecdf13981 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -1122,7 +1122,11 @@ message TbAttributeDeleteProto { int64 tenantIdLSB = 5; string scope = 6; repeated string keys = 7; - bool notifyDevice = 8; + // DEPRECATED. FOR REMOVAL + // Since 4.0, this field is no longer used. + // Device notifications are now handled directly by DefaultTelemetrySubscriptionService, + // eliminating the need to pass this parameter through the queue and proto to DefaultSubscriptionManagerService. + optional bool notifyDevice = 8 [deprecated = true]; } message TbTimeSeriesDeleteProto { diff --git a/common/queue/src/test/java/org/thingsboard/server/queue/discovery/ZkDiscoveryServiceTest.java b/common/queue/src/test/java/org/thingsboard/server/queue/discovery/ZkDiscoveryServiceTest.java index 19643f4ad4..c8d91b0c0f 100644 --- a/common/queue/src/test/java/org/thingsboard/server/queue/discovery/ZkDiscoveryServiceTest.java +++ b/common/queue/src/test/java/org/thingsboard/server/queue/discovery/ZkDiscoveryServiceTest.java @@ -42,7 +42,11 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.*; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; @ExtendWith(MockitoExtension.class) public class ZkDiscoveryServiceTest { diff --git a/monitoring/src/main/resources/root_rule_chain.json b/monitoring/src/main/resources/root_rule_chain.json index 46bdc72d9f..a1c12c8e9d 100644 --- a/monitoring/src/main/resources/root_rule_chain.json +++ b/monitoring/src/main/resources/root_rule_chain.json @@ -39,8 +39,11 @@ "type": "org.thingsboard.rule.engine.telemetry.TbMsgAttributesNode", "name": "Save Attributes", "singletonMode": false, - "configurationVersion": 1, + "configurationVersion": 3, "configuration": { + "processingSettings": { + "type": "ON_EVERY_MESSAGE" + }, "scope": "CLIENT_SCOPE", "notifyDevice": false, "sendAttributesUpdatedNotification": false, diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesDeleteRequest.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesDeleteRequest.java index 2ab6923899..374fcc45f6 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesDeleteRequest.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesDeleteRequest.java @@ -21,6 +21,7 @@ import lombok.AccessLevel; import lombok.AllArgsConstructor; import lombok.Getter; import lombok.ToString; +import org.thingsboard.common.util.NoOpFutureCallback; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; @@ -30,6 +31,8 @@ import org.thingsboard.server.common.data.msg.TbMsgType; import java.util.List; import java.util.UUID; +import static java.util.Objects.requireNonNullElse; + @Getter @ToString @AllArgsConstructor(access = AccessLevel.PRIVATE) @@ -61,8 +64,7 @@ public class AttributesDeleteRequest implements CalculatedFieldSystemAwareReques private TbMsgType tbMsgType; private FutureCallback callback; - Builder() { - } + Builder() {} public Builder tenantId(TenantId tenantId) { this.tenantId = tenantId; @@ -134,7 +136,9 @@ public class AttributesDeleteRequest implements CalculatedFieldSystemAwareReques } public AttributesDeleteRequest build() { - return new AttributesDeleteRequest(tenantId, entityId, scope, keys, notifyDevice, previousCalculatedFieldIds, tbMsgId, tbMsgType, callback); + return new AttributesDeleteRequest( + tenantId, entityId, scope, keys, notifyDevice, previousCalculatedFieldIds, tbMsgId, tbMsgType, requireNonNullElse(callback, NoOpFutureCallback.instance()) + ); } } diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java index 7c8e9d317e..c3095836bd 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java @@ -21,6 +21,7 @@ import lombok.AccessLevel; import lombok.AllArgsConstructor; import lombok.Getter; import lombok.ToString; +import org.thingsboard.common.util.NoOpFutureCallback; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; @@ -33,6 +34,8 @@ import org.thingsboard.server.common.data.msg.TbMsgType; import java.util.List; import java.util.UUID; +import static java.util.Objects.requireNonNullElse; + @Getter @ToString @AllArgsConstructor(access = AccessLevel.PRIVATE) @@ -43,11 +46,20 @@ public class AttributesSaveRequest implements CalculatedFieldSystemAwareRequest private final AttributeScope scope; private final List entries; private final boolean notifyDevice; + private final Strategy strategy; private final List previousCalculatedFieldIds; private final UUID tbMsgId; private final TbMsgType tbMsgType; private final FutureCallback callback; + public record Strategy(boolean saveAttributes, boolean sendWsUpdate, boolean processCalculatedFields) { + + public static final Strategy PROCESS_ALL = new Strategy(true, true, true); + public static final Strategy WS_ONLY = new Strategy(false, true, false); + public static final Strategy SKIP_ALL = new Strategy(false, false, false); + + } + public static Builder builder() { return new Builder(); } @@ -59,6 +71,7 @@ public class AttributesSaveRequest implements CalculatedFieldSystemAwareRequest private AttributeScope scope; private List entries; private boolean notifyDevice = true; + private Strategy strategy; private List previousCalculatedFieldIds; private UUID tbMsgId; private TbMsgType tbMsgType; @@ -109,6 +122,11 @@ public class AttributesSaveRequest implements CalculatedFieldSystemAwareRequest return this; } + public Builder strategy(Strategy strategy) { + this.strategy = strategy; + return this; + } + public Builder previousCalculatedFieldIds(List previousCalculatedFieldIds) { this.previousCalculatedFieldIds = previousCalculatedFieldIds; return this; @@ -144,7 +162,10 @@ public class AttributesSaveRequest implements CalculatedFieldSystemAwareRequest } public AttributesSaveRequest build() { - return new AttributesSaveRequest(tenantId, entityId, scope, entries, notifyDevice, previousCalculatedFieldIds, tbMsgId, tbMsgType, callback); + return new AttributesSaveRequest( + tenantId, entityId, scope, entries, notifyDevice, requireNonNullElse(strategy, Strategy.PROCESS_ALL), + previousCalculatedFieldIds, tbMsgId, tbMsgType, requireNonNullElse(callback, NoOpFutureCallback.instance()) + ); } } diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java index 01af278efb..c402a0c984 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java @@ -70,7 +70,7 @@ public class TimeseriesSaveRequest implements CalculatedFieldSystemAwareRequest private EntityId entityId; private List entries; private long ttl; - private Strategy strategy = Strategy.PROCESS_ALL; + private Strategy strategy; private List previousCalculatedFieldIds; private UUID tbMsgId; private TbMsgType tbMsgType; @@ -152,7 +152,7 @@ public class TimeseriesSaveRequest implements CalculatedFieldSystemAwareRequest public TimeseriesSaveRequest build() { return new TimeseriesSaveRequest( - tenantId, customerId, entityId, entries, ttl, strategy, + tenantId, customerId, entityId, entries, ttl, requireNonNullElse(strategy, Strategy.PROCESS_ALL), previousCalculatedFieldIds, tbMsgId, tbMsgType, requireNonNullElse(callback, NoOpFutureCallback.instance()) ); } diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeConfigurationTest.java b/rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/AttributesDeleteRequestTest.java similarity index 57% rename from rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeConfigurationTest.java rename to rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/AttributesDeleteRequestTest.java index 5c4d341fda..9b4a825a66 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeConfigurationTest.java +++ b/rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/AttributesDeleteRequestTest.java @@ -13,17 +13,27 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.rule.engine.telemetry; +package org.thingsboard.rule.engine.api; import org.junit.jupiter.api.Test; +import org.thingsboard.common.util.NoOpFutureCallback; import static org.assertj.core.api.Assertions.assertThat; -class TbMsgAttributesNodeConfigurationTest { +class AttributesDeleteRequestTest { @Test - void testDefaultConfig_givenUpdateAttributesOnlyOnValueChange_thenTrue_sinceVersion1() { - assertThat(new TbMsgAttributesNodeConfiguration().defaultConfiguration().isUpdateAttributesOnlyOnValueChange()).isTrue(); + void testDefaultCallbackIsNoOp() { + var request = AttributesDeleteRequest.builder().build(); + + assertThat(request.getCallback()).isEqualTo(NoOpFutureCallback.instance()); + } + + @Test + void testNullCallbackIsNoOp() { + var request = AttributesDeleteRequest.builder().callback(null).build(); + + assertThat(request.getCallback()).isEqualTo(NoOpFutureCallback.instance()); } } diff --git a/rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/AttributesSaveRequestTest.java b/rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/AttributesSaveRequestTest.java new file mode 100644 index 0000000000..d632edfbf9 --- /dev/null +++ b/rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/AttributesSaveRequestTest.java @@ -0,0 +1,68 @@ +/** + * 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.rule.engine.api; + +import org.junit.jupiter.api.Test; +import org.thingsboard.common.util.NoOpFutureCallback; + +import static org.assertj.core.api.Assertions.assertThat; + +class AttributesSaveRequestTest { + + @Test + void testDefaultProcessingStrategyIsProcessAll() { + var request = AttributesSaveRequest.builder().build(); + + assertThat(request.getStrategy()).isEqualTo(AttributesSaveRequest.Strategy.PROCESS_ALL); + } + + @Test + void testNullProcessingStrategyIsProcessAll() { + var request = AttributesSaveRequest.builder().strategy(null).build(); + + assertThat(request.getStrategy()).isEqualTo(AttributesSaveRequest.Strategy.PROCESS_ALL); + } + + @Test + void testProcessAllStrategy() { + assertThat(AttributesSaveRequest.Strategy.PROCESS_ALL).isEqualTo(new AttributesSaveRequest.Strategy(true, true, true)); + } + + @Test + void testWsOnlyStrategy() { + assertThat(AttributesSaveRequest.Strategy.WS_ONLY).isEqualTo(new AttributesSaveRequest.Strategy(false, true, false)); + } + + @Test + void testSkipAllStrategy() { + assertThat(AttributesSaveRequest.Strategy.SKIP_ALL).isEqualTo(new AttributesSaveRequest.Strategy(false, false, false)); + } + + @Test + void testDefaultCallbackIsNoOp() { + var request = AttributesSaveRequest.builder().build(); + + assertThat(request.getCallback()).isEqualTo(NoOpFutureCallback.instance()); + } + + @Test + void testNullCallbackIsNoOp() { + var request = AttributesSaveRequest.builder().callback(null).build(); + + assertThat(request.getCallback()).isEqualTo(NoOpFutureCallback.instance()); + } + +} diff --git a/rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequestTest.java b/rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequestTest.java index a67a11ad3d..c0eb04152a 100644 --- a/rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequestTest.java +++ b/rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequestTest.java @@ -23,14 +23,21 @@ import static org.assertj.core.api.Assertions.assertThat; class TimeseriesSaveRequestTest { @Test - void testDefaultSaveStrategyIsProcessAll() { + void testDefaultProcessingStrategyIsProcessAll() { var request = TimeseriesSaveRequest.builder().build(); assertThat(request.getStrategy()).isEqualTo(TimeseriesSaveRequest.Strategy.PROCESS_ALL); } @Test - void testSaveAllStrategy() { + void testNullProcessingStrategyIsProcessAll() { + var request = TimeseriesSaveRequest.builder().strategy(null).build(); + + assertThat(request.getStrategy()).isEqualTo(TimeseriesSaveRequest.Strategy.PROCESS_ALL); + } + + @Test + void testProcessAllStrategy() { assertThat(TimeseriesSaveRequest.Strategy.PROCESS_ALL).isEqualTo(new TimeseriesSaveRequest.Strategy(true, true, true, true)); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java index c817336eae..909fca51d4 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java @@ -23,6 +23,7 @@ import com.google.common.util.concurrent.MoreExecutors; import com.google.gson.JsonParser; import lombok.extern.slf4j.Slf4j; import org.thingsboard.common.util.DonAsynchron; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.RuleNode; import org.thingsboard.rule.engine.api.TbContext; @@ -30,6 +31,7 @@ import org.thingsboard.rule.engine.api.TbNode; import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeException; import org.thingsboard.rule.engine.api.util.TbNodeUtils; +import org.thingsboard.rule.engine.telemetry.settings.AttributesProcessingSettings; import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.StringUtils; @@ -43,9 +45,14 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.UUID; import java.util.function.Function; import java.util.stream.Collectors; +import static org.thingsboard.rule.engine.telemetry.settings.AttributesProcessingSettings.Advanced; +import static org.thingsboard.rule.engine.telemetry.settings.AttributesProcessingSettings.Deduplicate; +import static org.thingsboard.rule.engine.telemetry.settings.AttributesProcessingSettings.OnEveryMessage; +import static org.thingsboard.rule.engine.telemetry.settings.AttributesProcessingSettings.WebSocketsOnly; import static org.thingsboard.server.common.data.DataConstants.NOTIFY_DEVICE_METADATA_KEY; import static org.thingsboard.server.common.data.DataConstants.SCOPE; import static org.thingsboard.server.common.data.msg.TbMsgType.POST_ATTRIBUTES_REQUEST; @@ -55,13 +62,47 @@ import static org.thingsboard.server.common.data.msg.TbMsgType.POST_ATTRIBUTES_R type = ComponentType.ACTION, name = "save attributes", configClazz = TbMsgAttributesNodeConfiguration.class, - version = 2, - nodeDescription = "Saves attributes data", - nodeDetails = "Saves entity attributes based on configurable scope parameter. Expects messages with 'POST_ATTRIBUTES_REQUEST' message type. " + - "If upsert(update/insert) operation is completed successfully rule node will send the incoming message via Success chain, otherwise, Failure chain is used. " + - "Additionally if checkbox Send attributes updated notification is set to true, rule node will put the \"Attributes Updated\" " + - "event for SHARED_SCOPE and SERVER_SCOPE attributes updates to the corresponding rule engine queue." + - "Performance checkbox 'Save attributes only if the value changes' will skip attributes overwrites for values with no changes (avoid concurrent writes because this check is not transactional; will not update 'Last updated time' for skipped attributes).", + version = 3, + nodeDescription = """ + Saves attribute data with a configurable scope and according to configured processing strategies. + """, + nodeDetails = """ + Node performs two actions: +
    +
  • Attributes: save attribute data to a database.
    • +
  • WebSockets: notify WebSockets subscriptions about attribute data updates.
    • +
+ + For each action, three processing strategies are available: +
    +
  • On every message: perform the action for every message.
    • +
  • Deduplicate: perform the action only for the first message from a particular originator within a configurable interval.
    • +
  • Skip: never perform the action.
    • +
+ + Processing strategies are configured using processing settings, which support two modes: +
    +
  • Basic +
      +
    • On every message: applies the "On every message" strategy to all actions.
      • +
    • Deduplicate: applies the "Deduplicate" strategy (with a specified interval) to all actions.
      • +
    • WebSockets only: applies the "Skip" strategy to Attributes, and the "On every message" strategy to WebSockets.
      • +
    +
    • +
  • Advanced: configure each action’s strategy independently.
    • +
+ + Additionally: +
    +
  • If Save attributes only if the value changes is enabled, the rule node compares the received attribute value with the current stored value and skips the save operation if they match.
    • +
  • If Send attributes updated notification is enabled, the rule node will put the Attributes Updated event for SHARED_SCOPE and SERVER_SCOPE attribute updates to the queue named Main.
    • +
  • If Force notification to the device is enabled, then rule node will always notify device about SHARED_SCOPE attribute updates, regardless of the value of notifyDevice metadata property.
    • +
+ + This node expects messages of type POST_ATTRIBUTES_REQUEST. +

+ Output connections: Success, Failure. + """, configDirective = "tbActionNodeAttributesConfig", icon = "file_upload" ) @@ -73,9 +114,12 @@ public class TbMsgAttributesNode implements TbNode { private TbMsgAttributesNodeConfiguration config; + private AttributesProcessingSettings processingSettings; + @Override public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { - this.config = TbNodeUtils.convert(configuration, TbMsgAttributesNodeConfiguration.class); + config = TbNodeUtils.convert(configuration, TbMsgAttributesNodeConfiguration.class); + processingSettings = config.getProcessingSettings(); } @Override @@ -90,11 +134,20 @@ public class TbMsgAttributesNode implements TbNode { ctx.tellSuccess(msg); return; } + + AttributesSaveRequest.Strategy strategy = determineSaveStrategy(msg.getMetaDataTs(), msg.getOriginator().getId()); + + // short-circuit + if (!strategy.saveAttributes() && !strategy.sendWsUpdate() && !strategy.processCalculatedFields()) { + ctx.tellSuccess(msg); + return; + } + AttributeScope scope = getScope(msg.getMetaData().getValue(SCOPE)); boolean sendAttributesUpdateNotification = checkSendNotification(scope); if (!config.isUpdateAttributesOnlyOnValueChange()) { - saveAttr(newAttributes, ctx, msg, scope, sendAttributesUpdateNotification); + saveAttr(newAttributes, ctx, msg, scope, sendAttributesUpdateNotification, strategy); return; } @@ -104,13 +157,42 @@ public class TbMsgAttributesNode implements TbNode { DonAsynchron.withCallback(findFuture, currentAttributes -> { List attributesChanged = filterChangedAttr(currentAttributes, newAttributes); - saveAttr(attributesChanged, ctx, msg, scope, sendAttributesUpdateNotification); + saveAttr(attributesChanged, ctx, msg, scope, sendAttributesUpdateNotification, strategy); }, throwable -> ctx.tellFailure(msg, throwable), MoreExecutors.directExecutor()); } - void saveAttr(List attributes, TbContext ctx, TbMsg msg, AttributeScope scope, boolean sendAttributesUpdateNotification) { + private AttributesSaveRequest.Strategy determineSaveStrategy(long ts, UUID originatorUuid) { + if (processingSettings instanceof OnEveryMessage) { + return AttributesSaveRequest.Strategy.PROCESS_ALL; + } + if (processingSettings instanceof WebSocketsOnly) { + return AttributesSaveRequest.Strategy.WS_ONLY; + } + if (processingSettings instanceof Deduplicate deduplicate) { + boolean isFirstMsgInInterval = deduplicate.getProcessingStrategy().shouldProcess(ts, originatorUuid); + return isFirstMsgInInterval ? AttributesSaveRequest.Strategy.PROCESS_ALL : AttributesSaveRequest.Strategy.SKIP_ALL; + } + if (processingSettings instanceof Advanced advanced) { + return new AttributesSaveRequest.Strategy( + advanced.attributes().shouldProcess(ts, originatorUuid), + advanced.webSockets().shouldProcess(ts, originatorUuid), + advanced.calculatedFields().shouldProcess(ts, originatorUuid) + ); + } + // should not happen + throw new IllegalArgumentException("Unknown processing settings type: " + processingSettings.getClass().getSimpleName()); + } + + private void saveAttr( + List attributes, + TbContext ctx, + TbMsg msg, + AttributeScope scope, + boolean sendAttributesUpdateNotification, + AttributesSaveRequest.Strategy strategy + ) { if (attributes.isEmpty()) { ctx.tellSuccess(msg); return; @@ -124,6 +206,7 @@ public class TbMsgAttributesNode implements TbNode { .scope(scope) .entries(attributes) .notifyDevice(config.isNotifyDevice() || checkNotifyDeviceMdValue(msg.getMetaData().getValue(NOTIFY_DEVICE_METADATA_KEY))) + .strategy(strategy) .previousCalculatedFieldIds(msg.getPreviousCalculatedFieldIds()) .tbMsgId(msg.getId()) .tbMsgType(msg.getInternalType()) @@ -131,7 +214,7 @@ public class TbMsgAttributesNode implements TbNode { .build()); } - List filterChangedAttr(List currentAttributes, List newAttributes) { + private List filterChangedAttr(List currentAttributes, List newAttributes) { if (currentAttributes == null || currentAttributes.isEmpty()) { return newAttributes; } @@ -181,6 +264,9 @@ public class TbMsgAttributesNode implements TbNode { hasChanges = fixEscapedBooleanConfigParameter(oldConfiguration, SEND_ATTRIBUTES_UPDATED_NOTIFICATION_KEY, hasChanges, false); // update updateAttributesOnlyOnValueChange. hasChanges = fixEscapedBooleanConfigParameter(oldConfiguration, UPDATE_ATTRIBUTES_ONLY_ON_VALUE_CHANGE_KEY, hasChanges, true); + case 2: + hasChanges = true; + ((ObjectNode) oldConfiguration).set("processingSettings", JacksonUtil.valueToTree(new OnEveryMessage())); break; default: break; diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeConfiguration.java index 161aa64d5f..2687125a0c 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeConfiguration.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeConfiguration.java @@ -15,13 +15,20 @@ */ package org.thingsboard.rule.engine.telemetry; +import jakarta.validation.constraints.NotNull; import lombok.Data; import org.thingsboard.rule.engine.api.NodeConfiguration; +import org.thingsboard.rule.engine.telemetry.settings.AttributesProcessingSettings; import org.thingsboard.server.common.data.DataConstants; +import static org.thingsboard.rule.engine.telemetry.settings.AttributesProcessingSettings.OnEveryMessage; + @Data public class TbMsgAttributesNodeConfiguration implements NodeConfiguration { + @NotNull + private AttributesProcessingSettings processingSettings; + private String scope; private boolean notifyDevice; @@ -31,6 +38,7 @@ public class TbMsgAttributesNodeConfiguration implements NodeConfiguration { @@ -39,7 +28,7 @@ public class TbMsgTimeseriesNodeConfiguration implements NodeConfiguration newAttributes = new ArrayList<>(); + void givenProcessingSettingsAreNull_whenValidatingConstraints_thenThrowsException() { + // GIVEN + config.setProcessingSettings(null); - List filtered = node.filterChangedAttr(Collections.emptyList(), newAttributes); - assertThat(filtered).isSameAs(newAttributes); + // WHEN-THEN + assertThatThrownBy(() -> ConstraintValidator.validateFields(config)) + .isInstanceOf(DataValidationException.class) + .hasMessage("Validation error: processingSettings must not be null"); } @Test - void testFilterChangedAttr_whenCurrentAttributesContainsInAnyOrderNewAttributes_thenReturnEmptyList() { - List currentAttributes = List.of( - new BaseAttributeKvEntry(1694000000L, new StringDataEntry("address", "Peremohy ave 1")), - new BaseAttributeKvEntry(1694000000L, new BooleanDataEntry("valid", true)), - new BaseAttributeKvEntry(1694000000L, new LongDataEntry("counter", 100L)), - new BaseAttributeKvEntry(1694000000L, new DoubleDataEntry("temp", -18.35)), - new BaseAttributeKvEntry(1694000000L, new JsonDataEntry("json", "{\"warning\":\"out of paper\"}")) - ); - List newAttributes = new ArrayList<>(currentAttributes); - newAttributes.add(newAttributes.get(0)); - newAttributes.remove(0); - assertThat(newAttributes).hasSize(currentAttributes.size()); - assertThat(currentAttributes).isNotEmpty(); - assertThat(newAttributes).containsExactlyInAnyOrderElementsOf(currentAttributes); - - List filtered = node.filterChangedAttr(currentAttributes, newAttributes); - assertThat(filtered).isEmpty(); //no changes + void givenOnEveryMessageProcessingSettingsAndSameMessageTwoTimes_whenOnMsg_thenPersistSameMessageTwoTimes() throws TbNodeException { + // GIVEN + config.setUpdateAttributesOnlyOnValueChange(false); + config.setProcessingSettings(new OnEveryMessage()); + + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); + + var msg = TbMsg.newMsg() + .type(TbMsgType.POST_ATTRIBUTES_REQUEST) + .originator(deviceId) + .data(JacksonUtil.newObjectNode().put("temperature", 22.3).toString()) + .metaData(new TbMsgMetaData(Map.of(NOTIFY_DEVICE_METADATA_KEY, "false"))) + .build(); + + // WHEN-THEN + var expectedSaveRequest = builder() + .tenantId(tenantId) + .entityId(msg.getOriginator()) + .scope(AttributeScope.valueOf(config.getScope())) + .entry(new DoubleDataEntry("temperature", 22.3)) + .notifyDevice(false) + .strategy(Strategy.PROCESS_ALL) + .previousCalculatedFieldIds(msg.getPreviousCalculatedFieldIds()) + .tbMsgId(msg.getId()) + .tbMsgType(msg.getInternalType()) + .build(); + + node.onMsg(ctxMock, msg); + then(telemetryServiceMock).should(times(1)).saveAttributes(assertArg( + actualSaveRequest -> assertThat(actualSaveRequest) + .usingRecursiveComparison() + .ignoringFields("callback", "entries.lastUpdateTs") + .isEqualTo(expectedSaveRequest) + )); + + node.onMsg(ctxMock, msg); + then(telemetryServiceMock).should(times(2)).saveAttributes(assertArg( + actualSaveRequest -> assertThat(actualSaveRequest) + .usingRecursiveComparison() + .ignoringFields("callback", "entries.lastUpdateTs") + .isEqualTo(expectedSaveRequest) + )); + } + + @Test + void givenDeduplicateProcessingSettingsAndSameMessageTwoTimes_whenOnMsg_thenPersistThisMessageOnlyFirstTime() throws TbNodeException { + // GIVEN + config.setUpdateAttributesOnlyOnValueChange(false); + config.setProcessingSettings(new Deduplicate(10)); + + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); + + var msg = TbMsg.newMsg() + .type(TbMsgType.POST_ATTRIBUTES_REQUEST) + .originator(deviceId) + .data(JacksonUtil.newObjectNode().put("temperature", 22.3).toString()) + .metaData(new TbMsgMetaData(Map.of(NOTIFY_DEVICE_METADATA_KEY, "false"))) + .build(); + + // WHEN-THEN + var expectedSaveRequest = builder() + .tenantId(tenantId) + .entityId(msg.getOriginator()) + .scope(AttributeScope.valueOf(config.getScope())) + .entry(new DoubleDataEntry("temperature", 22.3)) + .notifyDevice(false) + .strategy(Strategy.PROCESS_ALL) + .previousCalculatedFieldIds(msg.getPreviousCalculatedFieldIds()) + .tbMsgId(msg.getId()) + .tbMsgType(msg.getInternalType()) + .build(); + + node.onMsg(ctxMock, msg); + then(telemetryServiceMock).should().saveAttributes(assertArg( + actualSaveRequest -> assertThat(actualSaveRequest) + .usingRecursiveComparison() + .ignoringFields("callback", "entries.lastUpdateTs") + .isEqualTo(expectedSaveRequest) + )); + + clearInvocations(telemetryServiceMock, ctxMock); + + node.onMsg(ctxMock, msg); + then(telemetryServiceMock).should(never()).saveAttributes(any()); + } + + @Test + void givenWebSocketsOnlyProcessingSettingsAndSameMessageTwoTimes_whenOnMsg_thenSendsOnlyWsUpdateTwoTimes() throws TbNodeException { + // GIVEN + config.setUpdateAttributesOnlyOnValueChange(false); + config.setProcessingSettings(new WebSocketsOnly()); + + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); + + var msg = TbMsg.newMsg() + .type(TbMsgType.POST_ATTRIBUTES_REQUEST) + .originator(deviceId) + .data(JacksonUtil.newObjectNode().put("temperature", 22.3).toString()) + .metaData(new TbMsgMetaData(Map.of(NOTIFY_DEVICE_METADATA_KEY, "false"))) + .build(); + + // WHEN-THEN + var expectedSaveRequest = builder() + .tenantId(tenantId) + .entityId(msg.getOriginator()) + .scope(AttributeScope.valueOf(config.getScope())) + .entry(new DoubleDataEntry("temperature", 22.3)) + .notifyDevice(false) + .strategy(Strategy.WS_ONLY) + .previousCalculatedFieldIds(msg.getPreviousCalculatedFieldIds()) + .tbMsgId(msg.getId()) + .tbMsgType(msg.getInternalType()) + .build(); + + node.onMsg(ctxMock, msg); + then(telemetryServiceMock).should(times(1)).saveAttributes(assertArg( + actualSaveRequest -> assertThat(actualSaveRequest) + .usingRecursiveComparison() + .ignoringFields("callback", "entries.lastUpdateTs") + .isEqualTo(expectedSaveRequest) + )); + + node.onMsg(ctxMock, msg); + then(telemetryServiceMock).should(times(2)).saveAttributes(assertArg( + actualSaveRequest -> assertThat(actualSaveRequest) + .usingRecursiveComparison() + .ignoringFields("callback", "entries.lastUpdateTs") + .isEqualTo(expectedSaveRequest) + )); + } + + @Test + void givenAdvancedProcessingSettingsWithOnEveryMessageStrategiesForAllActionsAndSameMessageTwoTimes_whenOnMsg_thenPersistSameMessageTwoTimes() throws TbNodeException { + // GIVEN + config.setUpdateAttributesOnlyOnValueChange(false); + config.setProcessingSettings(new Advanced( + ProcessingStrategy.onEveryMessage(), + ProcessingStrategy.onEveryMessage(), + ProcessingStrategy.onEveryMessage() + )); + + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); + + var msg = TbMsg.newMsg() + .type(TbMsgType.POST_ATTRIBUTES_REQUEST) + .originator(deviceId) + .data(JacksonUtil.newObjectNode().put("temperature", 22.3).toString()) + .metaData(new TbMsgMetaData(Map.of(NOTIFY_DEVICE_METADATA_KEY, "false"))) + .build(); + + // WHEN-THEN + var expectedSaveRequest = builder() + .tenantId(tenantId) + .entityId(msg.getOriginator()) + .scope(AttributeScope.valueOf(config.getScope())) + .entry(new DoubleDataEntry("temperature", 22.3)) + .notifyDevice(false) + .strategy(Strategy.PROCESS_ALL) + .previousCalculatedFieldIds(msg.getPreviousCalculatedFieldIds()) + .tbMsgId(msg.getId()) + .tbMsgType(msg.getInternalType()) + .build(); + + node.onMsg(ctxMock, msg); + then(telemetryServiceMock).should(times(1)).saveAttributes(assertArg( + actualSaveRequest -> assertThat(actualSaveRequest) + .usingRecursiveComparison() + .ignoringFields("callback", "entries.lastUpdateTs") + .isEqualTo(expectedSaveRequest) + )); + + node.onMsg(ctxMock, msg); + then(telemetryServiceMock).should(times(2)).saveAttributes(assertArg( + actualSaveRequest -> assertThat(actualSaveRequest) + .usingRecursiveComparison() + .ignoringFields("callback", "entries.lastUpdateTs") + .isEqualTo(expectedSaveRequest) + )); + } + + @Test + void givenAdvancedProcessingSettingsWithDifferentDeduplicateStrategyForEachAction_whenOnMsg_thenEvaluatesStrategiesForEachActionsIndependently() throws TbNodeException { + // GIVEN + config.setUpdateAttributesOnlyOnValueChange(false); + config.setProcessingSettings(new Advanced( + ProcessingStrategy.deduplicate(1), + ProcessingStrategy.deduplicate(2), + ProcessingStrategy.deduplicate(3) + )); + + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); + + long ts1 = 500L; + long ts2 = 1500L; + long ts3 = 2500L; + long ts4 = 3500L; + + // WHEN-THEN + node.onMsg(ctxMock, TbMsg.newMsg() + .type(TbMsgType.POST_ATTRIBUTES_REQUEST) + .originator(deviceId) + .data(JacksonUtil.newObjectNode().put("temperature", 22.3).toString()) + .metaData(new TbMsgMetaData(Map.of("ts", Long.toString(ts1)))) + .build()); + then(telemetryServiceMock).should().saveAttributes(assertArg( + actualSaveRequest -> assertThat(actualSaveRequest.getStrategy()).isEqualTo(Strategy.PROCESS_ALL) + )); + + clearInvocations(telemetryServiceMock); + + node.onMsg(ctxMock, TbMsg.newMsg() + .type(TbMsgType.POST_ATTRIBUTES_REQUEST) + .originator(deviceId) + .data(JacksonUtil.newObjectNode().put("temperature", 22.3).toString()) + .metaData(new TbMsgMetaData(Map.of("ts", Long.toString(ts2)))) + .build()); + then(telemetryServiceMock).should().saveAttributes(assertArg( + actualSaveRequest -> assertThat(actualSaveRequest.getStrategy()).isEqualTo(new Strategy(true, false, false)) + )); + + clearInvocations(telemetryServiceMock); + + node.onMsg(ctxMock, TbMsg.newMsg() + .type(TbMsgType.POST_ATTRIBUTES_REQUEST) + .originator(deviceId) + .data(JacksonUtil.newObjectNode().put("temperature", 22.3).toString()) + .metaData(new TbMsgMetaData(Map.of("ts", Long.toString(ts3)))) + .build()); + then(telemetryServiceMock).should().saveAttributes(assertArg( + actualSaveRequest -> assertThat(actualSaveRequest.getStrategy()).isEqualTo(new Strategy(true, true, false)) + )); + + clearInvocations(telemetryServiceMock); + + node.onMsg(ctxMock, TbMsg.newMsg() + .type(TbMsgType.POST_ATTRIBUTES_REQUEST) + .originator(deviceId) + .data(JacksonUtil.newObjectNode().put("temperature", 22.3).toString()) + .metaData(new TbMsgMetaData(Map.of("ts", Long.toString(ts4)))) + .build()); + then(telemetryServiceMock).should().saveAttributes(assertArg( + actualSaveRequest -> assertThat(actualSaveRequest.getStrategy()).isEqualTo(new Strategy(true, false, true)) + )); } @Test - void testFilterChangedAttr_whenCurrentAttributesContainsInAnyOrderNewAttributes_thenReturnExpectedList() { + public void givenAdvancedProcessingSettingsWithSkipStrategiesForAllActionsAndSameMessageTwoTimes_whenOnMsg_thenSkipsSameMessageTwoTimes() throws TbNodeException { + // GIVEN + config.setProcessingSettings(new Advanced( + ProcessingStrategy.skip(), + ProcessingStrategy.skip(), + ProcessingStrategy.skip() + )); + + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); + + var msg = TbMsg.newMsg() + .type(TbMsgType.POST_ATTRIBUTES_REQUEST) + .originator(deviceId) + .data(JacksonUtil.newObjectNode().put("temperature", 22.3).toString()) + .metaData(new TbMsgMetaData(Map.of(NOTIFY_DEVICE_METADATA_KEY, "false"))) + .build(); + + // WHEN-THEN + node.onMsg(ctxMock, msg); + then(telemetryServiceMock).should(never()).saveAttributes(any()); + then(ctxMock).should(times(1)).tellSuccess(msg); + + node.onMsg(ctxMock, msg); + then(telemetryServiceMock).should(never()).saveAttributes(any()); + then(ctxMock).should(times(2)).tellSuccess(msg); + } + + @Test + void givenVariousChangesToAttributes_whenUpdateOnlyOnValueChangeEnabled_thenShouldCorrectlyFilterChangedAttributes() throws TbNodeException { + // GIVEN + config.setUpdateAttributesOnlyOnValueChange(true); + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); + List currentAttributes = List.of( - new BaseAttributeKvEntry(1694000000L, new StringDataEntry("address", "Peremohy ave 1")), - new BaseAttributeKvEntry(1694000000L, new BooleanDataEntry("valid", true)), - new BaseAttributeKvEntry(1694000000L, new LongDataEntry("counter", 100L)), - new BaseAttributeKvEntry(1694000000L, new DoubleDataEntry("temp", -18.35)), - new BaseAttributeKvEntry(1694000000L, new JsonDataEntry("json", "{\"warning\":\"out of paper\"}")) + new BaseAttributeKvEntry(123L, new StringDataEntry("address", "Prospect Beresteiskyi 1")), + new BaseAttributeKvEntry(123L, new BooleanDataEntry("valid", true)), + new BaseAttributeKvEntry(123L, new LongDataEntry("counter", 100L)), + new BaseAttributeKvEntry(123L, new DoubleDataEntry("temp", -18.35)), + new BaseAttributeKvEntry(123L, new JsonDataEntry("json", "{\"warning\":\"out of paper\"}")) ); - List newAttributes = List.of( - new BaseAttributeKvEntry(1694000999L, new JsonDataEntry("json", "{\"status\":\"OK\"}")), // value changed, reordered - new BaseAttributeKvEntry(1694000999L, new StringDataEntry("valid", "true")), //type changed - new BaseAttributeKvEntry(1694000999L, new LongDataEntry("counter", 101L)), //value changed - new BaseAttributeKvEntry(1694000999L, new DoubleDataEntry("temp", -18.35)), - new BaseAttributeKvEntry(1694000999L, new StringDataEntry("address", "Peremohy ave 1")) // reordered - ); - List expected = List.of( - new BaseAttributeKvEntry(1694000999L, new StringDataEntry("valid", "true")), - new BaseAttributeKvEntry(1694000999L, new LongDataEntry("counter", 101L)), - new BaseAttributeKvEntry(1694000999L, new JsonDataEntry("json", "{\"status\":\"OK\"}")) + given(attributesServiceMock.find(eq(tenantId), eq(deviceId), eq(AttributeScope.valueOf(config.getScope())), anyList())).willReturn(immediateFuture(currentAttributes)); + + var data = JacksonUtil.newObjectNode() + .put("address", "Prospect Beresteiskyi 1") // no changes + .put("valid", "false") // type and value changed + .put("counter", 101L) // value changed + .put("temp", -18.35) // no changes + .put("json", "{\"warning\":\"out of paper\"}") // only type changed + .put("newKey", "newValue"); // new attribute + + var msg = TbMsg.newMsg() + .type(TbMsgType.POST_ATTRIBUTES_REQUEST) + .originator(deviceId) + .data(data.toString()) + .metaData(TbMsgMetaData.EMPTY) + .build(); + + // WHEN + node.onMsg(ctxMock, msg); + + // THEN + List expectedChangedAttributes = List.of( + new BaseAttributeKvEntry(456L, new StringDataEntry("valid", "false")), + new BaseAttributeKvEntry(456L, new LongDataEntry("counter", 101L)), + new BaseAttributeKvEntry(456L, new StringDataEntry("json", "{\"warning\":\"out of paper\"}")), + new BaseAttributeKvEntry(456L, new StringDataEntry("newKey", "newValue")) ); - List filtered = node.filterChangedAttr(currentAttributes, newAttributes); - assertThat(filtered).containsExactlyInAnyOrderElementsOf(expected); + then(telemetryServiceMock).should().saveAttributes(assertArg(request -> + assertThat(request.getEntries()) + .usingRecursiveComparison() + .ignoringCollectionOrder() + .ignoringFields("lastUpdateTs") + .isEqualTo(expectedChangedAttributes) + )); } - // Notify device backward-compatibility test arguments - private static Stream givenNotifyDeviceMdValue_whenSaveAndNotify_thenVerifyExpectedArgumentForNotifyDeviceInSaveAndNotifyMethod() { - return Stream.of( - Arguments.of(null, true), - Arguments.of("null", false), - Arguments.of("true", true), - Arguments.of("false", false) + @Test + void givenNoChangesToAttributes_whenUpdateOnlyOnValueChangeEnabled_thenShouldNotCallSaveAndJustTellSuccess() throws TbNodeException { + // GIVEN + config.setUpdateAttributesOnlyOnValueChange(true); + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); + + List currentAttributes = List.of( + new BaseAttributeKvEntry(123L, new StringDataEntry("address", "Prospect Beresteiskyi 1")), + new BaseAttributeKvEntry(123L, new BooleanDataEntry("valid", true)), + new BaseAttributeKvEntry(123L, new LongDataEntry("counter", 100L)) ); + given(attributesServiceMock.find(eq(tenantId), eq(deviceId), eq(AttributeScope.valueOf(config.getScope())), anyList())).willReturn(immediateFuture(currentAttributes)); + + var data = JacksonUtil.newObjectNode() + .put("address", "Prospect Beresteiskyi 1") + .put("valid", true) + .put("counter", 100L); + + var msg = TbMsg.newMsg() + .type(TbMsgType.POST_ATTRIBUTES_REQUEST) + .originator(deviceId) + .data(data.toString()) + .metaData(TbMsgMetaData.EMPTY) + .build(); + + // WHEN + node.onMsg(ctxMock, msg); + + // THEN + then(telemetryServiceMock).shouldHaveNoInteractions(); + then(ctxMock).should().tellSuccess(msg); } // Notify device backward-compatibility test @ParameterizedTest @MethodSource - void givenNotifyDeviceMdValue_whenSaveAndNotify_thenVerifyExpectedArgumentForNotifyDeviceInSaveAndNotifyMethod(String mdValue, boolean expectedArgumentValue) throws TbNodeException { - var ctxMock = mock(TbContext.class); - var telemetryServiceMock = mock(RuleEngineTelemetryService.class); - ObjectNode defaultConfig = (ObjectNode) JacksonUtil.valueToTree(new TbMsgAttributesNodeConfiguration().defaultConfiguration()); - defaultConfig.put("notifyDevice", false); - var tbNodeConfiguration = new TbNodeConfiguration(defaultConfig); - - assertThat(defaultConfig.has("notifyDevice")).as("pre condition has notifyDevice").isTrue(); - - when(ctxMock.getTenantId()).thenReturn(tenantId); - when(ctxMock.getTelemetryService()).thenReturn(telemetryServiceMock); - willCallRealMethod().given(node).init(any(TbContext.class), any(TbNodeConfiguration.class)); - willCallRealMethod().given(node).saveAttr(any(), eq(ctxMock), any(TbMsg.class), any(AttributeScope.class), anyBoolean()); - - node.init(ctxMock, tbNodeConfiguration); - - TbMsgMetaData md = new TbMsgMetaData(); - if (mdValue != null) { - md.putValue(NOTIFY_DEVICE_METADATA_KEY, mdValue); - } - // dummy list with one ts kv to pass the empty list check. - var testTbMsg = TbMsg.newMsg() - .type(TbMsgType.POST_TELEMETRY_REQUEST) + void givenVariousValuesForNotifyDeviceInMetadata_thenShouldCorrectlyParseValueFromMetadata(String mdValue, boolean expectedArgumentValue) throws TbNodeException { + // GIVEN + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); + + given(attributesServiceMock.find(tenantId, deviceId, AttributeScope.valueOf(config.getScope()), List.of("mode"))).willReturn( + immediateFuture(List.of(new BaseAttributeKvEntry(123L, new StringDataEntry("mode", "tilt")))) + ); + + var metadata = new TbMsgMetaData(); + metadata.putValue(NOTIFY_DEVICE_METADATA_KEY, mdValue); + + var msg = TbMsg.newMsg() + .type(TbMsgType.POST_ATTRIBUTES_REQUEST) .originator(deviceId) - .copyMetaData(md) - .data(TbMsg.EMPTY_STRING) + .data(JacksonUtil.newObjectNode().put("mode", "vibration").toString()) + .metaData(metadata) .build(); - List testAttrList = List.of(new BaseAttributeKvEntry(0L, new StringDataEntry("testKey", "testValue"))); - node.saveAttr(testAttrList, ctxMock, testTbMsg, AttributeScope.SHARED_SCOPE, false); + // WHEN + node.onMsg(ctxMock, msg); + + // THEN + then(telemetryServiceMock).should().saveAttributes(assertArg(request -> assertThat(request.isNotifyDevice()).isEqualTo(expectedArgumentValue))); + } - verify(telemetryServiceMock, times(1)).saveAttributes(assertArg(request -> { - assertThat(request.getTenantId()).isEqualTo(tenantId); - assertThat(request.getEntityId()).isEqualTo(deviceId); - assertThat(request.getScope()).isEqualTo(AttributeScope.SHARED_SCOPE); - assertThat(request.getEntries()).isEqualTo(testAttrList); - assertThat(request.isNotifyDevice()).isEqualTo(expectedArgumentValue); - })); + // Notify device backward-compatibility test arguments + static Stream givenVariousValuesForNotifyDeviceInMetadata_thenShouldCorrectlyParseValueFromMetadata() { + return Stream.of( + Arguments.of(null, true), + Arguments.of("null", false), + Arguments.of("true", true), + Arguments.of("false", false) + ); } // Rule nodes upgrade - private static Stream givenFromVersionAndConfig_whenUpgrade_thenVerifyHasChangesAndConfig() { + static Stream givenFromVersionAndConfig_whenUpgrade_thenVerifyHasChangesAndConfig() { return Stream.of( // default config for version 0 Arguments.of(0, - "{\"scope\":\"CLIENT_SCOPE\",\"notifyDevice\":\"false\",\"sendAttributesUpdatedNotification\":\"false\"}", + """ + { + "scope": "CLIENT_SCOPE", + "notifyDevice": "false", + "sendAttributesUpdatedNotification": "false" + } + """, true, - "{\"scope\":\"CLIENT_SCOPE\",\"notifyDevice\":false,\"sendAttributesUpdatedNotification\":false,\"updateAttributesOnlyOnValueChange\":false}"), + """ + { + "processingSettings": { + "type": "ON_EVERY_MESSAGE" + }, + "scope": "CLIENT_SCOPE", + "notifyDevice": false, + "sendAttributesUpdatedNotification": false, + "updateAttributesOnlyOnValueChange": false + } + """ + ), // default config for version 1 with upgrade from version 0 Arguments.of(0, - "{\"scope\":\"CLIENT_SCOPE\",\"notifyDevice\":false,\"sendAttributesUpdatedNotification\":false,\"updateAttributesOnlyOnValueChange\":true}", - false, - "{\"scope\":\"CLIENT_SCOPE\",\"notifyDevice\":false,\"sendAttributesUpdatedNotification\":false,\"updateAttributesOnlyOnValueChange\":true}"), + """ + { + "scope": "CLIENT_SCOPE", + "notifyDevice": false, + "sendAttributesUpdatedNotification": false, + "updateAttributesOnlyOnValueChange": true + } + """, + true, + """ + { + "processingSettings": { + "type": "ON_EVERY_MESSAGE" + }, + "scope": "CLIENT_SCOPE", + "notifyDevice": false, + "sendAttributesUpdatedNotification": false, + "updateAttributesOnlyOnValueChange": true + } + """ + ), // all flags are booleans Arguments.of(1, - "{\"scope\":\"SHARED_SCOPE\",\"notifyDevice\":true,\"sendAttributesUpdatedNotification\":false,\"updateAttributesOnlyOnValueChange\":true}", - false, - "{\"scope\":\"SHARED_SCOPE\",\"notifyDevice\":true,\"sendAttributesUpdatedNotification\":false,\"updateAttributesOnlyOnValueChange\":true}"), + """ + { + "scope": "SHARED_SCOPE", + "notifyDevice": true, + "sendAttributesUpdatedNotification": false, + "updateAttributesOnlyOnValueChange": true + } + """, + true, + """ + { + "processingSettings": { + "type": "ON_EVERY_MESSAGE" + }, + "scope": "SHARED_SCOPE", + "notifyDevice": true, + "sendAttributesUpdatedNotification": false, + "updateAttributesOnlyOnValueChange": true + } + """ + ), // no boolean flags set Arguments.of(1, - "{\"scope\":\"CLIENT_SCOPE\"}", + """ + { + "scope": "CLIENT_SCOPE" + } + """, true, - "{\"scope\":\"CLIENT_SCOPE\",\"notifyDevice\":true,\"sendAttributesUpdatedNotification\":false,\"updateAttributesOnlyOnValueChange\":true}"), + """ + { + "processingSettings": { + "type": "ON_EVERY_MESSAGE" + }, + "scope": "CLIENT_SCOPE", + "notifyDevice": true, + "sendAttributesUpdatedNotification": false, + "updateAttributesOnlyOnValueChange": true + } + """ + ), // all flags are boolean strings Arguments.of(1, - "{\"scope\":\"CLIENT_SCOPE\",\"notifyDevice\":\"false\",\"sendAttributesUpdatedNotification\":\"false\",\"updateAttributesOnlyOnValueChange\":\"true\"}", + """ + { + "scope": "CLIENT_SCOPE", + "notifyDevice": "false", + "sendAttributesUpdatedNotification": "false", + "updateAttributesOnlyOnValueChange": "true" + } + """, true, - "{\"scope\":\"CLIENT_SCOPE\",\"notifyDevice\":false,\"sendAttributesUpdatedNotification\":false,\"updateAttributesOnlyOnValueChange\":true}"), + """ + { + "processingSettings": { + "type": "ON_EVERY_MESSAGE" + }, + "scope": "CLIENT_SCOPE", + "notifyDevice": false, + "sendAttributesUpdatedNotification": false, + "updateAttributesOnlyOnValueChange": true + } + """ + ), // at least one flag is boolean string Arguments.of(1, - "{\"scope\":\"CLIENT_SCOPE\",\"notifyDevice\":\"false\",\"sendAttributesUpdatedNotification\":false,\"updateAttributesOnlyOnValueChange\":true}", + """ + { + "scope": "CLIENT_SCOPE", + "notifyDevice": "false", + "sendAttributesUpdatedNotification": false, + "updateAttributesOnlyOnValueChange": true + } + """, true, - "{\"scope\":\"CLIENT_SCOPE\",\"notifyDevice\":false,\"sendAttributesUpdatedNotification\":false,\"updateAttributesOnlyOnValueChange\":true}"), + """ + { + "processingSettings": { + "type": "ON_EVERY_MESSAGE" + }, + "scope": "CLIENT_SCOPE", + "notifyDevice": false, + "sendAttributesUpdatedNotification": false, + "updateAttributesOnlyOnValueChange": true + } + """ + ), // notify device flag is null Arguments.of(1, - "{\"scope\":\"CLIENT_SCOPE\",\"notifyDevice\":\"null\",\"sendAttributesUpdatedNotification\":false,\"updateAttributesOnlyOnValueChange\":true}", + """ + { + "scope": "CLIENT_SCOPE", + "notifyDevice": "null", + "sendAttributesUpdatedNotification": false, + "updateAttributesOnlyOnValueChange": true + } + """, + true, + """ + { + "processingSettings": { + "type": "ON_EVERY_MESSAGE" + }, + "scope": "CLIENT_SCOPE", + "notifyDevice": true, + "sendAttributesUpdatedNotification": false, + "updateAttributesOnlyOnValueChange": true + } + """ + ), + // default config for version 2 + Arguments.of(2, + """ + { + "scope": "SERVER_SCOPE", + "notifyDevice": false, + "sendAttributesUpdatedNotification": false, + "updateAttributesOnlyOnValueChange": true + } + """, true, - "{\"scope\":\"CLIENT_SCOPE\",\"notifyDevice\":true,\"sendAttributesUpdatedNotification\":false,\"updateAttributesOnlyOnValueChange\":true}") + """ + { + "processingSettings": { + "type": "ON_EVERY_MESSAGE" + }, + "scope": "SERVER_SCOPE", + "notifyDevice": false, + "sendAttributesUpdatedNotification": false, + "updateAttributesOnlyOnValueChange": true + } + """ + ) ); } diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeTest.java index ea8167ca18..1141ff09d1 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeTest.java @@ -73,6 +73,10 @@ import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoMoreInteractions; +import static org.thingsboard.rule.engine.telemetry.settings.TimeseriesProcessingSettings.Advanced; +import static org.thingsboard.rule.engine.telemetry.settings.TimeseriesProcessingSettings.Deduplicate; +import static org.thingsboard.rule.engine.telemetry.settings.TimeseriesProcessingSettings.OnEveryMessage; +import static org.thingsboard.rule.engine.telemetry.settings.TimeseriesProcessingSettings.WebSocketsOnly; @ExtendWith(MockitoExtension.class) public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest { @@ -110,7 +114,7 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest { @Test public void verifyDefaultConfig() { assertThat(config.getDefaultTTL()).isEqualTo(0L); - assertThat(config.getProcessingSettings()).isInstanceOf(TbMsgTimeseriesNodeConfiguration.ProcessingSettings.OnEveryMessage.class); + assertThat(config.getProcessingSettings()).isInstanceOf(OnEveryMessage.class); assertThat(config.isUseServerTs()).isFalse(); } @@ -220,11 +224,11 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest { // GIVEN config.setDefaultTTL(10L); - var timeseriesStrategy = ProcessingStrategy.onEveryMessage(); - var latestStrategy = ProcessingStrategy.skip(); + var timeseries = ProcessingStrategy.onEveryMessage(); + var latest = ProcessingStrategy.skip(); var webSockets = ProcessingStrategy.onEveryMessage(); var calculatedFields = ProcessingStrategy.onEveryMessage(); - var processingSettings = new TbMsgTimeseriesNodeConfiguration.ProcessingSettings.Advanced(timeseriesStrategy, latestStrategy, webSockets, calculatedFields); + var processingSettings = new Advanced(timeseries, latest, webSockets, calculatedFields); config.setProcessingSettings(processingSettings); node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); @@ -336,7 +340,7 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest { @Test public void givenOnEveryMessageProcessingSettingsAndSameMessageTwoTimes_whenOnMsg_thenPersistSameMessageTwoTimes() throws TbNodeException { // GIVEN - config.setProcessingSettings(new TbMsgTimeseriesNodeConfiguration.ProcessingSettings.OnEveryMessage()); + config.setProcessingSettings(new OnEveryMessage()); node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); @@ -374,7 +378,7 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest { @Test public void givenDeduplicateProcessingSettingsAndSameMessageTwoTimes_whenOnMsg_thenPersistThisMessageOnlyFirstTime() throws TbNodeException { // GIVEN - config.setProcessingSettings(new TbMsgTimeseriesNodeConfiguration.ProcessingSettings.Deduplicate(10)); + config.setProcessingSettings(new Deduplicate(10)); node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); @@ -412,7 +416,7 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest { @Test public void givenWebSocketsOnlyProcessingSettingsAndSameMessageTwoTimes_whenOnMsg_thenSendsOnlyWsUpdateTwoTimes() throws TbNodeException { // GIVEN - config.setProcessingSettings(new TbMsgTimeseriesNodeConfiguration.ProcessingSettings.WebSocketsOnly()); + config.setProcessingSettings(new WebSocketsOnly()); node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); @@ -450,7 +454,7 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest { @Test public void givenAdvancedProcessingSettingsWithOnEveryMessageStrategiesForAllActionsAndSameMessageTwoTimes_whenOnMsg_thenPersistSameMessageTwoTimes() throws TbNodeException { // GIVEN - config.setProcessingSettings(new TbMsgTimeseriesNodeConfiguration.ProcessingSettings.Advanced( + config.setProcessingSettings(new Advanced( ProcessingStrategy.onEveryMessage(), ProcessingStrategy.onEveryMessage(), ProcessingStrategy.onEveryMessage(), @@ -477,7 +481,6 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest { .previousCalculatedFieldIds(msg.getPreviousCalculatedFieldIds()) .tbMsgId(msg.getId()) .tbMsgType(msg.getInternalType()) - .callback(new TelemetryNodeCallback(ctxMock, msg)) .build(); node.onMsg(ctxMock, msg); @@ -494,7 +497,7 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest { @Test public void givenAdvancedProcessingSettingsWithDifferentDeduplicateStrategyForEachAction_whenOnMsg_thenEvaluatesStrategiesForEachActionsIndependently() throws TbNodeException { // GIVEN - config.setProcessingSettings(new TbMsgTimeseriesNodeConfiguration.ProcessingSettings.Advanced( + config.setProcessingSettings(new Advanced( ProcessingStrategy.deduplicate(1), ProcessingStrategy.deduplicate(2), ProcessingStrategy.deduplicate(3), @@ -580,7 +583,7 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest { @Test public void givenAdvancedProcessingSettingsWithSkipStrategiesForAllActionsAndSameMessageTwoTimes_whenOnMsg_thenSkipsSameMessageTwoTimes() throws TbNodeException { // GIVEN - config.setProcessingSettings(new TbMsgTimeseriesNodeConfiguration.ProcessingSettings.Advanced( + config.setProcessingSettings(new Advanced( ProcessingStrategy.skip(), ProcessingStrategy.skip(), ProcessingStrategy.skip(), diff --git a/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-processing-setting.component.html b/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-processing-setting.component.html index d72e7f1e43..edce8b12a9 100644 --- a/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-processing-setting.component.html +++ b/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-processing-setting.component.html @@ -18,22 +18,26 @@
- - + - - diff --git a/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-processing-setting.component.ts b/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-processing-setting.component.ts index 9e204c0219..4946de986a 100644 --- a/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-processing-setting.component.ts +++ b/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-processing-setting.component.ts @@ -19,12 +19,15 @@ import { FormBuilder, NG_VALIDATORS, NG_VALUE_ACCESSOR, + UntypedFormGroup, ValidationErrors, Validator } from '@angular/forms'; -import { Component, forwardRef } from '@angular/core'; +import { Component, DestroyRef, forwardRef, Input, OnInit } from '@angular/core'; import { takeUntilDestroyed } from '@angular/core/rxjs-interop'; import { AdvancedProcessingStrategy } from '@home/components/rule-node/action/timeseries-config.models'; +import { coerceBoolean } from '@shared/decorators/coercion'; +import { AttributeAdvancedProcessingStrategy } from '@home/components/rule-node/action/attributes-config.model'; @Component({ selector: 'tb-advanced-processing-settings', @@ -39,20 +42,55 @@ import { AdvancedProcessingStrategy } from '@home/components/rule-node/action/ti multi: true }] }) -export class AdvancedProcessingSettingComponent implements ControlValueAccessor, Validator { +export class AdvancedProcessingSettingComponent implements OnInit, ControlValueAccessor, Validator { - processingForm = this.fb.group({ - timeseries: [null], - latest: [null], - webSockets: [null], - calculatedFields: [null] - }); + @Input() + @coerceBoolean() + timeseries = false; + + @Input() + @coerceBoolean() + attributes = false; + + @Input() + @coerceBoolean() + latest = false; + + @Input() + @coerceBoolean() + webSockets = false; + + @Input() + @coerceBoolean() + calculatedFields = false; + + processingForm: UntypedFormGroup; private propagateChange: (value: any) => void = () => {}; - constructor(private fb: FormBuilder) { + constructor(private fb: FormBuilder, + private destroyRef: DestroyRef) { + } + + ngOnInit() { + this.processingForm = this.fb.group({}); + if (this.timeseries) { + this.processingForm.addControl('timeseries', this.fb.control(null, [])); + } + if (this.attributes) { + this.processingForm.addControl('attributes', this.fb.control(null, [])); + } + if (this.latest) { + this.processingForm.addControl('latest', this.fb.control(null, [])); + } + if (this.webSockets) { + this.processingForm.addControl('webSockets', this.fb.control(null, [])); + } + if (this.calculatedFields) { + this.processingForm.addControl('calculatedFields', this.fb.control(null, [])); + } this.processingForm.valueChanges.pipe( - takeUntilDestroyed() + takeUntilDestroyed(this.destroyRef) ).subscribe(value => this.propagateChange(value)); } @@ -77,7 +115,7 @@ export class AdvancedProcessingSettingComponent implements ControlValueAccessor, }; } - writeValue(value: AdvancedProcessingStrategy) { + writeValue(value: AdvancedProcessingStrategy | AttributeAdvancedProcessingStrategy) { this.processingForm.patchValue(value, {emitEvent: false}); } } diff --git a/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.component.html b/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.component.html index 03ea4fccdd..0de3f8020f 100644 --- a/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.component.html +++ b/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.component.html @@ -16,11 +16,54 @@ -->
+
+
+
+ rule-node-config.save-attribute.processing-settings +
+ + {{ 'rule-node-config.basic-mode' | translate}} + {{ 'rule-node-config.advanced-mode' | translate }} + +
+ @if(!attributesConfigForm.get('processingSettings.isAdvanced').value) { + + rule-node-config.save-attribute.strategy + + @for (strategy of processingStrategies; track strategy) { + {{ ProcessingTypeTranslationMap.get(strategy) | translate }} + } + + + + @if(attributesConfigForm.get('processingSettings.type').value === ProcessingType.DEDUPLICATE) { + + + } + } @else { + + } +
+
- - +
+ rule-node-config.save-attribute.scope +
- + {{ 'rule-node-config.attributes-scope' | translate }} @@ -29,7 +72,7 @@ - + {{ 'rule-node-config.attributes-scope-value' | translate }}
diff --git a/ui-ngx/src/assets/help/en_US/rulenode/save_attributes_node_advanced.md b/ui-ngx/src/assets/help/en_US/rulenode/save_attributes_node_advanced.md new file mode 100644 index 0000000000..f0705baa6f --- /dev/null +++ b/ui-ngx/src/assets/help/en_US/rulenode/save_attributes_node_advanced.md @@ -0,0 +1,32 @@ +#### Potential unexpected behavior with mixed processing strategies + +When configuring the processing strategies, certain combinations can lead to unexpected behavior. Consider the following scenarios: + +- **Skipping database storage** + + Choosing to disable attribute persistence introduces the risk of having only partial data available. + For example, if a message is processed solely for real-time notifications via WebSockets and not stored in the database, then attribute queries might not reflect the data visible on the dashboard. + +- **Disabling WebSocket (WS) updates** + + If WS updates are disabled, any changes to the attribute data won’t be pushed to dashboards (or other WS subscriptions). + This means that even if a database is updated, dashboards may not display the updated data until browser page is reloaded. + +- **Skipping calculated field recalculation** + + If attribute data is saved to the database while bypassing calculated field recalculation, the aggregated value may not update to reflect the saved data. + Conversely, if the calculated field is recalculated with new data but the corresponding attribute value is not persisted in the database, the calculated field's value might include data that isn’t stored. + +- **Different deduplication intervals across actions** + + When you configure different deduplication intervals for actions, the same incoming message might be processed differently for each action. + For example, a message might be stored immediately in the Attributes table (if set to *On every message*) while not being present on a dashboard because its deduplication interval hasn’t elapsed. + +- **Deduplication cache clearing** + + The deduplication mechanism uses a cache to track processed messages within each interval. + For performance and system stability reasons, this cache is periodically cleared. + As a result, if a cache entry is removed during the deduplication period, messages from the same originator may be processed more than once within that interval. + This means deduplication should be used as a performance optimization rather than an absolute guarantee of single processing per interval. + +We recommend using deduplication only when the occasional repeated processing is acceptable and won't cause system correctness issue or data inconsistencies. diff --git a/ui-ngx/src/assets/locale/locale.constant-en_US.json b/ui-ngx/src/assets/locale/locale.constant-en_US.json index 84fdd4f011..85e4daf264 100644 --- a/ui-ngx/src/assets/locale/locale.constant-en_US.json +++ b/ui-ngx/src/assets/locale/locale.constant-en_US.json @@ -5264,6 +5264,23 @@ "web-sockets": "WebSockets", "calculated-fields": "Calculated fields" }, + "save-attribute": { + "processing-settings": "Processing settings", + "processing-settings-hint": "Define how incoming messages are processed. In Basic mode, select a preconfigured processing strategy or enable only WebSocket updates. Advanced mode allows you to select individual processing strategies for each action.", + "advanced-settings-hint": "Be cautious when configuring processing strategies. Certain combinations can lead to unexpected behavior.", + "strategy": "Strategy", + "deduplication-interval": "Deduplication interval", + "deduplication-interval-required": "Deduplication interval is required", + "deduplication-interval-min-max-range": "Deduplication interval should be at least 1 second and at most 1 day", + "scope": "Scope", + "strategy-type": { + "every-message": "On every message", + "skip": "Skip", + "deduplicate": "Deduplicate", + "web-sockets-only": "WebSockets only" + }, + "attributes": "Attributes" + }, "key-val": { "key": "Key", "value": "Value",