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/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 01d7e69ba7..2ee3c439c1 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -817,7 +817,7 @@ spring: events: # Enable dedicated datasource (a separate database) for events and audit logs. # Before enabling this, make sure you have set up the following tables in the new DB: - # error_event, lc_event, rule_chain_debug_event, rule_node_debug_event, stats_event, audit_log + # error_event, lc_event, rule_chain_debug_event, rule_node_debug_event, stats_event, audit_log, cf_debug_event enabled: "${SPRING_DEDICATED_EVENTS_DATASOURCE_ENABLED:false}" # Database driver for Spring JPA for events datasource driverClassName: "${SPRING_EVENTS_DATASOURCE_DRIVER_CLASS_NAME:org.postgresql.Driver}" 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/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java index 06a0489037..f256b02d9a 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java @@ -140,6 +140,7 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura private long maxCalculatedFieldsPerEntity = 5; @Schema(example = "10") private long maxArgumentsPerCF = 10; + @Builder.Default @Min(value = 1, message = "must be at least 1") @Schema(example = "1000") private long maxDataPointsPerRollingArg = 1000; 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/dao/src/main/java/org/thingsboard/server/dao/config/DedicatedEventsJpaDaoConfig.java b/dao/src/main/java/org/thingsboard/server/dao/config/DedicatedEventsJpaDaoConfig.java index 5c24c45c16..258bac8dbd 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/config/DedicatedEventsJpaDaoConfig.java +++ b/dao/src/main/java/org/thingsboard/server/dao/config/DedicatedEventsJpaDaoConfig.java @@ -29,6 +29,7 @@ import org.springframework.orm.jpa.JpaTransactionManager; import org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean; import org.springframework.transaction.support.TransactionTemplate; import org.thingsboard.server.dao.model.sql.AuditLogEntity; +import org.thingsboard.server.dao.model.sql.CalculatedFieldDebugEventEntity; import org.thingsboard.server.dao.model.sql.ErrorEventEntity; import org.thingsboard.server.dao.model.sql.LifecycleEventEntity; import org.thingsboard.server.dao.model.sql.RuleChainDebugEventEntity; @@ -68,7 +69,7 @@ public class DedicatedEventsJpaDaoConfig { EntityManagerFactoryBuilder builder) { return builder .dataSource(eventsDataSource) - .packages(LifecycleEventEntity.class, StatisticsEventEntity.class, ErrorEventEntity.class, RuleNodeDebugEventEntity.class, RuleChainDebugEventEntity.class, AuditLogEntity.class) + .packages(LifecycleEventEntity.class, StatisticsEventEntity.class, ErrorEventEntity.class, RuleNodeDebugEventEntity.class, RuleChainDebugEventEntity.class, AuditLogEntity.class, CalculatedFieldDebugEventEntity.class) .persistenceUnit(EVENTS_PERSISTENCE_UNIT) .build(); } diff --git a/dao/src/main/resources/sql/schema-entities-idx-psql-addon.sql b/dao/src/main/resources/sql/schema-entities-idx-psql-addon.sql index b0c69ef6f3..b756d282f8 100644 --- a/dao/src/main/resources/sql/schema-entities-idx-psql-addon.sql +++ b/dao/src/main/resources/sql/schema-entities-idx-psql-addon.sql @@ -36,3 +36,5 @@ CREATE INDEX IF NOT EXISTS idx_lc_event_main CREATE INDEX IF NOT EXISTS idx_error_event_main ON error_event (tenant_id ASC, entity_id ASC, ts DESC NULLS LAST) WITH (FILLFACTOR=95); +CREATE INDEX IF NOT EXISTS idx_cf_debug_event_main + ON cf_debug_event (tenant_id ASC, entity_id ASC, ts DESC NULLS LAST) WITH (FILLFACTOR=95); 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..c04f5b474d 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,51 @@ 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 three actions: +
    +
  • Attributes: save attribute data to a database.
    • +
  • WebSockets: notify WebSockets subscriptions about attribute data updates.
    • +
  • Calculated fields: notify calculated fields 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: for all actions except WebSocket notifications, the "Skip" strategy is applied, while WebSocket notifications use the "On every message" strategy.
      • +
    +
    • +
  • Advanced: configure each action’s strategy independently.
    • +
