From 400e74b00defd0f3854fafadfa98e0018735f51e Mon Sep 17 00:00:00 2001 From: Dmytro Skarzhynets Date: Wed, 26 Feb 2025 16:28:46 +0200 Subject: [PATCH 1/9] Save attributes strategies: BE initial implementation --- .../rule_chains/edge_root_rule_chain.json | 5 +- .../device_profile/rule_chain_template.json | 5 +- .../tenant/rule_chains/root_rule_chain.json | 5 +- .../main/data/upgrade/basic/schema_update.sql | 30 +- .../DefaultTelemetrySubscriptionService.java | 12 +- ...faultTelemetrySubscriptionServiceTest.java | 96 ++- .../src/main/resources/root_rule_chain.json | 5 +- .../engine/api/AttributesSaveRequest.java | 17 +- .../engine/api/AttributesSaveRequestTest.java | 46 ++ .../engine/telemetry/TbMsgAttributesNode.java | 109 ++- .../TbMsgAttributesNodeConfiguration.java | 9 + .../engine/telemetry/TbMsgTimeseriesNode.java | 12 +- .../TbMsgTimeseriesNodeConfiguration.java | 62 +- .../AttributesProcessingSettings.java | 51 ++ .../settings/BaseProcessingSettings.java | 47 ++ .../TimeseriesProcessingSettings.java | 52 ++ .../TbMsgAttributesNodeConfigurationTest.java | 29 - .../telemetry/TbMsgAttributesNodeTest.java | 680 +++++++++++++++--- .../telemetry/TbMsgTimeseriesNodeTest.java | 20 +- 19 files changed, 1029 insertions(+), 263 deletions(-) create mode 100644 rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/AttributesSaveRequestTest.java create mode 100644 rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/settings/AttributesProcessingSettings.java create mode 100644 rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/settings/BaseProcessingSettings.java create mode 100644 rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/settings/TimeseriesProcessingSettings.java delete mode 100644 rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeConfigurationTest.java 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 6701b59e0e..9bb838963f 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 2929635949..7a11f22202 100644 --- a/application/src/main/data/upgrade/basic/schema_update.sql +++ b/application/src/main/data/upgrade/basic/schema_update.sql @@ -62,4 +62,32 @@ $$; -- UPDATE SAVE TIME SERIES NODES END -ALTER TABLE api_usage_state ADD COLUMN IF NOT EXISTS version BIGINT DEFAULT 1; \ No newline at end of file +-- UPDATE SAVE ATTRIBUTES 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 + || 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; + + END IF; + END; +$$; + +-- 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/telemetry/DefaultTelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java index 568d80787d..ace5247724 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 @@ -53,6 +53,7 @@ import org.thingsboard.server.service.entitiy.entityview.TbEntityViewService; import org.thingsboard.server.service.subscription.TbSubscriptionUtils; import java.util.ArrayList; +import java.util.Collections; import java.util.Comparator; import java.util.HashMap; import java.util.List; @@ -165,9 +166,16 @@ 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()); + ListenableFuture> saveFuture; + if (request.getStrategy().saveAttributes()) { + saveFuture = attrService.save(request.getTenantId(), request.getEntityId(), request.getScope(), request.getEntries()); + } else { + saveFuture = Futures.immediateFuture(Collections.emptyList()); + } addMainCallback(saveFuture, request.getCallback()); - addWsCallback(saveFuture, success -> onAttributesUpdate(request.getTenantId(), request.getEntityId(), request.getScope().name(), request.getEntries(), request.isNotifyDevice())); + if (request.getStrategy().sendWsUpdate()) { + addWsCallback(saveFuture, success -> onAttributesUpdate(request.getTenantId(), request.getEntityId(), request.getScope().name(), request.getEntries(), request.isNotifyDevice())); + } } @Override 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 dbd61b638e..3fb5321499 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 @@ -29,11 +29,13 @@ 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.AttributesSaveRequest; 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.EntityView; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; @@ -86,7 +88,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)) ); @@ -147,9 +149,9 @@ class DefaultTelemetrySubscriptionServiceTest { lenient().when(partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId)).thenReturn(tpi); - lenient().when(tsService.save(tenantId, entityId, sampleTelemetry, sampleTtl)).thenReturn(immediateFuture(sampleTelemetry.size())); - lenient().when(tsService.saveWithoutLatest(tenantId, entityId, sampleTelemetry, sampleTtl)).thenReturn(immediateFuture(sampleTelemetry.size())); - lenient().when(tsService.saveLatest(tenantId, entityId, sampleTelemetry)).thenReturn(immediateFuture(listOfNNumbers(sampleTelemetry.size()))); + lenient().when(tsService.save(tenantId, entityId, sampleTimeseries, sampleTtl)).thenReturn(immediateFuture(sampleTimeseries.size())); + lenient().when(tsService.saveWithoutLatest(tenantId, entityId, sampleTimeseries, sampleTtl)).thenReturn(immediateFuture(sampleTimeseries.size())); + lenient().when(tsService.saveLatest(tenantId, entityId, sampleTimeseries)).thenReturn(immediateFuture(listOfNNumbers(sampleTimeseries.size()))); // mock no entity views lenient().when(tbEntityViewService.findEntityViewsByTenantIdAndEntityIdAsync(tenantId, entityId)).thenReturn(immediateFuture(Collections.emptyList())); @@ -171,7 +173,7 @@ class DefaultTelemetrySubscriptionServiceTest { .tenantId(tenantId) .customerId(customerId) .entityId(entityId) - .entries(sampleTelemetry) + .entries(sampleTimeseries) .ttl(sampleTtl) .strategy(new TimeseriesSaveRequest.Strategy(true, false, false)) .callback(emptyCallback) @@ -181,7 +183,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 @@ -191,7 +193,7 @@ class DefaultTelemetrySubscriptionServiceTest { .tenantId(tenantId) .customerId(customerId) .entityId(entityId) - .entries(sampleTelemetry) + .entries(sampleTimeseries) .ttl(sampleTtl) .strategy(TimeseriesSaveRequest.Strategy.LATEST_AND_WS) .callback(emptyCallback) @@ -214,7 +216,7 @@ class DefaultTelemetrySubscriptionServiceTest { .tenantId(tenantId) .customerId(customerId) .entityId(entityId) - .entries(sampleTelemetry) + .entries(sampleTimeseries) .ttl(sampleTtl) .strategy(TimeseriesSaveRequest.Strategy.SAVE_ALL) .future(future) @@ -240,7 +242,7 @@ class DefaultTelemetrySubscriptionServiceTest { .tenantId(tenantId) .customerId(customerId) .entityId(entityId) - .entries(sampleTelemetry) + .entries(sampleTimeseries) .ttl(sampleTtl) .strategy(TimeseriesSaveRequest.Strategy.LATEST_AND_WS) .future(future) @@ -260,12 +262,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(listOfNNumbers(sampleTelemetry.size()))); + given(tsService.saveLatest(tenantId, entityView.getId(), sampleTimeseries)).willReturn(immediateFuture(listOfNNumbers(sampleTimeseries.size()))); // mock TPI for entity view given(partitionService.resolve(ServiceType.TB_CORE, tenantId, entityView.getId())).willReturn(tpi); @@ -273,7 +275,7 @@ class DefaultTelemetrySubscriptionServiceTest { .tenantId(tenantId) .customerId(customerId) .entityId(entityId) - .entries(sampleTelemetry) + .entries(sampleTimeseries) .ttl(sampleTtl) .strategy(new TimeseriesSaveRequest.Strategy(false, true, false)) .callback(emptyCallback) @@ -284,12 +286,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(); } @@ -300,7 +302,7 @@ class DefaultTelemetrySubscriptionServiceTest { .tenantId(tenantId) .customerId(customerId) .entityId(entityId) - .entries(sampleTelemetry) + .entries(sampleTimeseries) .ttl(sampleTtl) .strategy(new TimeseriesSaveRequest.Strategy(true, false, false)) .callback(emptyCallback) @@ -311,7 +313,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 @@ -319,14 +321,14 @@ class DefaultTelemetrySubscriptionServiceTest { } @ParameterizedTest - @MethodSource("booleanCombinations") - void shouldCallCorrectApiBasedOnBooleanFlagsInTheSaveRequest(boolean saveTimeseries, boolean saveLatest, boolean sendWsUpdate) { + @MethodSource("allCombinationsOfThreeBooleans") + void shouldCallCorrectSaveTimeseriesApiBasedOnBooleanFlagsInTheSaveRequest(boolean saveTimeseries, boolean saveLatest, boolean sendWsUpdate) { // GIVEN var request = TimeseriesSaveRequest.builder() .tenantId(tenantId) .customerId(customerId) .entityId(entityId) - .entries(sampleTelemetry) + .entries(sampleTimeseries) .ttl(sampleTtl) .strategy(new TimeseriesSaveRequest.Strategy(saveTimeseries, saveLatest, sendWsUpdate)) .callback(emptyCallback) @@ -337,22 +339,22 @@ 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); } 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 allCombinationsOfThreeBooleans() { return Stream.of( Arguments.of(true, true, true), Arguments.of(true, true, false), @@ -365,7 +367,49 @@ class DefaultTelemetrySubscriptionServiceTest { ); } - // used to emulate sequence numbers returned by save latest API + @ParameterizedTest + @MethodSource("allCombinationsOfTwoBooleans") + void shouldCallCorrectSaveAttributesApiBasedOnBooleanFlagsInTheSaveRequest(boolean saveAttributes, boolean sendWsUpdate) { + // 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)) + .callback(emptyCallback) + .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 (sendWsUpdate) { + then(subscriptionManagerService).should().onAttributesUpdate(tenantId, entityId, request.getScope().name(), request.getEntries(), request.isNotifyDevice(), TbCallback.EMPTY); + } else { + then(subscriptionManagerService).shouldHaveNoInteractions(); + } + } + + static Stream allCombinationsOfTwoBooleans() { + return Stream.of( + Arguments.of(true, true), + Arguments.of(true, false), + Arguments.of(false, true), + Arguments.of(false, false) + ); + } + + // used to emulate sequence numbers returned by save APIs private static List listOfNNumbers(int N) { return LongStream.range(0, N).boxed().toList(); } 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/AttributesSaveRequest.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java index c7d09d1525..5fb0ed3b92 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 @@ -40,8 +40,17 @@ public class AttributesSaveRequest { private final AttributeScope scope; private final List entries; private final boolean notifyDevice; + private final Strategy strategy; private final FutureCallback callback; + public record Strategy(boolean saveAttributes, boolean sendWsUpdate) { + + public static final Strategy PROCESS_ALL = new Strategy(true, true); + public static final Strategy WS_ONLY = new Strategy(false, true); + public static final Strategy SKIP_ALL = new Strategy(false, false); + + } + public static Builder builder() { return new Builder(); } @@ -53,6 +62,7 @@ public class AttributesSaveRequest { private AttributeScope scope; private List entries; private boolean notifyDevice = true; + private Strategy strategy = Strategy.PROCESS_ALL; private FutureCallback callback; Builder() {} @@ -100,6 +110,11 @@ public class AttributesSaveRequest { return this; } + public Builder strategy(Strategy strategy) { + this.strategy = strategy; + return this; + } + public Builder callback(FutureCallback callback) { this.callback = callback; return this; @@ -120,7 +135,7 @@ public class AttributesSaveRequest { } public AttributesSaveRequest build() { - return new AttributesSaveRequest(tenantId, entityId, scope, entries, notifyDevice, callback); + return new AttributesSaveRequest(tenantId, entityId, scope, entries, notifyDevice, strategy, callback); } } 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..1323130e7e --- /dev/null +++ b/rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/AttributesSaveRequestTest.java @@ -0,0 +1,46 @@ +/** + * 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 static org.assertj.core.api.Assertions.assertThat; + +class AttributesSaveRequestTest { + + @Test + void testDefaultSaveStrategyIsProcessAll() { + var request = AttributesSaveRequest.builder().build(); + + assertThat(request.getStrategy()).isEqualTo(AttributesSaveRequest.Strategy.PROCESS_ALL); + } + + @Test + void testProcessAllStrategy() { + assertThat(AttributesSaveRequest.Strategy.PROCESS_ALL).isEqualTo(new AttributesSaveRequest.Strategy(true, true)); + } + + @Test + void testWsOnlyStrategy() { + assertThat(AttributesSaveRequest.Strategy.WS_ONLY).isEqualTo(new AttributesSaveRequest.Strategy(false, true)); + } + + @Test + void testSkipAllStrategy() { + assertThat(AttributesSaveRequest.Strategy.SKIP_ALL).isEqualTo(new AttributesSaveRequest.Strategy(false, false)); + } + +} 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 da6d82707a..053e743596 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java @@ -23,6 +23,7 @@ import com.google.common.util.concurrent.MoreExecutors; import com.google.gson.JsonParser; import lombok.extern.slf4j.Slf4j; import org.thingsboard.common.util.DonAsynchron; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.RuleNode; import org.thingsboard.rule.engine.api.TbContext; @@ -30,6 +31,7 @@ import org.thingsboard.rule.engine.api.TbNode; import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeException; import org.thingsboard.rule.engine.api.util.TbNodeUtils; +import org.thingsboard.rule.engine.telemetry.settings.AttributesProcessingSettings; import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.StringUtils; @@ -43,9 +45,14 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.UUID; import java.util.function.Function; import java.util.stream.Collectors; +import static org.thingsboard.rule.engine.telemetry.settings.AttributesProcessingSettings.Advanced; +import static org.thingsboard.rule.engine.telemetry.settings.AttributesProcessingSettings.Deduplicate; +import static org.thingsboard.rule.engine.telemetry.settings.AttributesProcessingSettings.OnEveryMessage; +import static org.thingsboard.rule.engine.telemetry.settings.AttributesProcessingSettings.WebSocketsOnly; import static org.thingsboard.server.common.data.DataConstants.NOTIFY_DEVICE_METADATA_KEY; import static org.thingsboard.server.common.data.DataConstants.SCOPE; import static org.thingsboard.server.common.data.msg.TbMsgType.POST_ATTRIBUTES_REQUEST; @@ -55,13 +62,47 @@ import static org.thingsboard.server.common.data.msg.TbMsgType.POST_ATTRIBUTES_R type = ComponentType.ACTION, name = "save attributes", configClazz = TbMsgAttributesNodeConfiguration.class, - version = 2, - nodeDescription = "Saves attributes data", - nodeDetails = "Saves entity attributes based on configurable scope parameter. Expects messages with 'POST_ATTRIBUTES_REQUEST' message type. " + - "If upsert(update/insert) operation is completed successfully rule node will send the incoming message via Success chain, otherwise, Failure chain is used. " + - "Additionally if checkbox Send attributes updated notification is set to true, rule node will put the \"Attributes Updated\" " + - "event for SHARED_SCOPE and SERVER_SCOPE attributes updates to the corresponding rule engine queue." + - "Performance checkbox 'Save attributes only if the value changes' will skip attributes overwrites for values with no changes (avoid concurrent writes because this check is not transactional; will not update 'Last updated time' for skipped attributes).", + version = 3, + nodeDescription = """ + Saves attribute data with a configurable scope and according to configured processing strategies. + """, + nodeDetails = """ + Node performs two actions: +
    +
  • Attributes: save attribute data to a database.
    • +
  • WebSockets: notify WebSockets subscriptions about attribute data updates.
    • +
