From 28879cb73bbc331514a6fe22f235f3be1fa3f79f Mon Sep 17 00:00:00 2001 From: Dmytro Skarzhynets Date: Tue, 19 Sep 2023 14:02:23 +0300 Subject: [PATCH] Add review fixes --- .../device/DeviceActorMessageProcessor.java | 4 +- .../actors/ruleChain/DefaultTbContext.java | 44 ++-- .../queue/DefaultTbCoreConsumerService.java | 14 +- .../state/DefaultDeviceStateService.java | 45 ++-- .../service/state/DeviceStateService.java | 6 +- .../state/DefaultDeviceStateServiceTest.java | 99 ++++---- common/cluster-api/src/main/proto/queue.proto | 3 + .../queue/common/SimpleTbQueueCallback.java | 47 ++++ .../rule/engine/action/TbDeviceStateNode.java | 78 +++--- .../engine/action/TbDeviceStateNodeTest.java | 238 ++++++------------ 10 files changed, 276 insertions(+), 302 deletions(-) create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/common/SimpleTbQueueCallback.java diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java index b838f0b86d..55262cbf63 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java @@ -490,11 +490,11 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } private void reportSessionOpen() { - systemContext.getDeviceStateService().onDeviceConnect(tenantId, deviceId); + systemContext.getDeviceStateService().onDeviceConnect(tenantId, deviceId, System.currentTimeMillis()); } private void reportSessionClose() { - systemContext.getDeviceStateService().onDeviceDisconnect(tenantId, deviceId); + systemContext.getDeviceStateService().onDeviceDisconnect(tenantId, deviceId, System.currentTimeMillis()); } private void handleGetAttributesRequest(SessionInfoProto sessionInfo, GetAttributeRequestMsg request) { diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index ca1d174322..ed07456ed6 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -102,8 +102,7 @@ import org.thingsboard.server.dao.user.UserService; import org.thingsboard.server.dao.widget.WidgetTypeService; import org.thingsboard.server.dao.widget.WidgetsBundleService; import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.queue.TbQueueCallback; -import org.thingsboard.server.queue.TbQueueMsgMetadata; +import org.thingsboard.server.queue.common.SimpleTbQueueCallback; import org.thingsboard.server.service.script.RuleNodeJsScriptEngine; import org.thingsboard.server.service.script.RuleNodeTbelScriptEngine; @@ -209,7 +208,13 @@ class DefaultTbContext implements TbContext { if (nodeCtx.getSelf().isDebugMode()) { mainCtx.persistDebugOutput(nodeCtx.getTenantId(), nodeCtx.getSelf().getId(), tbMsg, "To Root Rule Chain"); } - mainCtx.getClusterService().pushMsgToRuleEngine(tpi, tbMsg.getId(), msg, new SimpleTbQueueCallback(onSuccess, onFailure)); + mainCtx.getClusterService().pushMsgToRuleEngine(tpi, tbMsg.getId(), msg, new SimpleTbQueueCallback(onSuccess, t -> { + if (onFailure != null) { + onFailure.accept(t); + } else { + log.debug("[{}] Failed to put item into queue!", nodeCtx.getTenantId().getId(), t); + } + })); } @Override @@ -295,7 +300,13 @@ class DefaultTbContext implements TbContext { relationTypes.forEach(relationType -> mainCtx.persistDebugOutput(nodeCtx.getTenantId(), nodeCtx.getSelf().getId(), tbMsg, relationType, null, failureMessage)); } - mainCtx.getClusterService().pushMsgToRuleEngine(tpi, tbMsg.getId(), msg.build(), new SimpleTbQueueCallback(onSuccess, onFailure)); + mainCtx.getClusterService().pushMsgToRuleEngine(tpi, tbMsg.getId(), msg.build(), new SimpleTbQueueCallback(onSuccess, t -> { + if (onFailure != null) { + onFailure.accept(t); + } else { + log.debug("[{}] Failed to put item into queue!", nodeCtx.getTenantId().getId(), t); + } + })); } @Override @@ -923,29 +934,4 @@ class DefaultTbContext implements TbContext { return failureMessage; } - private class SimpleTbQueueCallback implements TbQueueCallback { - private final Runnable onSuccess; - private final Consumer onFailure; - - public SimpleTbQueueCallback(Runnable onSuccess, Consumer onFailure) { - this.onSuccess = onSuccess; - this.onFailure = onFailure; - } - - @Override - public void onSuccess(TbQueueMsgMetadata metadata) { - if (onSuccess != null) { - onSuccess.run(); - } - } - - @Override - public void onFailure(Throwable t) { - if (onFailure != null) { - onFailure.accept(t); - } else { - log.debug("[{}] Failed to put item into queue", nodeCtx.getTenantId(), t); - } - } - } } 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 8f3b0efc5b..7fb60f5da0 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 @@ -605,10 +605,10 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService 0 && lastReportedActivity > stateData.getState().getLastActivityTime()) { updateActivityState(deviceId, stateData, lastReportedActivity); @@ -259,15 +259,14 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService kvEntries, String attributeName, long defaultValue) { if (kvEntries != null) { for (KvEntry entry : kvEntries) { diff --git a/application/src/main/java/org/thingsboard/server/service/state/DeviceStateService.java b/application/src/main/java/org/thingsboard/server/service/state/DeviceStateService.java index b724b6e431..82c91d9baf 100644 --- a/application/src/main/java/org/thingsboard/server/service/state/DeviceStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/state/DeviceStateService.java @@ -27,13 +27,13 @@ import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; */ public interface DeviceStateService extends ApplicationListener { - void onDeviceConnect(TenantId tenantId, DeviceId deviceId); + void onDeviceConnect(TenantId tenantId, DeviceId deviceId, long lastConnectTime); void onDeviceActivity(TenantId tenantId, DeviceId deviceId, long lastReportedActivityTime); - void onDeviceDisconnect(TenantId tenantId, DeviceId deviceId); + void onDeviceDisconnect(TenantId tenantId, DeviceId deviceId, long lastDisconnectTime); - void onDeviceInactivity(TenantId tenantId, DeviceId deviceId); + void onDeviceInactivity(TenantId tenantId, DeviceId deviceId, long lastInactivityTime); void onDeviceInactivityTimeoutUpdate(TenantId tenantId, DeviceId deviceId, long inactivityTimeout); diff --git a/application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java b/application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java index 435fdbc523..6f99f3cfce 100644 --- a/application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java @@ -25,6 +25,7 @@ import org.mockito.Mockito; import org.mockito.junit.MockitoJUnitRunner; import org.springframework.test.util.ReflectionTestUtils; import org.thingsboard.server.cluster.TbClusterService; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.DeviceIdInfo; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.TenantId; @@ -54,11 +55,12 @@ import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; import static org.awaitility.Awaitility.await; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; @@ -128,7 +130,7 @@ public class DefaultDeviceStateServiceTest { .willReturn(TopicPartitionInfo.builder().build()); // WHEN - service.onDeviceInactivity(tenantId, deviceId); + service.onDeviceInactivity(tenantId, deviceId, System.currentTimeMillis()); // THEN assertThat(service.deviceStates).isEmpty(); @@ -139,9 +141,13 @@ public class DefaultDeviceStateServiceTest { } @Test - public void givenDeviceBelongsToMyPartition_whenOnDeviceInactivity_thenReportsInactivity() throws InterruptedException { + public void givenDeviceBelongsToMyPartition_whenOnDeviceInactivity_thenReportsInactivity() { // GIVEN - initStateService(10000); + try { + initStateService(10000); + } catch (InterruptedException e) { + fail("Device state service failed to initialize!"); + } var deviceStateData = DeviceStateData.builder() .tenantId(tenantId) .deviceId(deviceId) @@ -150,33 +156,30 @@ public class DefaultDeviceStateServiceTest { .build(); service.deviceStates.put(deviceId, deviceStateData); + long lastInactivityTime = System.currentTimeMillis(); // WHEN - long timeBeforeCall = System.currentTimeMillis(); - service.onDeviceInactivity(tenantId, deviceId); - long timeAfterCall = System.currentTimeMillis(); + service.onDeviceInactivity(tenantId, deviceId, lastInactivityTime); // THEN - var inactivityTimeCaptor = ArgumentCaptor.forClass(Long.class); - then(telemetrySubscriptionService).should().saveAttrAndNotify( - any(), eq(deviceId), any(), eq(INACTIVITY_ALARM_TIME), inactivityTimeCaptor.capture(), any() + then(telemetrySubscriptionService).should(times(1)).saveAttrAndNotify( + eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(DataConstants.SERVER_SCOPE), + eq(INACTIVITY_ALARM_TIME), eq(lastInactivityTime), any() ); - var actualInactivityTime = inactivityTimeCaptor.getValue(); - assertThat(actualInactivityTime).isGreaterThanOrEqualTo(timeBeforeCall); - assertThat(actualInactivityTime).isLessThanOrEqualTo(timeAfterCall); - - then(telemetrySubscriptionService).should().saveAttrAndNotify( - any(), eq(deviceId), any(), eq(ACTIVITY_STATE), eq(false), any() + then(telemetrySubscriptionService).should(times(1)).saveAttrAndNotify( + eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(DataConstants.SERVER_SCOPE), + eq(ACTIVITY_STATE), eq(false), any() ); var msgCaptor = ArgumentCaptor.forClass(TbMsg.class); - then(clusterService).should().pushMsgToRuleEngine(eq(tenantId), eq(deviceId), msgCaptor.capture(), any()); + then(clusterService).should(times(1)) + .pushMsgToRuleEngine(eq(tenantId), eq(deviceId), msgCaptor.capture(), any()); var actualMsg = msgCaptor.getValue(); assertThat(actualMsg.getType()).isEqualTo(TbMsgType.INACTIVITY_EVENT.name()); assertThat(actualMsg.getOriginator()).isEqualTo(deviceId); var notificationCaptor = ArgumentCaptor.forClass(DeviceActivityTrigger.class); - then(notificationRuleProcessor).should().process(notificationCaptor.capture()); + then(notificationRuleProcessor).should(times(1)).process(notificationCaptor.capture()); var actualNotification = notificationCaptor.getValue(); assertThat(actualNotification.getTenantId()).isEqualTo(tenantId); assertThat(actualNotification.getDeviceId()).isEqualTo(deviceId); @@ -199,21 +202,24 @@ public class DefaultDeviceStateServiceTest { service.updateInactivityStateIfExpired(System.currentTimeMillis(), deviceId, deviceStateData); // THEN - then(telemetrySubscriptionService).should().saveAttrAndNotify( - any(), eq(deviceId), any(), eq(INACTIVITY_ALARM_TIME), anyLong(), any() + then(telemetrySubscriptionService).should(times(1)).saveAttrAndNotify( + eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(DataConstants.SERVER_SCOPE), + eq(INACTIVITY_ALARM_TIME), anyLong(), any() ); - then(telemetrySubscriptionService).should().saveAttrAndNotify( - any(), eq(deviceId), any(), eq(ACTIVITY_STATE), eq(false), any() + then(telemetrySubscriptionService).should(times(1)).saveAttrAndNotify( + eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(DataConstants.SERVER_SCOPE), + eq(ACTIVITY_STATE), eq(false), any() ); var msgCaptor = ArgumentCaptor.forClass(TbMsg.class); - then(clusterService).should().pushMsgToRuleEngine(eq(tenantId), eq(deviceId), msgCaptor.capture(), any()); + then(clusterService).should(times(1)) + .pushMsgToRuleEngine(eq(tenantId), eq(deviceId), msgCaptor.capture(), any()); var actualMsg = msgCaptor.getValue(); assertThat(actualMsg.getType()).isEqualTo(TbMsgType.INACTIVITY_EVENT.name()); assertThat(actualMsg.getOriginator()).isEqualTo(deviceId); var notificationCaptor = ArgumentCaptor.forClass(DeviceActivityTrigger.class); - then(notificationRuleProcessor).should().process(notificationCaptor.capture()); + then(notificationRuleProcessor).should(times(1)).process(notificationCaptor.capture()); var actualNotification = notificationCaptor.getValue(); assertThat(actualNotification.getTenantId()).isEqualTo(tenantId); assertThat(actualNotification.getDeviceId()).isEqualTo(deviceId); @@ -461,35 +467,38 @@ public class DefaultDeviceStateServiceTest { @Test public void givenConcurrentAccess_whenGetOrFetchDeviceStateData_thenFetchDeviceStateDataInvokedOnce() { - var deviceStateData = DeviceStateData.builder().build(); - var getOrFetchInvocationCounter = new AtomicInteger(); - doAnswer(invocation -> { - getOrFetchInvocationCounter.incrementAndGet(); Thread.sleep(100); - return deviceStateData; + return deviceStateDataMock; }).when(service).fetchDeviceStateDataUsingSeparateRequests(deviceId); int numberOfThreads = 10; var allThreadsReadyLatch = new CountDownLatch(numberOfThreads); - var executor = Executors.newFixedThreadPool(numberOfThreads); - - for (int i = 0; i < numberOfThreads; i++) { - executor.submit(() -> { - allThreadsReadyLatch.countDown(); - try { - allThreadsReadyLatch.await(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - service.getOrFetchDeviceStateData(deviceId); - }); - } - executor.shutdown(); - await().atMost(10, TimeUnit.SECONDS).until(executor::isTerminated); + ExecutorService executor = null; + try { + executor = Executors.newFixedThreadPool(numberOfThreads); + for (int i = 0; i < numberOfThreads; i++) { + executor.submit(() -> { + allThreadsReadyLatch.countDown(); + try { + allThreadsReadyLatch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + service.getOrFetchDeviceStateData(deviceId); + }); + } + + executor.shutdown(); + await().atMost(10, TimeUnit.SECONDS).until(executor::isTerminated); + } finally { + if (executor != null) { + executor.shutdownNow(); + } + } - assertThat(getOrFetchInvocationCounter.get()).isEqualTo(1); + then(service).should(times(1)).fetchDeviceStateDataUsingSeparateRequests(deviceId); } } diff --git a/common/cluster-api/src/main/proto/queue.proto b/common/cluster-api/src/main/proto/queue.proto index 5260a4fa26..ef6a644c7f 100644 --- a/common/cluster-api/src/main/proto/queue.proto +++ b/common/cluster-api/src/main/proto/queue.proto @@ -485,6 +485,7 @@ message DeviceConnectProto { int64 tenantIdLSB = 2; int64 deviceIdMSB = 3; int64 deviceIdLSB = 4; + int64 lastConnectTime = 5; } message DeviceActivityProto { @@ -500,6 +501,7 @@ message DeviceDisconnectProto { int64 tenantIdLSB = 2; int64 deviceIdMSB = 3; int64 deviceIdLSB = 4; + int64 lastDisconnectTime = 5; } message DeviceInactivityProto { @@ -507,6 +509,7 @@ message DeviceInactivityProto { int64 tenantIdLSB = 2; int64 deviceIdMSB = 3; int64 deviceIdLSB = 4; + int64 lastInactivityTime = 5; } //Used to report session state to tb-Service and persist this state in the cache on the tb-Service level. diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/SimpleTbQueueCallback.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/SimpleTbQueueCallback.java new file mode 100644 index 0000000000..805511a603 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/SimpleTbQueueCallback.java @@ -0,0 +1,47 @@ +/** + * Copyright © 2016-2023 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.server.queue.common; + +import org.thingsboard.server.queue.TbQueueCallback; +import org.thingsboard.server.queue.TbQueueMsgMetadata; + +import java.util.function.Consumer; + +public class SimpleTbQueueCallback implements TbQueueCallback { + + private final Runnable onSuccess; + private final Consumer onFailure; + + public SimpleTbQueueCallback(Runnable onSuccess, Consumer onFailure) { + this.onSuccess = onSuccess; + this.onFailure = onFailure; + } + + @Override + public void onSuccess(TbQueueMsgMetadata metadata) { + if (onSuccess != null) { + onSuccess.run(); + } + } + + @Override + public void onFailure(Throwable t) { + if (onFailure != null) { + onFailure.accept(t); + } + } + +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbDeviceStateNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbDeviceStateNode.java index 8adafe8d62..37ecf34cdd 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbDeviceStateNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbDeviceStateNode.java @@ -28,9 +28,9 @@ import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.TbQueueCallback; -import org.thingsboard.server.queue.TbQueueMsgMetadata; +import org.thingsboard.server.queue.common.SimpleTbQueueCallback; -import java.util.Map; +import java.util.Set; @Slf4j @RuleNode( @@ -39,8 +39,9 @@ import java.util.Map; nodeDescription = "Triggers device connectivity events", nodeDetails = "If incoming message originator is a device," + " registers configured event for that device in the Device State Service," + - " which sends appropriate message to the Rule Engine. " + - "Incoming message is forwarded using the Success chain," + + " which sends appropriate message to the Rule Engine." + + " If metadata ts property is present, it will be used as event timestamp." + + " Incoming message is forwarded using the Success chain," + " unless an unexpected error occurs during message processing" + " then incoming message is forwarded using the Failure chain." + "
" + @@ -59,19 +60,10 @@ import java.util.Map; ) public class TbDeviceStateNode implements TbNode { - private final Map SUPPORTED_EVENTS = Map.of( - TbMsgType.CONNECT_EVENT, this::sendDeviceConnectMsg, - TbMsgType.ACTIVITY_EVENT, this::sendDeviceActivityMsg, - TbMsgType.DISCONNECT_EVENT, this::sendDeviceDisconnectMsg, - TbMsgType.INACTIVITY_EVENT, this::sendDeviceInactivityMsg + private static final Set SUPPORTED_EVENTS = Set.of( + TbMsgType.CONNECT_EVENT, TbMsgType.ACTIVITY_EVENT, TbMsgType.DISCONNECT_EVENT, TbMsgType.INACTIVITY_EVENT ); - private interface ConnectivityEvent { - - void sendEvent(TbContext ctx, TbMsg msg); - - } - private TbMsgType event; @Override @@ -80,7 +72,7 @@ public class TbDeviceStateNode implements TbNode { if (event == null) { throw new TbNodeException("Event cannot be null!", true); } - if (!SUPPORTED_EVENTS.containsKey(event)) { + if (!SUPPORTED_EVENTS.contains(event)) { throw new TbNodeException("Unsupported event: " + event, true); } this.event = event; @@ -90,15 +82,36 @@ public class TbDeviceStateNode implements TbNode { public void onMsg(TbContext ctx, TbMsg msg) { var originator = msg.getOriginator(); if (!ctx.isLocalEntity(originator)) { - log.warn("[{}][device-state-node] Received message from non-local entity [{}]!", ctx.getSelfId(), originator); + log.warn("[{}] Node [{}] received message from non-local entity [{}]!", + ctx.getTenantId().getId(), ctx.getSelfId().getId(), originator.getId()); + ctx.ack(msg); return; } if (!EntityType.DEVICE.equals(originator.getEntityType())) { ctx.tellSuccess(msg); return; } - SUPPORTED_EVENTS.get(event).sendEvent(ctx, msg); - ctx.tellSuccess(msg); + switch (event) { + case CONNECT_EVENT: { + sendDeviceConnectMsg(ctx, msg); + break; + } + case ACTIVITY_EVENT: { + sendDeviceActivityMsg(ctx, msg); + break; + } + case DISCONNECT_EVENT: { + sendDeviceDisconnectMsg(ctx, msg); + break; + } + case INACTIVITY_EVENT: { + sendDeviceInactivityMsg(ctx, msg); + break; + } + default: { + ctx.tellFailure(msg, new IllegalStateException("Configured event [" + event + "] is not supported!")); + } + } } private void sendDeviceConnectMsg(TbContext ctx, TbMsg msg) { @@ -109,12 +122,13 @@ public class TbDeviceStateNode implements TbNode { .setTenantIdLSB(tenantUuid.getLeastSignificantBits()) .setDeviceIdMSB(deviceUuid.getMostSignificantBits()) .setDeviceIdLSB(deviceUuid.getLeastSignificantBits()) + .setLastConnectTime(msg.getMetaDataTs()) .build(); var toCoreMsg = TransportProtos.ToCoreMsg.newBuilder() .setDeviceConnectMsg(deviceConnectMsg) .build(); ctx.getClusterService().pushMsgToCore( - ctx.getTenantId(), msg.getOriginator(), toCoreMsg, getMsgProcessedCallback(ctx, msg) + ctx.getTenantId(), msg.getOriginator(), toCoreMsg, getMsgEnqueuedCallback(ctx, msg) ); } @@ -126,13 +140,13 @@ public class TbDeviceStateNode implements TbNode { .setTenantIdLSB(tenantUuid.getLeastSignificantBits()) .setDeviceIdMSB(deviceUuid.getMostSignificantBits()) .setDeviceIdLSB(deviceUuid.getLeastSignificantBits()) - .setLastActivityTime(System.currentTimeMillis()) + .setLastActivityTime(msg.getMetaDataTs()) .build(); var toCoreMsg = TransportProtos.ToCoreMsg.newBuilder() .setDeviceActivityMsg(deviceActivityMsg) .build(); ctx.getClusterService().pushMsgToCore( - ctx.getTenantId(), msg.getOriginator(), toCoreMsg, getMsgProcessedCallback(ctx, msg) + ctx.getTenantId(), msg.getOriginator(), toCoreMsg, getMsgEnqueuedCallback(ctx, msg) ); } @@ -144,12 +158,13 @@ public class TbDeviceStateNode implements TbNode { .setTenantIdLSB(tenantUuid.getLeastSignificantBits()) .setDeviceIdMSB(deviceUuid.getMostSignificantBits()) .setDeviceIdLSB(deviceUuid.getLeastSignificantBits()) + .setLastDisconnectTime(msg.getMetaDataTs()) .build(); var toCoreMsg = TransportProtos.ToCoreMsg.newBuilder() .setDeviceDisconnectMsg(deviceDisconnectMsg) .build(); ctx.getClusterService().pushMsgToCore( - ctx.getTenantId(), msg.getOriginator(), toCoreMsg, getMsgProcessedCallback(ctx, msg) + ctx.getTenantId(), msg.getOriginator(), toCoreMsg, getMsgEnqueuedCallback(ctx, msg) ); } @@ -161,27 +176,18 @@ public class TbDeviceStateNode implements TbNode { .setTenantIdLSB(tenantUuid.getLeastSignificantBits()) .setDeviceIdMSB(deviceUuid.getMostSignificantBits()) .setDeviceIdLSB(deviceUuid.getLeastSignificantBits()) + .setLastInactivityTime(msg.getMetaDataTs()) .build(); var toCoreMsg = TransportProtos.ToCoreMsg.newBuilder() .setDeviceInactivityMsg(deviceInactivityMsg) .build(); ctx.getClusterService().pushMsgToCore( - ctx.getTenantId(), msg.getOriginator(), toCoreMsg, getMsgProcessedCallback(ctx, msg) + ctx.getTenantId(), msg.getOriginator(), toCoreMsg, getMsgEnqueuedCallback(ctx, msg) ); } - private TbQueueCallback getMsgProcessedCallback(TbContext ctx, TbMsg msg) { - return new TbQueueCallback() { - @Override - public void onSuccess(TbQueueMsgMetadata metadata) { - ctx.tellSuccess(msg); - } - - @Override - public void onFailure(Throwable t) { - ctx.tellFailure(msg, t); - } - }; + private TbQueueCallback getMsgEnqueuedCallback(TbContext ctx, TbMsg msg) { + return new SimpleTbQueueCallback(() -> ctx.tellSuccess(msg), t -> ctx.tellFailure(msg, t)); } } diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbDeviceStateNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbDeviceStateNodeTest.java index 123dffd0ea..176136244d 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbDeviceStateNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbDeviceStateNodeTest.java @@ -15,10 +15,12 @@ */ package org.thingsboard.rule.engine.action; -import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; 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.MethodSource; import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; @@ -31,18 +33,20 @@ import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceProfileId; +import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.queue.TbQueueCallback; import java.util.UUID; +import java.util.stream.Stream; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.assertj.core.api.Assertions.fail; -import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.then; @@ -51,9 +55,6 @@ import static org.mockito.Mockito.times; @ExtendWith(MockitoExtension.class) public class TbDeviceStateNodeTest { - private static TenantId DUMMY_TENANT_ID; - private static TbMsg DUMMY_MSG; - private static DeviceId DUMMY_MSG_ORIGINATOR; @Mock private TbContext ctxMock; @Mock @@ -61,26 +62,26 @@ public class TbDeviceStateNodeTest { private TbDeviceStateNode node; private TbDeviceStateNodeConfiguration config; - @BeforeAll - public static void init() { - DUMMY_TENANT_ID = TenantId.fromUUID(UUID.randomUUID()); + private static final TenantId TENANT_ID = TenantId.fromUUID(UUID.randomUUID()); + private static final DeviceId DEVICE_ID = new DeviceId(UUID.randomUUID()); + private static final long METADATA_TS = System.currentTimeMillis(); + private TbMsg msg; + @BeforeEach + public void setup() { var device = new Device(); - device.setTenantId(DUMMY_TENANT_ID); - device.setId(new DeviceId(UUID.randomUUID())); + device.setTenantId(TENANT_ID); + device.setId(DEVICE_ID); device.setName("My humidity sensor"); device.setType("Humidity sensor"); device.setDeviceProfileId(new DeviceProfileId(UUID.randomUUID())); var metaData = new TbMsgMetaData(); metaData.putValue("deviceName", device.getName()); metaData.putValue("deviceType", device.getType()); - metaData.putValue("ts", String.valueOf(System.currentTimeMillis())); + metaData.putValue("ts", String.valueOf(METADATA_TS)); var data = JacksonUtil.newObjectNode(); data.put("humidity", 58.3); - DUMMY_MSG = TbMsg.newMsg( - TbMsgType.POST_TELEMETRY_REQUEST, device.getId(), metaData, JacksonUtil.toString(data) - ); - DUMMY_MSG_ORIGINATOR = device.getId(); + msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, device.getId(), metaData, JacksonUtil.toString(data)); } @BeforeEach @@ -128,15 +129,15 @@ public class TbDeviceStateNodeTest { @Test public void givenNonDeviceOriginator_whenOnMsg_thenTellsSuccessAndNoActivityActionsTriggered() { // GIVEN - var originator = new AssetId(UUID.randomUUID()); - var msg = TbMsg.newMsg(TbMsgType.ENTITY_CREATED, originator, new TbMsgMetaData(), "{}"); - given(ctxMock.isLocalEntity(originator)).willReturn(true); + var asset = new AssetId(UUID.randomUUID()); + var msg = TbMsg.newMsg(TbMsgType.ENTITY_CREATED, asset, new TbMsgMetaData(), "{}"); + given(ctxMock.isLocalEntity(asset)).willReturn(true); // WHEN node.onMsg(ctxMock, msg); // THEN - then(ctxMock).should(times(1)).isLocalEntity(originator); + then(ctxMock).should(times(1)).isLocalEntity(asset); then(ctxMock).should(times(1)).tellSuccess(msg); then(ctxMock).shouldHaveNoMoreInteractions(); } @@ -144,178 +145,97 @@ public class TbDeviceStateNodeTest { @Test public void givenNonLocalOriginator_whenOnMsg_thenTellsSuccessAndNoActivityActionsTriggered() { // GIVEN - given(ctxMock.isLocalEntity(DUMMY_MSG.getOriginator())).willReturn(false); + given(ctxMock.isLocalEntity(msg.getOriginator())).willReturn(false); + given(ctxMock.getSelfId()).willReturn(new RuleNodeId(UUID.randomUUID())); + given(ctxMock.getTenantId()).willReturn(TENANT_ID); // WHEN - node.onMsg(ctxMock, DUMMY_MSG); + node.onMsg(ctxMock, msg); // THEN - then(ctxMock).should(times(1)).isLocalEntity(DUMMY_MSG_ORIGINATOR); + then(ctxMock).should(times(1)).isLocalEntity(DEVICE_ID); + then(ctxMock).should().getTenantId(); then(ctxMock).should().getSelfId(); + then(ctxMock).should(times(1)).ack(msg); then(ctxMock).shouldHaveNoMoreInteractions(); } - @Test - public void givenConnectEventInConfig_whenOnMsg_thenOnDeviceConnectCalledAndTellsSuccess() { + @ParameterizedTest + @MethodSource("provideSupportedEventsAndExpectedMessages") + public void givenSupportedEvent_whenOnMsg_thenCorrectMsgIsSent( + TbMsgType event, TransportProtos.ToCoreMsg expectedToCoreMsg + ) { // GIVEN - config.setEvent(TbMsgType.CONNECT_EVENT); + config.setEvent(event); var nodeConfig = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); try { node.init(ctxMock, nodeConfig); } catch (TbNodeException e) { fail("Node failed to initialize!", e); } - given(ctxMock.getTenantId()).willReturn(DUMMY_TENANT_ID); + given(ctxMock.getTenantId()).willReturn(TENANT_ID); given(ctxMock.getClusterService()).willReturn(tbClusterServiceMock); - given(ctxMock.isLocalEntity(DUMMY_MSG.getOriginator())).willReturn(true); + given(ctxMock.isLocalEntity(msg.getOriginator())).willReturn(true); // WHEN - node.onMsg(ctxMock, DUMMY_MSG); - - // THEN - var protoCaptor = ArgumentCaptor.forClass(TransportProtos.ToCoreMsg.class); - then(tbClusterServiceMock).should(times(1)) - .pushMsgToCore(eq(DUMMY_TENANT_ID), eq(DUMMY_MSG_ORIGINATOR), protoCaptor.capture(), any()); - - TransportProtos.DeviceConnectProto expectedDeviceConnectMsg = TransportProtos.DeviceConnectProto.newBuilder() - .setTenantIdMSB(DUMMY_TENANT_ID.getId().getMostSignificantBits()) - .setTenantIdLSB(DUMMY_TENANT_ID.getId().getLeastSignificantBits()) - .setDeviceIdMSB(DUMMY_MSG_ORIGINATOR.getId().getMostSignificantBits()) - .setDeviceIdLSB(DUMMY_MSG_ORIGINATOR.getId().getLeastSignificantBits()) - .build(); - TransportProtos.ToCoreMsg expectedToCoreMsg = TransportProtos.ToCoreMsg.newBuilder() - .setDeviceConnectMsg(expectedDeviceConnectMsg) - .build(); - assertThat(expectedToCoreMsg).isEqualTo(protoCaptor.getValue()); - - then(ctxMock).should(times(1)).tellSuccess(DUMMY_MSG); - then(tbClusterServiceMock).shouldHaveNoMoreInteractions(); - then(ctxMock).shouldHaveNoMoreInteractions(); - } - - @Test - public void givenActivityEventInConfig_whenOnMsg_thenOnDeviceActivityCalledWithCorrectTimeAndTellsSuccess() - throws TbNodeException { - // GIVEN - config.setEvent(TbMsgType.ACTIVITY_EVENT); - var nodeConfig = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); - node.init(ctxMock, nodeConfig); - given(ctxMock.getTenantId()).willReturn(DUMMY_TENANT_ID); - given(ctxMock.getClusterService()).willReturn(tbClusterServiceMock); - given(ctxMock.isLocalEntity(DUMMY_MSG.getOriginator())).willReturn(true); - - - // WHEN - long timeBeforeCall = System.currentTimeMillis(); - node.onMsg(ctxMock, DUMMY_MSG); - long timeAfterCall = System.currentTimeMillis(); + node.onMsg(ctxMock, msg); // THEN var protoCaptor = ArgumentCaptor.forClass(TransportProtos.ToCoreMsg.class); + var callbackCaptor = ArgumentCaptor.forClass(TbQueueCallback.class); then(tbClusterServiceMock).should(times(1)) - .pushMsgToCore(eq(DUMMY_TENANT_ID), eq(DUMMY_MSG_ORIGINATOR), protoCaptor.capture(), any()); - - TransportProtos.ToCoreMsg actualToCoreMsg = protoCaptor.getValue(); - long actualLastActivityTime = actualToCoreMsg.getDeviceActivityMsg().getLastActivityTime(); - - assertThat(actualLastActivityTime).isGreaterThanOrEqualTo(timeBeforeCall); - assertThat(actualLastActivityTime).isLessThanOrEqualTo(timeAfterCall); - - TransportProtos.DeviceActivityProto updatedActivityMsg = actualToCoreMsg.getDeviceActivityMsg().toBuilder() - .setLastActivityTime(123L) - .build(); - TransportProtos.ToCoreMsg updatedToCoreMsg = actualToCoreMsg.toBuilder() - .setDeviceActivityMsg(updatedActivityMsg) - .build(); + .pushMsgToCore(eq(TENANT_ID), eq(DEVICE_ID), protoCaptor.capture(), callbackCaptor.capture()); - TransportProtos.DeviceActivityProto expectedDeviceActivityMsg = TransportProtos.DeviceActivityProto.newBuilder() - .setTenantIdMSB(DUMMY_TENANT_ID.getId().getMostSignificantBits()) - .setTenantIdLSB(DUMMY_TENANT_ID.getId().getLeastSignificantBits()) - .setDeviceIdMSB(DUMMY_MSG_ORIGINATOR.getId().getMostSignificantBits()) - .setDeviceIdLSB(DUMMY_MSG_ORIGINATOR.getId().getLeastSignificantBits()) - .setLastActivityTime(123L) - .build(); - TransportProtos.ToCoreMsg expectedToCoreMsg = TransportProtos.ToCoreMsg.newBuilder() - .setDeviceActivityMsg(expectedDeviceActivityMsg) - .build(); + TbQueueCallback actualCallback = callbackCaptor.getValue(); - assertThat(updatedToCoreMsg).isEqualTo(expectedToCoreMsg); - - then(ctxMock).should(times(1)).tellSuccess(DUMMY_MSG); - then(tbClusterServiceMock).shouldHaveNoMoreInteractions(); - then(ctxMock).shouldHaveNoMoreInteractions(); - } - - @Test - public void givenInactivityEventInConfig_whenOnMsg_thenOnDeviceInactivityCalledAndTellsSuccess() - throws TbNodeException { - // GIVEN - config.setEvent(TbMsgType.INACTIVITY_EVENT); - var nodeConfig = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); - node.init(ctxMock, nodeConfig); - given(ctxMock.getTenantId()).willReturn(DUMMY_TENANT_ID); - given(ctxMock.getClusterService()).willReturn(tbClusterServiceMock); - given(ctxMock.isLocalEntity(DUMMY_MSG.getOriginator())).willReturn(true); - - // WHEN - node.onMsg(ctxMock, DUMMY_MSG); + actualCallback.onSuccess(null); + then(ctxMock).should(times(1)).tellSuccess(msg); - // THEN - var protoCaptor = ArgumentCaptor.forClass(TransportProtos.ToCoreMsg.class); - then(tbClusterServiceMock).should(times(1)) - .pushMsgToCore(eq(DUMMY_TENANT_ID), eq(DUMMY_MSG_ORIGINATOR), protoCaptor.capture(), any()); + var throwable = new Throwable(); + actualCallback.onFailure(throwable); + then(ctxMock).should(times(1)).tellFailure(msg, throwable); - TransportProtos.DeviceInactivityProto expectedDeviceInactivityMsg = - TransportProtos.DeviceInactivityProto.newBuilder() - .setTenantIdMSB(DUMMY_TENANT_ID.getId().getMostSignificantBits()) - .setTenantIdLSB(DUMMY_TENANT_ID.getId().getLeastSignificantBits()) - .setDeviceIdMSB(DUMMY_MSG_ORIGINATOR.getId().getMostSignificantBits()) - .setDeviceIdLSB(DUMMY_MSG_ORIGINATOR.getId().getLeastSignificantBits()) - .build(); - TransportProtos.ToCoreMsg expectedToCoreMsg = TransportProtos.ToCoreMsg.newBuilder() - .setDeviceInactivityMsg(expectedDeviceInactivityMsg) - .build(); assertThat(expectedToCoreMsg).isEqualTo(protoCaptor.getValue()); - then(ctxMock).should(times(1)).tellSuccess(DUMMY_MSG); then(tbClusterServiceMock).shouldHaveNoMoreInteractions(); then(ctxMock).shouldHaveNoMoreInteractions(); } - @Test - public void givenDisconnectEventInConfig_whenOnMsg_thenOnDeviceDisconnectCalledAndTellsSuccess() - throws TbNodeException { - // GIVEN - config.setEvent(TbMsgType.DISCONNECT_EVENT); - var nodeConfig = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); - node.init(ctxMock, nodeConfig); - given(ctxMock.getTenantId()).willReturn(DUMMY_TENANT_ID); - given(ctxMock.getClusterService()).willReturn(tbClusterServiceMock); - given(ctxMock.isLocalEntity(DUMMY_MSG.getOriginator())).willReturn(true); - - // WHEN - node.onMsg(ctxMock, DUMMY_MSG); - - // THEN - var protoCaptor = ArgumentCaptor.forClass(TransportProtos.ToCoreMsg.class); - then(tbClusterServiceMock).should(times(1)) - .pushMsgToCore(eq(DUMMY_TENANT_ID), eq(DUMMY_MSG_ORIGINATOR), protoCaptor.capture(), any()); - - TransportProtos.DeviceDisconnectProto expectedDeviceDisconnectMsg = - TransportProtos.DeviceDisconnectProto.newBuilder() - .setTenantIdMSB(DUMMY_TENANT_ID.getId().getMostSignificantBits()) - .setTenantIdLSB(DUMMY_TENANT_ID.getId().getLeastSignificantBits()) - .setDeviceIdMSB(DUMMY_MSG_ORIGINATOR.getId().getMostSignificantBits()) - .setDeviceIdLSB(DUMMY_MSG_ORIGINATOR.getId().getLeastSignificantBits()) - .build(); - TransportProtos.ToCoreMsg expectedToCoreMsg = TransportProtos.ToCoreMsg.newBuilder() - .setDeviceDisconnectMsg(expectedDeviceDisconnectMsg) - .build(); - assertThat(expectedToCoreMsg).isEqualTo(protoCaptor.getValue()); - - then(ctxMock).should(times(1)).tellSuccess(DUMMY_MSG); - then(tbClusterServiceMock).shouldHaveNoMoreInteractions(); - then(ctxMock).shouldHaveNoMoreInteractions(); + private static Stream provideSupportedEventsAndExpectedMessages() { + return Stream.of( + Arguments.of(TbMsgType.CONNECT_EVENT, TransportProtos.ToCoreMsg.newBuilder().setDeviceConnectMsg( + TransportProtos.DeviceConnectProto.newBuilder() + .setTenantIdMSB(TENANT_ID.getId().getMostSignificantBits()) + .setTenantIdLSB(TENANT_ID.getId().getLeastSignificantBits()) + .setDeviceIdMSB(DEVICE_ID.getId().getMostSignificantBits()) + .setDeviceIdLSB(DEVICE_ID.getId().getLeastSignificantBits()) + .setLastConnectTime(METADATA_TS) + .build()).build()), + Arguments.of(TbMsgType.ACTIVITY_EVENT, TransportProtos.ToCoreMsg.newBuilder().setDeviceActivityMsg( + TransportProtos.DeviceActivityProto.newBuilder() + .setTenantIdMSB(TENANT_ID.getId().getMostSignificantBits()) + .setTenantIdLSB(TENANT_ID.getId().getLeastSignificantBits()) + .setDeviceIdMSB(DEVICE_ID.getId().getMostSignificantBits()) + .setDeviceIdLSB(DEVICE_ID.getId().getLeastSignificantBits()) + .setLastActivityTime(METADATA_TS) + .build()).build()), + Arguments.of(TbMsgType.DISCONNECT_EVENT, TransportProtos.ToCoreMsg.newBuilder().setDeviceDisconnectMsg( + TransportProtos.DeviceDisconnectProto.newBuilder() + .setTenantIdMSB(TENANT_ID.getId().getMostSignificantBits()) + .setTenantIdLSB(TENANT_ID.getId().getLeastSignificantBits()) + .setDeviceIdMSB(DEVICE_ID.getId().getMostSignificantBits()) + .setDeviceIdLSB(DEVICE_ID.getId().getLeastSignificantBits()) + .setLastDisconnectTime(METADATA_TS) + .build()).build()), + Arguments.of(TbMsgType.INACTIVITY_EVENT, TransportProtos.ToCoreMsg.newBuilder().setDeviceInactivityMsg( + TransportProtos.DeviceInactivityProto.newBuilder() + .setTenantIdMSB(TENANT_ID.getId().getMostSignificantBits()) + .setTenantIdLSB(TENANT_ID.getId().getLeastSignificantBits()) + .setDeviceIdMSB(DEVICE_ID.getId().getMostSignificantBits()) + .setDeviceIdLSB(DEVICE_ID.getId().getLeastSignificantBits()) + .setLastInactivityTime(METADATA_TS) + .build()).build()) + ); } }