tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Add review fixes

pull/9030/head
Dmytro Skarzhynets 3 years ago
parent
6b5b654546
commit
28879cb73b
10 changed files with 276 additions and 302 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 4
      application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
  2. 44
      application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
  3. 14
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
  4. 45
      application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java
  5. 6
      application/src/main/java/org/thingsboard/server/service/state/DeviceStateService.java
  6. 99
      application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java
  7. 3
      common/cluster-api/src/main/proto/queue.proto
  8. 47
      common/queue/src/main/java/org/thingsboard/server/queue/common/SimpleTbQueueCallback.java
  9. 78
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbDeviceStateNode.java
  10. 238
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbDeviceStateNodeTest.java

4
application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
View File

@ -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) {

44
application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
View File

@ -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<Throwable> onFailure;
public SimpleTbQueueCallback(Runnable onSuccess, Consumer<Throwable> 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);
}
}
}
}

14
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
View File

@ -605,10 +605,10 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
var tenantId = toTenantId(deviceConnectMsg.getTenantIdMSB(), deviceConnectMsg.getTenantIdLSB());
var deviceId = new DeviceId(new UUID(deviceConnectMsg.getDeviceIdMSB(), deviceConnectMsg.getDeviceIdLSB()));
try {
stateService.onDeviceConnect(tenantId, deviceId);
stateService.onDeviceConnect(tenantId, deviceId, deviceConnectMsg.getLastConnectTime());
callback.onSuccess();
} catch (Exception e) {
log.warn("Failed to process device connect message for device [{}]", deviceId, e);
log.warn("[{}] Failed to process device connect message for device [{}]", tenantId.getId(), deviceId.getId(), e);
callback.onFailure(e);
}
}
@ -623,7 +623,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
stateService.onDeviceActivity(tenantId, deviceId, deviceActivityMsg.getLastActivityTime());
callback.onSuccess();
} catch (Exception e) {
log.warn("Failed to process device activity message for device [{}]", deviceId, e);
log.warn("[{}] Failed to process device activity message for device [{}]", tenantId.getId(), deviceId.getId(), e);
callback.onFailure(e);
}
}
@ -632,10 +632,10 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
var tenantId = toTenantId(deviceDisconnectMsg.getTenantIdMSB(), deviceDisconnectMsg.getTenantIdLSB());
var deviceId = new DeviceId(new UUID(deviceDisconnectMsg.getDeviceIdMSB(), deviceDisconnectMsg.getDeviceIdLSB()));
try {
stateService.onDeviceDisconnect(tenantId, deviceId);
stateService.onDeviceDisconnect(tenantId, deviceId, deviceDisconnectMsg.getLastDisconnectTime());
callback.onSuccess();
} catch (Exception e) {
log.warn("Failed to process device activity message for device [{}]", deviceId, e);
log.warn("[{}] Failed to process device activity message for device [{}]", tenantId.getId(), deviceId.getId(), e);
callback.onFailure(e);
}
}
@ -644,10 +644,10 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
var tenantId = toTenantId(deviceInactivityMsg.getTenantIdMSB(), deviceInactivityMsg.getTenantIdLSB());
var deviceId = new DeviceId(new UUID(deviceInactivityMsg.getDeviceIdMSB(), deviceInactivityMsg.getDeviceIdLSB()));
try {
stateService.onDeviceInactivity(tenantId, deviceId);
stateService.onDeviceInactivity(tenantId, deviceId, deviceInactivityMsg.getLastInactivityTime());
callback.onSuccess();
} catch (Exception e) {
log.warn("Failed to process device inactivity message for device [{}]", deviceId, e);
log.warn("[{}] Failed to process device inactivity message for device [{}]", tenantId.getId(), deviceId.getId(), e);
callback.onFailure(e);
}
}

45
application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java
View File