+ + The node supports three attribute scopes: Client attributes, Shared attributes, and Server attributes. + You can set the default scope in the node configuration, or override it by specifying a valid scope property in the message metadata. +

+ 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 +118,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 +138,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 +161,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 +210,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 +218,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 +268,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 NodeConfigurationactions: + Node performs four actions:
  • Time series: save time series data to a ts_kv table in a DB.
  • Latest values: save time series data to a ts_kv_latest table in a DB.
    • @@ -82,7 +82,7 @@ import static org.thingsboard.server.common.data.msg.TbMsgType.POST_TELEMETRY_RE
    • 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 Time series and Latest values, and the "On every message" strategy to WebSockets.
      • +
    • WebSockets only: for all actions except WebSocket notifications, the "Skip" strategy is applied, while WebSocket notifications use the "On every message" strategy.
  • Advanced: configure each action’s strategy independently.
    • @@ -111,7 +111,7 @@ public class TbMsgTimeseriesNode implements TbNode { private TbContext ctx; private long tenantProfileDefaultStorageTtl; - private ProcessingSettings processingSettings; + private TimeseriesProcessingSettings processingSettings; @Override public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeConfiguration.java index a6cd278e67..48264978cf 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeConfiguration.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeConfiguration.java @@ -15,23 +15,12 @@ */ package org.thingsboard.rule.engine.telemetry; -import com.fasterxml.jackson.annotation.JsonCreator; -import com.fasterxml.jackson.annotation.JsonIgnore; -import com.fasterxml.jackson.annotation.JsonProperty; -import com.fasterxml.jackson.annotation.JsonSubTypes; -import com.fasterxml.jackson.annotation.JsonTypeInfo; import jakarta.validation.constraints.NotNull; import lombok.Data; -import lombok.Getter; import org.thingsboard.rule.engine.api.NodeConfiguration; -import org.thingsboard.rule.engine.telemetry.strategy.ProcessingStrategy; +import org.thingsboard.rule.engine.telemetry.settings.TimeseriesProcessingSettings; -import java.util.Objects; - -import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.ProcessingSettings.Advanced; -import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.ProcessingSettings.Deduplicate; -import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.ProcessingSettings.OnEveryMessage; -import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.ProcessingSettings.WebSocketsOnly; +import static org.thingsboard.rule.engine.telemetry.settings.TimeseriesProcessingSettings.OnEveryMessage; @Data public class TbMsgTimeseriesNodeConfiguration 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/calculated-fields/calculated-fields-table-config.ts b/ui-ngx/src/app/modules/home/components/calculated-fields/calculated-fields-table-config.ts index 6932860be4..cd9f0373db 100644 --- a/ui-ngx/src/app/modules/home/components/calculated-fields/calculated-fields-table-config.ts +++ b/ui-ngx/src/app/modules/home/components/calculated-fields/calculated-fields-table-config.ts @@ -14,7 +14,11 @@ /// limitations under the License. /// -import { EntityTableColumn, EntityTableConfig } from '@home/models/entity/entities-table-config.models'; +import { + DateEntityTableColumn, + EntityTableColumn, + EntityTableConfig +} from '@home/models/entity/entities-table-config.models'; import { EntityType, entityTypeTranslations } from '@shared/models/entity-type.models'; import { TranslateService } from '@ngx-translate/core'; import { Direction } from '@shared/models/page/sort-order'; @@ -45,6 +49,7 @@ import { getCalculatedFieldArgumentsEditorCompleter, getCalculatedFieldArgumentsHighlights, CalculatedFieldTypeTranslations, + CalculatedFieldType, } from '@shared/models/calculated-field.models'; import { CalculatedFieldDebugDialogComponent, @@ -53,6 +58,7 @@ import { } from './components/public-api'; import { ImportExportService } from '@shared/import-export/import-export.service'; import { isObject } from '@core/utils'; +import { DatePipe } from '@angular/common'; export class CalculatedFieldsTableConfig extends EntityTableConfig { @@ -69,6 +75,7 @@ export class CalculatedFieldsTableConfig extends EntityTableConfig, private durationLeft: DurationLeftPipe, @@ -107,11 +114,20 @@ export class CalculatedFieldsTableConfig extends EntityTableConfig('expression', 'calculated-fields.expression', '33%', entity => entity.configuration?.expression); + const expressionColumn = new EntityTableColumn('expression', 'calculated-fields.expression', '300px'); expressionColumn.sortable = false; + expressionColumn.cellContentFunction = entity => { + const expressionLabel = this.getExpressionLabel(entity); + return expressionLabel.length < 45 ? expressionLabel : `${expressionLabel.substring(0, 44)}…`; + } + expressionColumn.cellTooltipFunction = entity => { + const expressionLabel = this.getExpressionLabel(entity); + return expressionLabel.length < 45 ? null : expressionLabel + }; + this.columns.push(new DateEntityTableColumn('createdTime', 'common.created-time', this.datePipe, '150px')); this.columns.push(new EntityTableColumn('name', 'common.name', '33%')); this.columns.push(new EntityTableColumn('type', 'common.type', '50px', entity => this.translate.instant(CalculatedFieldTypeTranslations.get(entity.type)))); this.columns.push(expressionColumn); @@ -146,6 +162,14 @@ export class CalculatedFieldsTableConfig extends EntityTableConfig> { return this.calculatedFieldsService.getCalculatedFields(this.entityId, pageLink); } diff --git a/ui-ngx/src/app/modules/home/components/calculated-fields/calculated-fields-table.component.ts b/ui-ngx/src/app/modules/home/components/calculated-fields/calculated-fields-table.component.ts index be6fb11480..1d0b9dcb15 100644 --- a/ui-ngx/src/app/modules/home/components/calculated-fields/calculated-fields-table.component.ts +++ b/ui-ngx/src/app/modules/home/components/calculated-fields/calculated-fields-table.component.ts @@ -35,6 +35,7 @@ import { DurationLeftPipe } from '@shared/pipe/duration-left.pipe'; import { TbPopoverService } from '@shared/components/popover.service'; import { CalculatedFieldsService } from '@core/http/calculated-fields.service'; import { ImportExportService } from '@shared/import-export/import-export.service'; +import { DatePipe } from '@angular/common'; @Component({ selector: 'tb-calculated-fields-table', @@ -56,6 +57,7 @@ export class CalculatedFieldsTableComponent { private translate: TranslateService, private dialog: MatDialog, private store: Store, + private datePipe: DatePipe, private durationLeft: DurationLeftPipe, private popoverService: TbPopoverService, private cd: ChangeDetectorRef, @@ -69,6 +71,7 @@ export class CalculatedFieldsTableComponent { this.calculatedFieldsService, this.translate, this.dialog, + this.datePipe, this.entityId(), this.store, this.durationLeft, diff --git a/ui-ngx/src/app/modules/home/components/calculated-fields/components/arguments-table/calculated-field-arguments-table.component.html b/ui-ngx/src/app/modules/home/components/calculated-fields/components/arguments-table/calculated-field-arguments-table.component.html index 6bab01f976..162bd3aa1e 100644 --- a/ui-ngx/src/app/modules/home/components/calculated-fields/components/arguments-table/calculated-field-arguments-table.component.html +++ b/ui-ngx/src/app/modules/home/components/calculated-fields/components/arguments-table/calculated-field-arguments-table.component.html @@ -20,18 +20,27 @@ - +
    {{ 'common.name' | translate }}
     - -
    {{ argument.argumentName }}
    + +
    +
    {{ argument.argumentName }}
    + +
    - + {{ 'entity.entity-type' | translate }} - +
    @if (argument.refEntityId?.entityType === ArgumentEntityType.Tenant) { {{ 'calculated-fields.argument-current-tenant' | translate }} @@ -98,7 +107,7 @@
    - } +
    {{ 'calculated-fields.output' | translate }}
    @@ -154,26 +152,35 @@ }
    @if (fieldFormGroup.get('type').value === CalculatedFieldType.SIMPLE) { - - - {{ (outputFormGroup.get('type').value === OutputType.Timeseries - ? 'calculated-fields.timeseries-key' - : 'calculated-fields.attribute-key') - | translate }} - - - @if (outputFormGroup.get('name').errors && outputFormGroup.get('name').touched) { - - @if (outputFormGroup.get('name').hasError('required')) { - {{ 'common.hint.key-required' | translate }} - } @else if (outputFormGroup.get('name').hasError('pattern')) { - {{ 'common.hint.key-pattern' | translate }} - } @else if (outputFormGroup.get('name').hasError('maxlength')) { - {{ 'common.hint.key-max-length' | translate }} - } - - } - +
    + + + {{ (outputFormGroup.get('type').value === OutputType.Timeseries + ? 'calculated-fields.timeseries-key' + : 'calculated-fields.attribute-key') + | translate }} + + + @if (outputFormGroup.get('name').errors && outputFormGroup.get('name').touched) { + + @if (outputFormGroup.get('name').hasError('required')) { + {{ 'common.hint.key-required' | translate }} + } @else if (outputFormGroup.get('name').hasError('pattern')) { + {{ 'common.hint.key-pattern' | translate }} + } @else if (outputFormGroup.get('name').hasError('maxlength')) { + {{ 'common.hint.key-max-length' | translate }} + } + + } + + + {{ 'calculated-fields.decimals-by-default' | translate }} + + @if (outputFormGroup.get('decimalsByDefault').errors && outputFormGroup.get('decimalsByDefault').touched) { + {{ 'calculated-fields.hint.decimals-range' | translate }} + } + +
    }     diff --git a/ui-ngx/src/app/modules/home/components/calculated-fields/components/dialog/calculated-field-dialog.component.scss b/ui-ngx/src/app/modules/home/components/calculated-fields/components/dialog/calculated-field-dialog.component.scss index 0e994ed825..8bc422eed1 100644 --- a/ui-ngx/src/app/modules/home/components/calculated-fields/components/dialog/calculated-field-dialog.component.scss +++ b/ui-ngx/src/app/modules/home/components/calculated-fields/components/dialog/calculated-field-dialog.component.scss @@ -13,39 +13,43 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -:host { - .dialog-container { - width: 869px; - max-width: 100%; - } - .script-lang-chip { - line-height: 20px; - font-size: 14px; - font-weight: 500; - color: white; - border-radius: 100px; - width: 70px; - display: flex; - justify-content: center; - margin-top: 2px; - margin-right: 4px; - } +.calculated-field-dialog-container { + width: 869px; + max-width: 100%; +} + +.tbel-script-lang-chip { + line-height: 20px; + font-size: 14px; + font-weight: 500; + color: white; + border-radius: 100px; + width: 70px; + min-width: 70px; + display: flex; + justify-content: center; + margin-top: 2px; + margin-right: 4px; } -:host ::ng-deep { - .expression-edit { - .ace_tb { - &.ace_calculated-field { - &-key { - color: #C52F00; - } - &-ts, &-time-window, &-values, &-value, &-func { - color: #7214D0; - } - &-start-ts, &-end-ts, &-limit { - color: #185F2A; - } +.tb-js-func { + .ace_tb { + &.ace_calculated-field { + &-ctx { + color: #C52F00; + } + &-args { + color: #185F2A; + } + &-key { + color: #c24c1a; + } + &-time-window, &-values, &-func, &-value, &-ts { + color: #7214D0; + } + &-start-ts, &-end-ts { + color: #2CAA00; } } } diff --git a/ui-ngx/src/app/modules/home/components/calculated-fields/components/dialog/calculated-field-dialog.component.ts b/ui-ngx/src/app/modules/home/components/calculated-fields/components/dialog/calculated-field-dialog.component.ts index 68abd646e3..052b660c3c 100644 --- a/ui-ngx/src/app/modules/home/components/calculated-fields/components/dialog/calculated-field-dialog.component.ts +++ b/ui-ngx/src/app/modules/home/components/calculated-fields/components/dialog/calculated-field-dialog.component.ts @@ -14,7 +14,7 @@ /// limitations under the License. /// -import { Component, DestroyRef, Inject } from '@angular/core'; +import { Component, DestroyRef, Inject, ViewEncapsulation } from '@angular/core'; import { MAT_DIALOG_DATA, MatDialogRef } from '@angular/material/dialog'; import { Store } from '@ngrx/store'; import { AppState } from '@core/core.state'; @@ -24,6 +24,7 @@ import { DialogComponent } from '@shared/components/dialog.component'; import { CalculatedField, CalculatedFieldConfiguration, + calculatedFieldDefaultScript, CalculatedFieldDialogData, CalculatedFieldType, CalculatedFieldTypeTranslations, @@ -32,7 +33,7 @@ import { OutputType, OutputTypeTranslations } from '@shared/models/calculated-field.models'; -import { noLeadTrailSpacesRegex } from '@shared/models/regex.constants'; +import { digitsRegex, oneSpaceInsideRegex } from '@shared/models/regex.constants'; import { AttributeScope } from '@shared/models/telemetry/telemetry.models'; import { EntityType } from '@shared/models/entity-type.models'; import { map, startWith, switchMap } from 'rxjs/operators'; @@ -45,21 +46,23 @@ import { Observable } from 'rxjs'; selector: 'tb-calculated-field-dialog', templateUrl: './calculated-field-dialog.component.html', styleUrls: ['./calculated-field-dialog.component.scss'], + encapsulation: ViewEncapsulation.None }) export class CalculatedFieldDialogComponent extends DialogComponent { fieldFormGroup = this.fb.group({ - name: ['', [Validators.required, Validators.pattern(noLeadTrailSpacesRegex), Validators.maxLength(255)]], + name: ['', [Validators.required, Validators.pattern(oneSpaceInsideRegex), Validators.maxLength(255)]], type: [CalculatedFieldType.SIMPLE], debugSettings: [], configuration: this.fb.group({ arguments: this.fb.control({}), - expressionSIMPLE: ['', [Validators.required, Validators.pattern(noLeadTrailSpacesRegex), Validators.maxLength(255)]], - expressionSCRIPT: [], + expressionSIMPLE: ['', [Validators.required, Validators.pattern(oneSpaceInsideRegex), Validators.maxLength(255)]], + expressionSCRIPT: [calculatedFieldDefaultScript], output: this.fb.group({ - name: ['', [Validators.required, Validators.pattern(noLeadTrailSpacesRegex), Validators.maxLength(255)]], + name: ['', [Validators.required, Validators.pattern(oneSpaceInsideRegex), Validators.maxLength(255)]], scope: [{ value: AttributeScope.SERVER_SCOPE, disabled: true }], - type: [OutputType.Timeseries] + type: [OutputType.Timeseries], + decimalsByDefault: [null as number, [Validators.min(0), Validators.max(15), Validators.pattern(digitsRegex)]], }), }), }); @@ -67,7 +70,7 @@ export class CalculatedFieldDialogComponent extends DialogComponent Object.keys(argumentsObj)) + map(argumentsObj => ['ctx', ...Object.keys(argumentsObj)]) ); argumentsEditorCompleter$ = this.configFormGroup.get('arguments').valueChanges @@ -119,9 +122,18 @@ export class CalculatedFieldDialogComponent extends DialogComponent warning + } @else if (argumentFormGroup.get('argumentName').touched && argumentFormGroup.get('argumentName').hasError('equalCtx')) { + + warning + } diff --git a/ui-ngx/src/app/modules/home/components/calculated-fields/components/panel/calculated-field-argument-panel.component.ts b/ui-ngx/src/app/modules/home/components/calculated-fields/components/panel/calculated-field-argument-panel.component.ts index 9d9614ba28..482851c59c 100644 --- a/ui-ngx/src/app/modules/home/components/calculated-fields/components/panel/calculated-field-argument-panel.component.ts +++ b/ui-ngx/src/app/modules/home/components/calculated-fields/components/panel/calculated-field-argument-panel.component.ts @@ -17,7 +17,7 @@ import { ChangeDetectorRef, Component, Input, OnInit, output } from '@angular/core'; import { TbPopoverComponent } from '@shared/components/popover.component'; import { FormBuilder, FormControl, FormGroup, ValidatorFn, Validators } from '@angular/forms'; -import { charsWithNumRegex, noLeadTrailSpacesRegex } from '@shared/models/regex.constants'; +import { charsWithNumRegex, oneSpaceInsideRegex } from '@shared/models/regex.constants'; import { ArgumentEntityType, ArgumentEntityTypeParamsMap, @@ -64,17 +64,17 @@ export class CalculatedFieldArgumentPanelComponent implements OnInit { readonly defaultLimit = Math.floor(this.maxDataPointsPerRollingArg / 10); argumentFormGroup = this.fb.group({ - argumentName: ['', [Validators.required, this.uniqNameRequired(), Validators.pattern(charsWithNumRegex), Validators.maxLength(255)]], + argumentName: ['', [Validators.required, this.uniqNameRequired(), this.notEqualCtxValidator(), Validators.pattern(charsWithNumRegex), Validators.maxLength(255)]], refEntityId: this.fb.group({ entityType: [ArgumentEntityType.Current], id: [''] }), refEntityKey: this.fb.group({ type: [ArgumentType.LatestTelemetry, [Validators.required]], - key: [''], + key: ['', [Validators.pattern(oneSpaceInsideRegex)]], scope: [{ value: AttributeScope.SERVER_SCOPE, disabled: true }, [Validators.required]], }), - defaultValue: ['', [Validators.pattern(noLeadTrailSpacesRegex)]], + defaultValue: ['', [Validators.pattern(oneSpaceInsideRegex)]], limit: [{ value: this.defaultLimit, disabled: !this.maxDataPointsPerRollingArg }], timeWindow: [MINUTE * 15], }); @@ -142,6 +142,10 @@ export class CalculatedFieldArgumentPanelComponent implements OnInit { if (refEntityId.entityType === ArgumentEntityType.Tenant) { refEntityId.id = this.tenantId; } + if (value.defaultValue) { + value.defaultValue = value.defaultValue.trim(); + } + value.refEntityKey.key = value.refEntityKey.key.trim(); this.argumentsDataApplied.emit({ value, index: this.index }); } @@ -233,6 +237,13 @@ export class CalculatedFieldArgumentPanelComponent implements OnInit { } } + private notEqualCtxValidator(): ValidatorFn { + return (control: FormControl) => { + const trimmedValue = control.value.trim().toLowerCase(); + return trimmedValue === 'ctx' ? { equalCtx: true } : null; + }; + } + private observeUpdatePosition(): void { merge( this.refEntityIdFormGroup.get('entityType').valueChanges, diff --git a/ui-ngx/src/app/modules/home/components/calculated-fields/components/test-dialog/calculated-field-script-test-dialog.component.html b/ui-ngx/src/app/modules/home/components/calculated-fields/components/test-dialog/calculated-field-script-test-dialog.component.html index cda99f6c09..6859ff6dab 100644 --- a/ui-ngx/src/app/modules/home/components/calculated-fields/components/test-dialog/calculated-field-script-test-dialog.component.html +++ b/ui-ngx/src/app/modules/home/components/calculated-fields/components/test-dialog/calculated-field-script-test-dialog.component.html @@ -15,7 +15,7 @@ limitations under the License. --> -
    +

    {{ 'calculated-fields.test-script-function' | translate }} ({{ 'api-usage.tbel' | translate }})