+ + For each action, three processing strategies are available: +
    +
  • On every message: perform the action for every message.
    • +
  • Deduplicate: perform the action only for the first message from a particular originator within a configurable interval.
    • +
  • Skip: never perform the action.
    • +
+ + Processing strategies are configured using processing settings, which support two modes: +
    +
  • Basic +
      +
    • On every message: applies the "On every message" strategy to all actions.
      • +
    • Deduplicate: applies the "Deduplicate" strategy (with a specified interval) to all actions.
      • +
    • WebSockets only: applies the "Skip" strategy to Attributes, and the "On every message" strategy to WebSockets.
      • +
    +
    • +
  • Advanced: configure each action’s strategy independently.
    • +
+ + Additionally: +
    +
  • If Save attributes only if the value changes is enabled, the rule node compares the received attribute value with the current stored value and skips the save operation if they match.
    • +
  • If Send attributes updated notification is enabled, the rule node will put the Attributes Updated event for SHARED_SCOPE and SERVER_SCOPE attribute updates to the queue named Main.
    • +
  • If Force notification to the device is enabled, then rule node will always notify device about SHARED_SCOPE attribute updates, regardless of the value of notifyDevice metadata property.
    • +
+ + This node expects messages of type POST_ATTRIBUTES_REQUEST. +