@ -44,6 +44,7 @@ import org.thingsboard.server.common.data.exception.TenantNotFoundException;
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.id.UUIDBased;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.BooleanDataEntry;
@ -217,15 +218,14 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
}
@Override
public void onDeviceConnect(TenantId tenantId, DeviceId deviceId) {
public void onDeviceConnect(TenantId tenantId, DeviceId deviceId, long lastConnectTime) {
if (cleanDeviceStateIfBelongsToExternalPartition(tenantId, deviceId)) {
return;
}
log.trace("on Device Connect [{}]", deviceId.getId());
log.trace("[{}] on Device Connect [{}]", tenantId.getId(), deviceId.getId());
DeviceStateData stateData = getOrFetchDeviceStateData(deviceId);
long ts = System.currentTimeMillis();
stateData.getState().setLastConnectTime(ts);
save(deviceId, LAST_CONNECT_TIME, ts);
stateData.getState().setLastConnectTime(lastConnectTime);
save(deviceId, LAST_CONNECT_TIME, lastConnectTime);
pushRuleEngineMessage(stateData, TbMsgType.CONNECT_EVENT);
checkAndUpdateState(deviceId, stateData);
}
@ -235,7 +235,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
if (cleanDeviceStateIfBelongsToExternalPartition(tenantId, deviceId)) {
return;
}
log.trace("on Device Activity [{}], lastReportedActivity [{}]", deviceId.getId(), lastReportedActivity);
log.trace("[{}] on Device Activity [{}], lastReportedActivity [{}]", tenantId.getId(), deviceId.getId(), lastReportedActivity);
final DeviceStateData stateData = getOrFetchDeviceStateData(deviceId);
if (lastReportedActivity > 0 && lastReportedActivity > stateData.getState().getLastActivityTime()) {
updateActivityState(deviceId, stateData, lastReportedActivity);
@ -259,15 +259,14 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
}
@Override
public void onDeviceDisconnect(TenantId tenantId, DeviceId deviceId) {
public void onDeviceDisconnect(TenantId tenantId, DeviceId deviceId, long lastDisconnectTime) {
if (cleanDeviceStateIfBelongsToExternalPartition(tenantId, deviceId)) {
return;
}
log.trace("on Device Disconnect [{}]", deviceId.getId());
log.trace("[{}] on Device Disconnect [{}]", tenantId.getId(), deviceId.getId());
DeviceStateData stateData = getOrFetchDeviceStateData(deviceId);
long ts = System.currentTimeMillis();
stateData.getState().setLastDisconnectTime(ts);
save(deviceId, LAST_DISCONNECT_TIME, ts);
stateData.getState().setLastDisconnectTime(lastDisconnectTime);
save(deviceId, LAST_DISCONNECT_TIME, lastDisconnectTime);
pushRuleEngineMessage(stateData, TbMsgType.DISCONNECT_EVENT);
}
@ -279,20 +278,20 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
if (inactivityTimeout <= 0L) {
inactivityTimeout = defaultInactivityTimeoutMs;
}
log.trace("on Device Activity Timeout Update device id {} inactivityTimeout {}", deviceId, inactivityTimeout);
log.trace("[{}] on Device Activity Timeout Update device id {} inactivityTimeout {}", tenantId.getId(), deviceId.getId(), inactivityTimeout);
DeviceStateData stateData = getOrFetchDeviceStateData(deviceId);
stateData.getState().setInactivityTimeout(inactivityTimeout);
checkAndUpdateState(deviceId, stateData);
}
@Override
public void onDeviceInactivity(TenantId tenantId, DeviceId deviceId) {
public void onDeviceInactivity(TenantId tenantId, DeviceId deviceId, long lastInactivityTime) {
if (cleanDeviceStateIfBelongsToExternalPartition(tenantId, deviceId)) {
return;
}
log.trace("on Device Inactivity [{}]", deviceId.getId());
log.trace("[{}] on Device Inactivity [{}]", tenantId.getId(), deviceId.getId());
DeviceStateData stateData = getOrFetchDeviceStateData(deviceId);
reportInactivity(System.currentTimeMillis(), deviceId, stateData);
reportInactivity(lastInactivityTime, deviceId, stateData);
}
@Override
@ -532,13 +531,13 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
final Device device = deviceService.findDeviceById(TenantId.SYS_TENANT_ID, deviceId);
if (device == null) {
log.warn("[{}] Failed to fetch device by Id!", deviceId);
throw new RuntimeException("Failed to fetch device by Id " + deviceId);
throw new RuntimeException("Failed to fetch device by id [" + deviceId + "]!");
}
try {
return fetchDeviceState(device).get();
} catch (InterruptedException | ExecutionException e) {
log.warn("[{}] Failed to fetch device state!", deviceId, e);
throw new RuntimeException(e);
throw new RuntimeException("Failed to fetch device state for device [" + deviceId + "]");
}
}
@ -648,7 +647,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
return deviceStateData;
} catch (Exception e) {
log.warn("[{}] Failed to fetch device state data", device.getId(), e);
throw new RuntimeException(e);
throw new RuntimeException("Failed to fetch device state data for device [" + device.getId() + "]", e);
}
}
};
@ -672,8 +671,13 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
}
return success ? result : result.stream().filter(Objects::nonNull).collect(Collectors.toList());
} catch (InterruptedException | ExecutionException | TimeoutException e) {
log.warn("Failed to initialized device state futures for ids: {} due to:", deviceIds, e);
throw new RuntimeException(e);
String deviceIdsStr = deviceIds.stream()
.map(DeviceIdInfo::getDeviceId)
.map(UUIDBased::getId)
.map(UUID::toString)
.collect(Collectors.joining(", "));
log.warn("Failed to initialized device state futures for ids [{}] due to:", deviceIdsStr, e);
throw new RuntimeException("Failed to initialized device state futures for ids [" + deviceIdsStr + "]!", e);
}
}
@ -759,7 +763,6 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
return defaultValue;
}
private long getEntryValue(List<? extends KvEntry> kvEntries, String attributeName, long defaultValue) {
if (kvEntries != null) {
for (KvEntry entry : kvEntries) {

6
application/src/main/java/org/thingsboard/server/service/state/DeviceStateService.java
View File

@ -27,13 +27,13 @@ import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
*/
public interface DeviceStateService extends ApplicationListener<PartitionChangeEvent> {
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);

99
application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java
View File

@ -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);
}
}

3
common/cluster-api/src/main/proto/queue.proto
View File

@ -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.

47
common/queue/src/main/java/org/thingsboard/server/queue/common/SimpleTbQueueCallback.java
View File

@ -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<Throwable> onFailure;
public SimpleTbQueueCallback(Runnable onSuccess, Consumer<Throwable> 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);
}
}
}

78
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbDeviceStateNode.java
View File

@ -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 <code>Success</code> chain," +
" which sends appropriate message to the Rule Engine." +
" If metadata <code>ts</code> property is present, it will be used as event timestamp." +
" Incoming message is forwarded using the <code>Success</code> chain," +
" unless an unexpected error occurs during message processing" +
" then incoming message is forwarded using the <code>Failure</code> chain." +
"<br>" +
@ -59,19 +60,10 @@ import java.util.Map;
)
public class TbDeviceStateNode implements TbNode {
private final Map<TbMsgType, ConnectivityEvent> 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<TbMsgType> 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));
}
}

238
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbDeviceStateNodeTest.java
View File

@ -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<Arguments> 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())
);
}
}

Write Preview
Loading…
Cancel
Save