+ Output connections: Success, Failure. + """, configDirective = "tbActionNodeAttributesConfig", icon = "file_upload" ) @@ -73,9 +114,12 @@ public class TbMsgAttributesNode implements TbNode { private TbMsgAttributesNodeConfiguration config; + private AttributesProcessingSettings processingSettings; + @Override public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { - this.config = TbNodeUtils.convert(configuration, TbMsgAttributesNodeConfiguration.class); + config = TbNodeUtils.convert(configuration, TbMsgAttributesNodeConfiguration.class); + processingSettings = config.getProcessingSettings(); } @Override @@ -90,11 +134,20 @@ public class TbMsgAttributesNode implements TbNode { ctx.tellSuccess(msg); return; } + + AttributesSaveRequest.Strategy strategy = determineSaveStrategy(msg.getMetaDataTs(), msg.getOriginator().getId()); + + // short-circuit + if (!strategy.saveAttributes() && !strategy.sendWsUpdate()) { + ctx.tellSuccess(msg); + return; + } + AttributeScope scope = getScope(msg.getMetaData().getValue(SCOPE)); boolean sendAttributesUpdateNotification = checkSendNotification(scope); if (!config.isUpdateAttributesOnlyOnValueChange()) { - saveAttr(newAttributes, ctx, msg, scope, sendAttributesUpdateNotification); + saveAttr(newAttributes, ctx, msg, scope, sendAttributesUpdateNotification, strategy); return; } @@ -104,13 +157,41 @@ 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) + ); + } + // 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,11 +205,12 @@ public class TbMsgAttributesNode implements TbNode { .scope(scope) .entries(attributes) .notifyDevice(config.isNotifyDevice() || checkNotifyDeviceMdValue(msg.getMetaData().getValue(NOTIFY_DEVICE_METADATA_KEY))) + .strategy(strategy) .callback(callback) .build()); } - List filterChangedAttr(List currentAttributes, List newAttributes) { + private List filterChangedAttr(List currentAttributes, List newAttributes) { if (currentAttributes == null || currentAttributes.isEmpty()) { return newAttributes; } @@ -178,6 +260,9 @@ public class TbMsgAttributesNode implements TbNode { hasChanges = fixEscapedBooleanConfigParameter(oldConfiguration, SEND_ATTRIBUTES_UPDATED_NOTIFICATION_KEY, hasChanges, false); // update updateAttributesOnlyOnValueChange. hasChanges = fixEscapedBooleanConfigParameter(oldConfiguration, UPDATE_ATTRIBUTES_ONLY_ON_VALUE_CHANGE_KEY, hasChanges, true); + case 2: + hasChanges = true; + ((ObjectNode) oldConfiguration).set("processingSettings", JacksonUtil.valueToTree(new OnEveryMessage())); break; default: break; diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeConfiguration.java index 161aa64d5f..2687125a0c 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeConfiguration.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeConfiguration.java @@ -15,13 +15,20 @@ */ package org.thingsboard.rule.engine.telemetry; +import jakarta.validation.constraints.NotNull; import lombok.Data; import org.thingsboard.rule.engine.api.NodeConfiguration; +import org.thingsboard.rule.engine.telemetry.settings.AttributesProcessingSettings; import org.thingsboard.server.common.data.DataConstants; +import static org.thingsboard.rule.engine.telemetry.settings.AttributesProcessingSettings.OnEveryMessage; + @Data public class TbMsgAttributesNodeConfiguration implements NodeConfiguration { + @NotNull + private AttributesProcessingSettings processingSettings; + private String scope; private boolean notifyDevice; @@ -31,6 +38,7 @@ public class TbMsgAttributesNodeConfiguration implements NodeConfiguration { @@ -39,7 +28,7 @@ public class TbMsgTimeseriesNodeConfiguration implements NodeConfiguration newAttributes = new ArrayList<>(); + void verifyDefaultConfig() { + assertThat(config.getProcessingSettings()).isInstanceOf(OnEveryMessage.class); + assertThat(config.getScope()).isEqualTo("SERVER_SCOPE"); + assertThat(config.isNotifyDevice()).isFalse(); + assertThat(config.isSendAttributesUpdatedNotification()).isFalse(); + assertThat(config.isUpdateAttributesOnlyOnValueChange()).isTrue(); + } - List filtered = node.filterChangedAttr(Collections.emptyList(), newAttributes); - assertThat(filtered).isSameAs(newAttributes); + @Test + void givenProcessingSettingsAreNull_whenValidatingConstraints_thenThrowsException() { + // GIVEN + config.setProcessingSettings(null); + + // 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) + .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) + .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 testFilterChangedAttr_whenCurrentAttributesContainsInAnyOrderNewAttributes_thenReturnExpectedList() { + 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) + .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() + )); + + 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) + .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) + )); + + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); + + long ts1 = 500L; + long ts2 = 1500L; + long ts3 = 2500L; + + // 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)) + )); + + 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(Strategy.PROCESS_ALL) + )); + } + + @Test + public void givenAdvancedProcessingSettingsWithSkipStrategiesForAllActionsAndSameMessageTwoTimes_whenOnMsg_thenSkipsSameMessageTwoTimes() throws TbNodeException { + // GIVEN + config.setProcessingSettings(new Advanced( + 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\"}")) - ); - 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 + 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 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); - 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); - })); + // THEN + then(telemetryServiceMock).should().saveAttributes(assertArg(request -> 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 de651e9841..c08808a4cb 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(); } @@ -223,7 +227,7 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest { var timeseriesStrategy = ProcessingStrategy.onEveryMessage(); var latestStrategy = ProcessingStrategy.skip(); var webSockets = ProcessingStrategy.onEveryMessage(); - var processingSettings = new TbMsgTimeseriesNodeConfiguration.ProcessingSettings.Advanced(timeseriesStrategy, latestStrategy, webSockets); + var processingSettings = new Advanced(timeseriesStrategy, latestStrategy, webSockets); config.setProcessingSettings(processingSettings); node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); @@ -335,7 +339,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))); @@ -370,7 +374,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))); @@ -405,7 +409,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))); @@ -440,7 +444,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() @@ -479,7 +483,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) @@ -530,7 +534,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() From f42e0e49b91f9587609e77f80ce0613286627627 Mon Sep 17 00:00:00 2001 From: Artem Dzhereleiko Date: Fri, 28 Feb 2025 15:16:13 +0200 Subject: [PATCH 2/9] UI: Processing settings for save attribute node --- ...dvanced-persistence-setting.component.html | 12 ++- .../advanced-persistence-setting.component.ts | 52 +++++++++-- .../action/attributes-config.component.html | 41 +++++++++ .../action/attributes-config.component.ts | 91 ++++++++++++++++++- .../action/attributes-config.model.ts | 59 ++++++++++++ .../action/timeseries-config.component.html | 1 + .../assets/locale/locale.constant-en_US.json | 16 ++++ 7 files changed, 257 insertions(+), 15 deletions(-) create mode 100644 ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.model.ts diff --git a/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-persistence-setting.component.html b/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-persistence-setting.component.html index 094f6dbc2a..f951a9f826 100644 --- a/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-persistence-setting.component.html +++ b/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-persistence-setting.component.html @@ -18,18 +18,22 @@
- - + - diff --git a/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-persistence-setting.component.ts b/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-persistence-setting.component.ts index 01314ec4f3..4609a7d010 100644 --- a/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-persistence-setting.component.ts +++ b/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-persistence-setting.component.ts @@ -19,12 +19,15 @@ import { FormBuilder, NG_VALIDATORS, NG_VALUE_ACCESSOR, + UntypedFormGroup, ValidationErrors, Validator } from '@angular/forms'; -import { Component, forwardRef } from '@angular/core'; +import { Component, DestroyRef, forwardRef, Input, OnInit } from '@angular/core'; import { takeUntilDestroyed } from '@angular/core/rxjs-interop'; import { AdvancedProcessingStrategy } from '@home/components/rule-node/action/timeseries-config.models'; +import { coerceBoolean } from '@shared/decorators/coercion'; +import { AttributeAdvancedProcessingStrategy } from '@home/components/rule-node/action/attributes-config.model'; @Component({ selector: 'tb-advanced-persistence-settings', @@ -39,19 +42,48 @@ import { AdvancedProcessingStrategy } from '@home/components/rule-node/action/ti multi: true }] }) -export class AdvancedPersistenceSettingComponent implements ControlValueAccessor, Validator { +export class AdvancedPersistenceSettingComponent implements OnInit, ControlValueAccessor, Validator { - persistenceForm = this.fb.group({ - timeseries: [null], - latest: [null], - webSockets: [null] - }); + @Input() + @coerceBoolean() + timeseries = false; + + @Input() + @coerceBoolean() + attribute = false; + + @Input() + @coerceBoolean() + latest = false; + + @Input() + @coerceBoolean() + webSockets = false; + + persistenceForm: UntypedFormGroup; private propagateChange: (value: any) => void = () => {}; - constructor(private fb: FormBuilder) { + constructor(private fb: FormBuilder, + private destroyRef: DestroyRef) { + } + + ngOnInit() { + this.persistenceForm = this.fb.group({}); + if (this.timeseries) { + this.persistenceForm.addControl('timeseries', this.fb.control(null, [])); + } + if (this.attribute) { + this.persistenceForm.addControl('attribute', this.fb.control(null, [])); + } + if (this.attribute) { + this.persistenceForm.addControl('latest', this.fb.control(null, [])); + } + if (this.attribute) { + this.persistenceForm.addControl('webSockets', this.fb.control(null, [])); + } this.persistenceForm.valueChanges.pipe( - takeUntilDestroyed() + takeUntilDestroyed(this.destroyRef) ).subscribe(value => this.propagateChange(value)); } @@ -76,7 +108,7 @@ export class AdvancedPersistenceSettingComponent implements ControlValueAccessor }; } - writeValue(value: AdvancedProcessingStrategy) { + writeValue(value: AdvancedProcessingStrategy | AttributeAdvancedProcessingStrategy) { this.persistenceForm.patchValue(value, {emitEvent: false}); } } diff --git a/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.component.html b/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.component.html index 03ea4fccdd..69feec1732 100644 --- a/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.component.html +++ b/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.component.html @@ -16,6 +16,47 @@ -->
+
+
+
+ rule-node-config.save-attribute.processing-settings +
+ + {{ 'rule-node-config.basic-mode' | translate}} + {{ 'rule-node-config.advanced-mode' | translate }} + +
+ @if(!attributesConfigForm.get('processingSettings.isAdvanced').value) { + + rule-node-config.save-attribute.strategy + + @for (strategy of persistenceStrategies; track strategy) { + {{ PersistenceTypeTranslationMap.get(strategy) | translate }} + } + + + + @if(attributesConfigForm.get('processingSettings.type').value === PersistenceType.DEDUPLICATE) { + + + } + } @else { + + } +
diff --git a/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.component.ts b/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.component.ts index 5a404c8de4..7257565608 100644 --- a/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.component.ts +++ b/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.component.ts @@ -15,10 +15,22 @@ /// import { Component } from '@angular/core'; -import { UntypedFormBuilder, UntypedFormGroup, Validators } from '@angular/forms'; +import { FormGroup, UntypedFormBuilder, UntypedFormGroup, Validators } from '@angular/forms'; import { RuleNodeConfiguration, RuleNodeConfigurationComponent } from '@shared/models/rule-node.models'; import { AttributeScope, telemetryTypeTranslations } from '@app/shared/models/telemetry/telemetry.models'; import { takeUntilDestroyed } from '@angular/core/rxjs-interop'; +import { + maxDeduplicateTimeSecs, + ProcessingSettings, + ProcessingSettingsForm, + ProcessingType, + ProcessingTypeTranslationMap +} from '@home/components/rule-node/action/timeseries-config.models'; +import { + AttributeNodeConfiguration, + AttributeNodeConfigurationForm, + defaultAttributeAdvancedPersistenceStrategy +} from '@home/components/rule-node/action/attributes-config.model'; @Component({ selector: 'tb-action-node-attributes-config', @@ -31,6 +43,12 @@ export class AttributesConfigComponent extends RuleNodeConfigurationComponent { attributeScopes = Object.keys(AttributeScope); telemetryTypeTranslationsMap = telemetryTypeTranslations; + PersistenceType = ProcessingType; + persistenceStrategies = [ProcessingType.ON_EVERY_MESSAGE, ProcessingType.DEDUPLICATE, ProcessingType.WEBSOCKETS_ONLY]; + PersistenceTypeTranslationMap = ProcessingTypeTranslationMap; + + maxDeduplicateTime = maxDeduplicateTimeSecs; + attributesConfigForm: UntypedFormGroup; constructor(private fb: UntypedFormBuilder) { @@ -41,8 +59,64 @@ export class AttributesConfigComponent extends RuleNodeConfigurationComponent { return this.attributesConfigForm; } + protected validatorTriggers(): string[] { + return ['processingSettings.isAdvanced', 'processingSettings.type']; + } + + protected prepareInputConfig(config: AttributeNodeConfiguration): AttributeNodeConfigurationForm { + let processingSettings: ProcessingSettingsForm; + if (config?.processingSettings) { + const isAdvanced = config?.processingSettings?.type === ProcessingType.ADVANCED; + processingSettings = { + type: isAdvanced ? ProcessingType.ON_EVERY_MESSAGE : config.processingSettings.type, + isAdvanced: isAdvanced, + deduplicationIntervalSecs: config.processingSettings?.deduplicationIntervalSecs ?? 60, + advanced: isAdvanced ? config.processingSettings : defaultAttributeAdvancedPersistenceStrategy + } + } else { + processingSettings = { + type: ProcessingType.ON_EVERY_MESSAGE, + isAdvanced: false, + deduplicationIntervalSecs: 60, + advanced: defaultAttributeAdvancedPersistenceStrategy + }; + } + return { + ...config, + processingSettings: processingSettings + } + } + + protected prepareOutputConfig(config: AttributeNodeConfigurationForm): AttributeNodeConfiguration { + let processingSettings: ProcessingSettings; + if (config.processingSettings.isAdvanced) { + processingSettings = { + ...config.processingSettings.advanced, + type: ProcessingType.ADVANCED + }; + } else { + processingSettings = { + type: config.processingSettings.type, + deduplicationIntervalSecs: config.processingSettings?.deduplicationIntervalSecs + }; + } + return { + ...config, + processingSettings + }; + } + protected onConfigurationSet(configuration: RuleNodeConfiguration) { this.attributesConfigForm = this.fb.group({ + processingSettings: this.fb.group({ + isAdvanced: [configuration?.processingSettings?.isAdvanced ?? false], + type: [configuration?.processingSettings?.type ?? ProcessingType.ON_EVERY_MESSAGE], + deduplicationIntervalSecs: [ + {value: configuration?.processingSettings?.deduplicationIntervalSecs ?? 60, disabled: true}, + [Validators.required, Validators.max(maxDeduplicateTimeSecs)] + ], + advanced: [{value: null, disabled: true}] + }), scope: [configuration ? configuration.scope : null, [Validators.required]], notifyDevice: [configuration ? configuration.notifyDevice : true, []], sendAttributesUpdatedNotification: [configuration ? configuration.sendAttributesUpdatedNotification : false, []], @@ -62,4 +136,19 @@ export class AttributesConfigComponent extends RuleNodeConfigurationComponent { }); } + protected updateValidators(emitEvent: boolean, _trigger?: string) { + const processingForm = this.attributesConfigForm.get('processingSettings') as FormGroup; + const isAdvanced: boolean = processingForm.get('isAdvanced').value; + const type: ProcessingType = processingForm.get('type').value; + if (!isAdvanced && type === ProcessingType.DEDUPLICATE) { + processingForm.get('deduplicationIntervalSecs').enable({emitEvent}); + } else { + processingForm.get('deduplicationIntervalSecs').disable({emitEvent}); + } + if (isAdvanced) { + processingForm.get('advanced').enable({emitEvent}); + } else { + processingForm.get('advanced').disable({emitEvent}); + } + } } diff --git a/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.model.ts b/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.model.ts new file mode 100644 index 0000000000..fb53a1ceaa --- /dev/null +++ b/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.model.ts @@ -0,0 +1,59 @@ +/// +/// 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. +/// + +import { DAY, SECOND } from '@shared/models/time/time.models'; +import { AttributeScope } from '@shared/models/telemetry/telemetry.models'; +import { BasicProcessingSettings, ProcessingType } from '@home/components/rule-node/action/timeseries-config.models'; + +export interface AttributeNodeConfiguration { + processingSettings: AttributeProcessingSettings; + scope: AttributeScope; + notifyDevice: boolean; + sendAttributesUpdatedNotification: boolean; + updateAttributesOnlyOnValueChange: boolean; +} + +export interface AttributeNodeConfigurationForm extends Omit { + processingSettings: AttributeProcessingSettingsForm +} + +export type AttributeProcessingSettings = BasicProcessingSettings & Partial & Partial; + +export type AttributeProcessingSettingsForm = Omit & { + isAdvanced: boolean; + advanced?: Partial; + type: ProcessingType; +}; + +export interface AttributeDeduplicateProcessingStrategy extends BasicProcessingSettings{ + deduplicationIntervalSecs: number; +} + +export interface AttributeAdvancedProcessingStrategy extends BasicProcessingSettings{ + attribute: AttributeAdvancedProcessingConfig; + webSockets: AttributeAdvancedProcessingConfig; +} + +export type AttributeAdvancedProcessingConfig = WithOptional; + +export const defaultAdvancedProcessingConfig: AttributeAdvancedProcessingConfig = { + type: ProcessingType.ON_EVERY_MESSAGE +} + +export const defaultAttributeAdvancedPersistenceStrategy: Omit = { + attribute: defaultAdvancedProcessingConfig, + webSockets: defaultAdvancedProcessingConfig, +} diff --git a/ui-ngx/src/app/modules/home/components/rule-node/action/timeseries-config.component.html b/ui-ngx/src/app/modules/home/components/rule-node/action/timeseries-config.component.html index 6e3e9fb6d1..250e66541b 100644 --- a/ui-ngx/src/app/modules/home/components/rule-node/action/timeseries-config.component.html +++ b/ui-ngx/src/app/modules/home/components/rule-node/action/timeseries-config.component.html @@ -53,6 +53,7 @@ }
diff --git a/ui-ngx/src/assets/locale/locale.constant-en_US.json b/ui-ngx/src/assets/locale/locale.constant-en_US.json index 18801bda2c..291cae525f 100644 --- a/ui-ngx/src/assets/locale/locale.constant-en_US.json +++ b/ui-ngx/src/assets/locale/locale.constant-en_US.json @@ -5155,6 +5155,22 @@ "latest": "Latest values", "web-sockets": "WebSockets" }, + "save-attribute": { + "processing-settings": "Processing settings", + "processing-settings-hint": "Define how incoming messages are processed. In Basic mode, select a preconfigured processing strategy or enable only WebSocket updates. Advanced mode allows you to select individual processing strategies for each action.", + "advanced-settings-hint": "Be cautious when configuring processing strategies. Certain combinations can lead to unexpected behavior.", + "strategy": "Strategy", + "deduplication-interval": "Deduplication interval", + "deduplication-interval-required": "Deduplication interval is required", + "deduplication-interval-min-max-range": "Deduplication interval should be at least 1 second and at most 1 day", + "strategy-type": { + "every-message": "On every message", + "skip": "Skip", + "deduplicate": "Deduplicate", + "web-sockets-only": "WebSockets only" + }, + "attribute": "Attribute" + }, "key-val": { "key": "Key", "value": "Value", From ed6262ba9f4d48aeae1c82c0c9fad0a9fd1f9482 Mon Sep 17 00:00:00 2001 From: Dmytro Skarzhynets Date: Mon, 3 Mar 2025 12:04:51 +0200 Subject: [PATCH 3/9] Save attributes strategies: fix UI bugs --- .../action/advanced-processing-setting.component.html | 4 ++-- .../action/advanced-processing-setting.component.ts | 10 +++++----- .../rule-node/action/attributes-config.component.html | 2 +- .../rule-node/action/attributes-config.model.ts | 4 ++-- 4 files changed, 10 insertions(+), 10 deletions(-) diff --git a/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-processing-setting.component.html b/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-processing-setting.component.html index 02d8f7be8f..02965fe535 100644 --- a/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-processing-setting.component.html +++ b/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-processing-setting.component.html @@ -25,8 +25,8 @@ formControlName="timeseries" title="{{ 'rule-node-config.save-time-series.time-series' | translate }}" > - } diff --git a/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.model.ts b/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.model.ts index 441fe9d816..1f24b69116 100644 --- a/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.model.ts +++ b/ui-ngx/src/app/modules/home/components/rule-node/action/attributes-config.model.ts @@ -42,7 +42,7 @@ export interface AttributeDeduplicateProcessingStrategy extends BasicProcessingS } export interface AttributeAdvancedProcessingStrategy extends BasicProcessingSettings { - attribute: AttributeAdvancedProcessingConfig; + attributes: AttributeAdvancedProcessingConfig; webSockets: AttributeAdvancedProcessingConfig; calculatedFields: AttributeAdvancedProcessingConfig; } @@ -54,7 +54,7 @@ export const defaultAdvancedProcessingConfig: AttributeAdvancedProcessingConfig } export const defaultAttributeAdvancedProcessingStrategy: Omit = { - attribute: defaultAdvancedProcessingConfig, + attributes: defaultAdvancedProcessingConfig, webSockets: defaultAdvancedProcessingConfig, calculatedFields: defaultAdvancedProcessingConfig, } From 2b7deda3d16bad64ea9754ed64e29a3d179a9c58 Mon Sep 17 00:00:00 2001 From: Dmytro Skarzhynets Date: Mon, 3 Mar 2025 12:26:20 +0200 Subject: [PATCH 4/9] Save attributes strategies: add advanced mode warning text; minor title correction --- ...advanced-processing-setting.component.html | 4 +-- .../rulenode/save_attributes_node_advanced.md | 32 +++++++++++++++++++ .../assets/locale/locale.constant-en_US.json | 2 +- 3 files changed, 35 insertions(+), 3 deletions(-) create mode 100644 ui-ngx/src/assets/help/en_US/rulenode/save_attributes_node_advanced.md diff --git a/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-processing-setting.component.html b/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-processing-setting.component.html index 02965fe535..edce8b12a9 100644 --- a/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-processing-setting.component.html +++ b/ui-ngx/src/app/modules/home/components/rule-node/action/advanced-processing-setting.component.html @@ -18,7 +18,7 @@
Date: Mon, 3 Mar 2025 14:47:51 +0200 Subject: [PATCH 5/9] Save attributes strategies: ensure shared attributes subscription update is sent even if WS updates are disabled --- .../DefaultSubscriptionManagerService.java | 4 - .../DefaultTelemetrySubscriptionService.java | 17 ++ ...faultTelemetrySubscriptionServiceTest.java | 205 ++++++++++++++++++ .../thingsboard/common/util/DonAsynchron.java | 6 + 4 files changed, 228 insertions(+), 4 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java index 23adacfc2e..d5a5318798 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java @@ -217,10 +217,6 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene 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(); 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 d60d1621b7..3af2c04643 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 @@ -35,9 +35,11 @@ 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.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; @@ -45,6 +47,7 @@ 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,6 +55,7 @@ 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.subscription.TbAttributeSubscriptionScope; import org.thingsboard.server.service.subscription.TbSubscriptionUtils; import java.util.ArrayList; @@ -197,6 +201,15 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer } }, t -> request.getCallback().onFailure(t)); + if (strategy.saveAttributes() + && entityId.getEntityType() == EntityType.DEVICE + && TbAttributeSubscriptionScope.SHARED_SCOPE.name().equalsIgnoreCase(request.getScope().name()) + && request.isNotifyDevice()) { + addMainCallback(resultFuture, success -> clusterService.pushMsgToCore( + DeviceAttributesEventNotificationMsg.onUpdate(tenantId, new DeviceId(entityId.getId()), DataConstants.SHARED_SCOPE, request.getEntries()), null + )); + } + if (strategy.sendWsUpdate()) { addWsCallback(resultFuture, success -> onAttributesUpdate(tenantId, entityId, request.getScope().name(), request.getEntries(), request.isNotifyDevice())); } @@ -342,6 +355,10 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer addMainCallback(saveFuture, result -> callback.onSuccess(null), callback::onFailure); } + private void addMainCallback(ListenableFuture saveFuture, Consumer onSuccess) { + DonAsynchron.withCallback(saveFuture, onSuccess, null, tsCallBackExecutor); + } + private void addMainCallback(ListenableFuture saveFuture, Consumer onSuccess, Consumer onFailure) { DonAsynchron.withCallback(saveFuture, onSuccess, onFailure, tsCallBackExecutor); } 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 c96b96b2e0..8bdcbc1349 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,6 +24,7 @@ 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; @@ -35,15 +36,21 @@ 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.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; @@ -51,6 +58,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; @@ -74,13 +82,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 { @@ -428,6 +440,199 @@ class DefaultTelemetrySubscriptionServiceTest { ); } + @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()); + } + // 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/util/src/main/java/org/thingsboard/common/util/DonAsynchron.java b/common/util/src/main/java/org/thingsboard/common/util/DonAsynchron.java index 008158246e..0f1a56cb17 100644 --- a/common/util/src/main/java/org/thingsboard/common/util/DonAsynchron.java +++ b/common/util/src/main/java/org/thingsboard/common/util/DonAsynchron.java @@ -36,6 +36,9 @@ public class DonAsynchron { FutureCallback callback = new FutureCallback() { @Override public void onSuccess(T result) { + if (onSuccess == null) { + return; + } try { onSuccess.accept(result); } catch (Throwable th) { @@ -45,6 +48,9 @@ public class DonAsynchron { @Override public void onFailure(Throwable t) { + if (onFailure == null) { + return; + } onFailure.accept(t); } }; From f318fa0ebdd4b0a86a9751bfc32edcc94905d231 Mon Sep 17 00:00:00 2001 From: Dmytro Skarzhynets Date: Mon, 3 Mar 2025 17:23:01 +0200 Subject: [PATCH 6/9] Save attributes strategies: handle attributes deleted subscription notification in DefaultTelemetrySubscriptionService --- .../main/data/upgrade/basic/schema_update.sql | 106 +++++------- .../queue/DefaultTbCoreConsumerService.java | 2 +- .../DefaultSubscriptionManagerService.java | 14 +- .../SubscriptionManagerService.java | 4 +- .../subscription/TbSubscriptionUtils.java | 3 +- .../DefaultTelemetrySubscriptionService.java | 36 +++-- ...faultTelemetrySubscriptionServiceTest.java | 152 +++++++++++++++++- common/proto/src/main/proto/queue.proto | 3 +- .../engine/api/AttributesDeleteRequest.java | 10 +- .../api/AttributesDeleteRequestTest.java | 39 +++++ 10 files changed, 268 insertions(+), 101 deletions(-) create mode 100644 rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/AttributesDeleteRequestTest.java diff --git a/application/src/main/data/upgrade/basic/schema_update.sql b/application/src/main/data/upgrade/basic/schema_update.sql index 215ac5248d..04dce447c5 100644 --- a/application/src/main/data/upgrade/basic/schema_update.sql +++ b/application/src/main/data/upgrade/basic/schema_update.sql @@ -16,78 +16,52 @@ -- 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); - - END IF; - END; -$$; +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 -- UPDATE SAVE ATTRIBUTES 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 - || 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; - - END IF; - 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 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 7219179b2c..0152e34d50 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 @@ -545,7 +545,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService 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) { @@ -223,16 +214,13 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene } @Override - public void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List keys, boolean notifyDevice, TbCallback callback) { + public void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List keys, 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); } } callback.onSuccess(); 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..3fd58a1243 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,9 +39,7 @@ 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, boolean notifyDevice, TbCallback empty); + void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List keys, 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/DefaultTelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java index 3af2c04643..252f696c72 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 @@ -211,7 +211,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer } if (strategy.sendWsUpdate()) { - addWsCallback(resultFuture, success -> onAttributesUpdate(tenantId, entityId, request.getScope().name(), request.getEntries(), request.isNotifyDevice())); + addWsCallback(resultFuture, success -> onAttributesUpdate(tenantId, entityId, request.getScope().name(), request.getEntries())); } } @@ -223,11 +223,25 @@ 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 (entityId.getEntityType() == EntityType.DEVICE + && TbAttributeSubscriptionScope.SHARED_SCOPE.name().equalsIgnoreCase(request.getScope().name()) + && request.isNotifyDevice()) { + addMainCallback(deleteFuture, success -> clusterService.pushMsgToCore( + DeviceAttributesEventNotificationMsg.onDelete(tenantId, new DeviceId(entityId.getId()), DataConstants.SHARED_SCOPE, request.getKeys()), null + )); + } + + addWsCallback(deleteFuture, success -> onAttributesDelete(tenantId, entityId, request.getScope().name(), request.getKeys())); } @Override @@ -312,16 +326,16 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer } } - private void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes, boolean notifyDevice) { + private void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes) { forwardToSubscriptionManagerService(tenantId, entityId, - subscriptionManagerService -> subscriptionManagerService.onAttributesUpdate(tenantId, entityId, scope, attributes, notifyDevice, TbCallback.EMPTY), + subscriptionManagerService -> subscriptionManagerService.onAttributesUpdate(tenantId, entityId, scope, attributes, TbCallback.EMPTY), () -> TbSubscriptionUtils.toAttributesUpdateProto(tenantId, entityId, scope, attributes)); } - private void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List keys, boolean notifyDevice) { + private void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List keys) { forwardToSubscriptionManagerService(tenantId, entityId, - subscriptionManagerService -> subscriptionManagerService.onAttributesDelete(tenantId, entityId, scope, keys, notifyDevice, TbCallback.EMPTY), - () -> TbSubscriptionUtils.toAttributesDeleteProto(tenantId, entityId, scope, keys, notifyDevice)); + subscriptionManagerService -> subscriptionManagerService.onAttributesDelete(tenantId, entityId, scope, keys, TbCallback.EMPTY), + () -> TbSubscriptionUtils.toAttributesDeleteProto(tenantId, entityId, scope, keys)); } private void onTimeSeriesUpdate(TenantId tenantId, EntityId entityId, List ts) { diff --git a/application/src/test/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionServiceTest.java b/application/src/test/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionServiceTest.java index 8bdcbc1349..44b7f146a5 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 @@ -29,6 +29,7 @@ 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.TimeseriesSaveRequest; import org.thingsboard.server.cluster.TbClusterService; @@ -421,7 +422,7 @@ class DefaultTelemetrySubscriptionServiceTest { } if (sendWsUpdate) { - then(subscriptionManagerService).should().onAttributesUpdate(tenantId, entityId, request.getScope().name(), request.getEntries(), request.isNotifyDevice(), TbCallback.EMPTY); + then(subscriptionManagerService).should().onAttributesUpdate(tenantId, entityId, request.getScope().name(), request.getEntries(), TbCallback.EMPTY); } else { then(subscriptionManagerService).shouldHaveNoInteractions(); } @@ -633,6 +634,155 @@ class DefaultTelemetrySubscriptionServiceTest { then(clusterService).should(never()).pushMsgToCore(any(), any()); } + /* --- 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()); + } + // used to emulate versions returned by save APIs private static List listOfNNumbers(int N) { return LongStream.range(0, N).boxed().toList(); diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 77dab2cb57..3007d48584 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -1079,7 +1079,8 @@ message TbAttributeDeleteProto { int64 tenantIdLSB = 5; string scope = 6; repeated string keys = 7; - bool notifyDevice = 8; + // not used anymore since device notification are now handled in DefaultTelemetrySubscriptionService instead of DefaultSubscriptionManagerService + bool notifyDevice = 8 [deprecated = true]; } message TbTimeSeriesDeleteProto { 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/test/java/org/thingsboard/rule/engine/api/AttributesDeleteRequestTest.java b/rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/AttributesDeleteRequestTest.java new file mode 100644 index 0000000000..9b4a825a66 --- /dev/null +++ b/rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/AttributesDeleteRequestTest.java @@ -0,0 +1,39 @@ +/** + * 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 AttributesDeleteRequestTest { + + @Test + 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()); + } + +} From 9d9fd93c89f7b23b976fd869b994f1aaa66c4617 Mon Sep 17 00:00:00 2001 From: Dmytro Skarzhynets Date: Mon, 3 Mar 2025 17:23:42 +0200 Subject: [PATCH 7/9] Save attributes strategies: improve proto deprecation note --- common/proto/src/main/proto/queue.proto | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 3007d48584..969b47f847 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -1079,7 +1079,7 @@ message TbAttributeDeleteProto { int64 tenantIdLSB = 5; string scope = 6; repeated string keys = 7; - // not used anymore since device notification are now handled in DefaultTelemetrySubscriptionService instead of DefaultSubscriptionManagerService + // Deprecated since 4.0, not used anymore since device notification are now handled in DefaultTelemetrySubscriptionService instead of DefaultSubscriptionManagerService bool notifyDevice = 8 [deprecated = true]; } From 264a767efeaa2df3b9aebc9fe15d451b5ec3f325 Mon Sep 17 00:00:00 2001 From: Dmytro Skarzhynets Date: Tue, 4 Mar 2025 11:44:41 +0200 Subject: [PATCH 8/9] Save attributes strategies: preserve backwards compatibility when deprecating 'notifyDevice' --- .../queue/DefaultTbCoreConsumerService.java | 17 +++++++++++++---- .../DefaultSubscriptionManagerService.java | 11 +++++++++++ .../SubscriptionManagerService.java | 9 +++++++++ common/proto/src/main/proto/queue.proto | 7 +++++-- 4 files changed, 38 insertions(+), 6 deletions(-) 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 0152e34d50..4f0522e62a 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 @@ -542,10 +542,19 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService 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); } } callback.onSuccess(); 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 3fd58a1243..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 @@ -41,6 +41,15 @@ public interface SubscriptionManagerService extends ApplicationListener 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); void onAlarmUpdate(TenantId tenantId, EntityId entityId, AlarmInfo alarm, TbCallback callback); diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 969b47f847..76477d7801 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -1079,8 +1079,11 @@ message TbAttributeDeleteProto { int64 tenantIdLSB = 5; string scope = 6; repeated string keys = 7; - // Deprecated since 4.0, not used anymore since device notification are now handled in DefaultTelemetrySubscriptionService instead of DefaultSubscriptionManagerService - bool notifyDevice = 8 [deprecated = true]; + // 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 { From 4495a2fa4b25f4b65c11ca08ecd9997924244f6d Mon Sep 17 00:00:00 2001 From: Dmytro Skarzhynets Date: Fri, 7 Mar 2025 11:53:25 +0200 Subject: [PATCH 9/9] Save attributes strategies: ensure device state service is notified when inactivity timeout is updated --- .../DefaultSubscriptionManagerService.java | 61 +--- .../DefaultTelemetrySubscriptionService.java | 83 ++++- ...faultTelemetrySubscriptionServiceTest.java | 299 +++++++++++++++++- 3 files changed, 374 insertions(+), 69 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java index 3e2c873030..1fafb562c1 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java @@ -48,8 +48,6 @@ import org.thingsboard.server.queue.discovery.event.OtherServiceShutdownEvent; import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; import org.thingsboard.server.queue.util.TbCoreComponent; -import org.thingsboard.server.service.state.DefaultDeviceStateService; -import org.thingsboard.server.service.state.DeviceStateService; import org.thingsboard.server.service.ws.notification.sub.NotificationUpdate; import org.thingsboard.server.service.ws.notification.sub.NotificationsSubscriptionUpdate; @@ -75,7 +73,6 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueProducerProvider producerProvider; private final TbLocalSubscriptionService localSubscriptionService; - private final DeviceStateService deviceStateService; private final TbClusterService clusterService; private final SubscriptionSchedulerComponent scheduler; @@ -170,7 +167,7 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene callback.onSuccess(); } - public void onTimeSeriesUpdate(EntityId entityId, List update) { + private void onTimeSeriesUpdate(EntityId entityId, List update) { getEntityUpdatesInfo(entityId).timeSeriesUpdateTs = System.currentTimeMillis(); TbEntityRemoteSubsInfo subInfo = entitySubscriptions.get(entityId); if (subInfo != null) { @@ -202,11 +199,6 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene public void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes, 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); - } - } callback.onSuccess(); } @@ -219,19 +211,13 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene 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); @@ -259,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); @@ -344,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/telemetry/DefaultTelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java index 0201bb4942..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,10 +31,12 @@ 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; @@ -43,6 +45,7 @@ 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; @@ -55,12 +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.subscription.TbAttributeSubscriptionScope; +import org.thingsboard.server.service.state.DefaultDeviceStateService; import org.thingsboard.server.service.subscription.TbSubscriptionUtils; import java.util.ArrayList; import java.util.Collections; -import java.util.Comparator; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -70,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. */ @@ -83,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; @@ -94,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 @@ -201,20 +211,51 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer } }, t -> request.getCallback().onFailure(t)); - if (strategy.saveAttributes() - && entityId.getEntityType() == EntityType.DEVICE - && TbAttributeSubscriptionScope.SHARED_SCOPE.name().equalsIgnoreCase(request.getScope().name()) - && request.isNotifyDevice()) { + 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 public void deleteAttributes(AttributesDeleteRequest request) { checkInternalEntity(request.getEntityId()); @@ -233,17 +274,37 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer t -> request.getCallback().onFailure(t) ); - if (entityId.getEntityType() == EntityType.DEVICE - && TbAttributeSubscriptionScope.SHARED_SCOPE.name().equalsIgnoreCase(request.getScope().name()) - && request.isNotifyDevice()) { + 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 public void deleteTimeseries(TimeseriesDeleteRequest request) { checkInternalEntity(request.getEntityId()); @@ -293,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); } } 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 e96566c116..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 @@ -31,6 +31,7 @@ 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; @@ -51,6 +52,7 @@ 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; @@ -137,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)); @@ -658,6 +662,184 @@ class DefaultTelemetrySubscriptionServiceTest { 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 @@ -807,6 +989,121 @@ class DefaultTelemetrySubscriptionServiceTest { 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();