Artem Dzhereleiko
3 years ago
262 changed files with 17213 additions and 1461 deletions
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@ -0,0 +1,62 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.service.notification.rule.trigger; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import org.apache.commons.collections.CollectionUtils; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.notification.info.EdgeCommunicationFailureNotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.info.RuleOriginatedNotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.EdgeCommunicationFailureTrigger; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.config.EdgeCommunicationFailureNotificationRuleTriggerConfig; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.config.NotificationRuleTriggerType; |
|||
|
|||
@Service |
|||
@RequiredArgsConstructor |
|||
public class EdgeCommunicationFailureTriggerProcessor implements NotificationRuleTriggerProcessor<EdgeCommunicationFailureTrigger, EdgeCommunicationFailureNotificationRuleTriggerConfig> { |
|||
|
|||
@Override |
|||
public boolean matchesFilter(EdgeCommunicationFailureTrigger trigger, EdgeCommunicationFailureNotificationRuleTriggerConfig triggerConfig) { |
|||
if (CollectionUtils.isNotEmpty(triggerConfig.getEdges())) { |
|||
return !triggerConfig.getEdges().contains(trigger.getEdgeId().getId()); |
|||
} |
|||
return true; |
|||
} |
|||
|
|||
@Override |
|||
public RuleOriginatedNotificationInfo constructNotificationInfo(EdgeCommunicationFailureTrigger trigger) { |
|||
return EdgeCommunicationFailureNotificationInfo.builder() |
|||
.tenantId(trigger.getTenantId()) |
|||
.edgeId(trigger.getEdgeId()) |
|||
.customerId(trigger.getCustomerId()) |
|||
.edgeName(trigger.getEdgeName()) |
|||
.failureMsg(truncateFailureMsg(trigger.getFailureMsg())) |
|||
.build(); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationRuleTriggerType getTriggerType() { |
|||
return NotificationRuleTriggerType.EDGE_COMMUNICATION_FAILURE; |
|||
} |
|||
|
|||
private String truncateFailureMsg(String input) { |
|||
int maxLength = 500; |
|||
if (input != null && input.length() > maxLength) { |
|||
return input.substring(0, maxLength); |
|||
} |
|||
return input; |
|||
} |
|||
} |
|||
@ -0,0 +1,60 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.service.notification.rule.trigger; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import org.apache.commons.collections.CollectionUtils; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.notification.info.EdgeConnectionNotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.info.RuleOriginatedNotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.EdgeConnectionTrigger; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.config.EdgeConnectionNotificationRuleTriggerConfig; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.config.EdgeConnectionNotificationRuleTriggerConfig.EdgeConnectivityEvent; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.config.NotificationRuleTriggerType; |
|||
|
|||
@Service |
|||
@RequiredArgsConstructor |
|||
public class EdgeConnectionTriggerProcessor implements NotificationRuleTriggerProcessor<EdgeConnectionTrigger, EdgeConnectionNotificationRuleTriggerConfig> { |
|||
|
|||
@Override |
|||
public boolean matchesFilter(EdgeConnectionTrigger trigger, EdgeConnectionNotificationRuleTriggerConfig triggerConfig) { |
|||
EdgeConnectivityEvent event = trigger.isConnected() ? EdgeConnectivityEvent.CONNECTED : EdgeConnectivityEvent.DISCONNECTED; |
|||
if (CollectionUtils.isEmpty(triggerConfig.getNotifyOn()) || !triggerConfig.getNotifyOn().contains(event)) { |
|||
return false; |
|||
} |
|||
if (CollectionUtils.isNotEmpty(triggerConfig.getEdges())) { |
|||
return triggerConfig.getEdges().contains(trigger.getEdgeId().getId()); |
|||
} |
|||
return true; |
|||
} |
|||
|
|||
@Override |
|||
public RuleOriginatedNotificationInfo constructNotificationInfo(EdgeConnectionTrigger trigger) { |
|||
return EdgeConnectionNotificationInfo.builder() |
|||
.eventType(trigger.isConnected() ? "connected" : "disconnected") |
|||
.tenantId(trigger.getTenantId()) |
|||
.customerId(trigger.getCustomerId()) |
|||
.edgeId(trigger.getEdgeId()) |
|||
.edgeName(trigger.getEdgeName()) |
|||
.build(); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationRuleTriggerType getTriggerType() { |
|||
return NotificationRuleTriggerType.EDGE_CONNECTION; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,196 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.service.state; |
|||
|
|||
import lombok.Getter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.rule.engine.api.RuleEngineDeviceStateManager; |
|||
import org.thingsboard.server.cluster.TbClusterService; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
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.gen.transport.TransportProtos; |
|||
import org.thingsboard.server.queue.common.SimpleTbQueueCallback; |
|||
import org.thingsboard.server.queue.discovery.PartitionService; |
|||
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; |
|||
import org.thingsboard.server.queue.util.TbRuleEngineComponent; |
|||
|
|||
import java.util.Optional; |
|||
import java.util.UUID; |
|||
|
|||
@Slf4j |
|||
@Service |
|||
@TbRuleEngineComponent |
|||
public class DefaultRuleEngineDeviceStateManager implements RuleEngineDeviceStateManager { |
|||
|
|||
private final TbServiceInfoProvider serviceInfoProvider; |
|||
private final PartitionService partitionService; |
|||
|
|||
private final Optional<DeviceStateService> deviceStateService; |
|||
private final TbClusterService clusterService; |
|||
|
|||
public DefaultRuleEngineDeviceStateManager( |
|||
TbServiceInfoProvider serviceInfoProvider, PartitionService partitionService, |
|||
Optional<DeviceStateService> deviceStateServiceOptional, TbClusterService clusterService |
|||
) { |
|||
this.serviceInfoProvider = serviceInfoProvider; |
|||
this.partitionService = partitionService; |
|||
this.deviceStateService = deviceStateServiceOptional; |
|||
this.clusterService = clusterService; |
|||
} |
|||
|
|||
@Getter |
|||
private abstract static class ConnectivityEventInfo { |
|||
|
|||
private final TenantId tenantId; |
|||
private final DeviceId deviceId; |
|||
private final long eventTime; |
|||
|
|||
private ConnectivityEventInfo(TenantId tenantId, DeviceId deviceId, long eventTime) { |
|||
this.tenantId = tenantId; |
|||
this.deviceId = deviceId; |
|||
this.eventTime = eventTime; |
|||
} |
|||
|
|||
abstract void forwardToLocalService(); |
|||
|
|||
abstract TransportProtos.ToCoreMsg toQueueMsg(); |
|||
|
|||
} |
|||
|
|||
@Override |
|||
public void onDeviceConnect(TenantId tenantId, DeviceId deviceId, long connectTime, TbCallback callback) { |
|||
routeEvent(new ConnectivityEventInfo(tenantId, deviceId, connectTime) { |
|||
@Override |
|||
void forwardToLocalService() { |
|||
deviceStateService.ifPresent(service -> service.onDeviceConnect(tenantId, deviceId, connectTime)); |
|||
} |
|||
|
|||
@Override |
|||
TransportProtos.ToCoreMsg toQueueMsg() { |
|||
var deviceConnectMsg = TransportProtos.DeviceConnectProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastConnectTime(connectTime) |
|||
.build(); |
|||
return TransportProtos.ToCoreMsg.newBuilder() |
|||
.setDeviceConnectMsg(deviceConnectMsg) |
|||
.build(); |
|||
} |
|||
}, callback); |
|||
} |
|||
|
|||
@Override |
|||
public void onDeviceActivity(TenantId tenantId, DeviceId deviceId, long activityTime, TbCallback callback) { |
|||
routeEvent(new ConnectivityEventInfo(tenantId, deviceId, activityTime) { |
|||
@Override |
|||
void forwardToLocalService() { |
|||
deviceStateService.ifPresent(service -> service.onDeviceActivity(tenantId, deviceId, activityTime)); |
|||
} |
|||
|
|||
@Override |
|||
TransportProtos.ToCoreMsg toQueueMsg() { |
|||
var deviceActivityMsg = TransportProtos.DeviceActivityProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastActivityTime(activityTime) |
|||
.build(); |
|||
return TransportProtos.ToCoreMsg.newBuilder() |
|||
.setDeviceActivityMsg(deviceActivityMsg) |
|||
.build(); |
|||
} |
|||
}, callback); |
|||
} |
|||
|
|||
@Override |
|||
public void onDeviceDisconnect(TenantId tenantId, DeviceId deviceId, long disconnectTime, TbCallback callback) { |
|||
routeEvent(new ConnectivityEventInfo(tenantId, deviceId, disconnectTime) { |
|||
@Override |
|||
void forwardToLocalService() { |
|||
deviceStateService.ifPresent(service -> service.onDeviceDisconnect(tenantId, deviceId, disconnectTime)); |
|||
} |
|||
|
|||
@Override |
|||
TransportProtos.ToCoreMsg toQueueMsg() { |
|||
var deviceDisconnectMsg = TransportProtos.DeviceDisconnectProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastDisconnectTime(disconnectTime) |
|||
.build(); |
|||
return TransportProtos.ToCoreMsg.newBuilder() |
|||
.setDeviceDisconnectMsg(deviceDisconnectMsg) |
|||
.build(); |
|||
} |
|||
}, callback); |
|||
} |
|||
|
|||
@Override |
|||
public void onDeviceInactivity(TenantId tenantId, DeviceId deviceId, long inactivityTime, TbCallback callback) { |
|||
routeEvent(new ConnectivityEventInfo(tenantId, deviceId, inactivityTime) { |
|||
@Override |
|||
void forwardToLocalService() { |
|||
deviceStateService.ifPresent(service -> service.onDeviceInactivity(tenantId, deviceId, inactivityTime)); |
|||
} |
|||
|
|||
@Override |
|||
TransportProtos.ToCoreMsg toQueueMsg() { |
|||
var deviceInactivityMsg = TransportProtos.DeviceInactivityProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastInactivityTime(inactivityTime) |
|||
.build(); |
|||
return TransportProtos.ToCoreMsg.newBuilder() |
|||
.setDeviceInactivityMsg(deviceInactivityMsg) |
|||
.build(); |
|||
} |
|||
}, callback); |
|||
} |
|||
|
|||
private void routeEvent(ConnectivityEventInfo eventInfo, TbCallback callback) { |
|||
var tenantId = eventInfo.getTenantId(); |
|||
var deviceId = eventInfo.getDeviceId(); |
|||
long eventTime = eventInfo.getEventTime(); |
|||
|
|||
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, deviceId); |
|||
if (serviceInfoProvider.isService(ServiceType.TB_CORE) && tpi.isMyPartition() && deviceStateService.isPresent()) { |
|||
log.debug("[{}][{}] Forwarding device connectivity event to local service. Event time: [{}].", tenantId.getId(), deviceId.getId(), eventTime); |
|||
try { |
|||
eventInfo.forwardToLocalService(); |
|||
} catch (Exception e) { |
|||
log.error("[{}][{}] Failed to process device connectivity event. Event time: [{}].", tenantId.getId(), deviceId.getId(), eventTime, e); |
|||
callback.onFailure(e); |
|||
return; |
|||
} |
|||
callback.onSuccess(); |
|||
} else { |
|||
TransportProtos.ToCoreMsg msg = eventInfo.toQueueMsg(); |
|||
log.debug("[{}][{}] Sending device connectivity message to core. Event time: [{}].", tenantId.getId(), deviceId.getId(), eventTime); |
|||
clusterService.pushMsgToCore(tpi, UUID.randomUUID(), msg, new SimpleTbQueueCallback(__ -> callback.onSuccess(), callback::onFailure)); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,38 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.service.ws.notification.sub; |
|||
|
|||
|
|||
import lombok.Getter; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.service.subscription.TbSubscription; |
|||
import org.thingsboard.server.service.subscription.TbSubscriptionType; |
|||
|
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
import java.util.function.BiConsumer; |
|||
|
|||
@Getter |
|||
public abstract class AbstractNotificationSubscription<T> extends TbSubscription<T> { |
|||
|
|||
protected final AtomicInteger sequence = new AtomicInteger(); |
|||
protected final AtomicInteger totalUnreadCounter = new AtomicInteger(); |
|||
|
|||
public AbstractNotificationSubscription(String serviceId, String sessionId, int subscriptionId, TenantId tenantId, EntityId entityId, TbSubscriptionType type, BiConsumer<TbSubscription<T>, T> updateProcessor) { |
|||
super(serviceId, sessionId, subscriptionId, tenantId, entityId, type, updateProcessor); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,532 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.service.queue; |
|||
|
|||
import com.google.common.util.concurrent.ListeningExecutorService; |
|||
import com.google.common.util.concurrent.MoreExecutors; |
|||
import org.junit.jupiter.api.AfterEach; |
|||
import org.junit.jupiter.api.BeforeEach; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.api.extension.ExtendWith; |
|||
import org.mockito.ArgumentCaptor; |
|||
import org.mockito.Mock; |
|||
import org.mockito.junit.jupiter.MockitoExtension; |
|||
import org.springframework.test.util.ReflectionTestUtils; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.queue.TbCallback; |
|||
import org.thingsboard.server.gen.transport.TransportProtos; |
|||
import org.thingsboard.server.service.state.DeviceStateService; |
|||
|
|||
import java.util.UUID; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.mockito.ArgumentMatchers.any; |
|||
import static org.mockito.BDDMockito.then; |
|||
import static org.mockito.Mockito.doCallRealMethod; |
|||
import static org.mockito.Mockito.doThrow; |
|||
import static org.mockito.Mockito.never; |
|||
|
|||
@ExtendWith(MockitoExtension.class) |
|||
public class DefaultTbCoreConsumerServiceTest { |
|||
|
|||
@Mock |
|||
private DeviceStateService stateServiceMock; |
|||
@Mock |
|||
private TbCoreConsumerStats statsMock; |
|||
|
|||
@Mock |
|||
private TbCallback tbCallbackMock; |
|||
|
|||
private final TenantId tenantId = TenantId.fromUUID(UUID.randomUUID()); |
|||
private final DeviceId deviceId = new DeviceId(UUID.randomUUID()); |
|||
private final long time = System.currentTimeMillis(); |
|||
|
|||
private ListeningExecutorService executor; |
|||
|
|||
@Mock |
|||
private DefaultTbCoreConsumerService defaultTbCoreConsumerServiceMock; |
|||
|
|||
@BeforeEach |
|||
public void setup() { |
|||
executor = MoreExecutors.newDirectExecutorService(); |
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stateService", stateServiceMock); |
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "deviceActivityEventsExecutor", executor); |
|||
} |
|||
|
|||
@AfterEach |
|||
public void cleanup() { |
|||
if (executor != null) { |
|||
executor.shutdown(); |
|||
try { |
|||
if (!executor.awaitTermination(10L, TimeUnit.SECONDS)) { |
|||
executor.shutdownNow(); |
|||
} |
|||
} catch (InterruptedException e) { |
|||
executor.shutdownNow(); |
|||
Thread.currentThread().interrupt(); |
|||
} |
|||
} |
|||
} |
|||
|
|||
@Test |
|||
public void givenProcessingSuccess_whenForwardingDeviceStateMsgToStateService_thenOnSuccessCallbackIsCalled() { |
|||
// GIVEN
|
|||
var stateMsg = TransportProtos.DeviceStateServiceMsgProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setAdded(true) |
|||
.setUpdated(false) |
|||
.setDeleted(false) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(stateMsg, tbCallbackMock); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(stateMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(stateServiceMock).should().onQueueMsg(stateMsg, tbCallbackMock); |
|||
} |
|||
|
|||
@Test |
|||
public void givenStatsEnabled_whenForwardingDeviceStateMsgToStateService_thenStatsAreRecorded() { |
|||
// GIVEN
|
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", true); |
|||
|
|||
var stateMsg = TransportProtos.DeviceStateServiceMsgProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setAdded(true) |
|||
.setUpdated(false) |
|||
.setDeleted(false) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(stateMsg, tbCallbackMock); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(stateMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(statsMock).should().log(stateMsg); |
|||
} |
|||
|
|||
@Test |
|||
public void givenStatsDisabled_whenForwardingDeviceStateMsgToStateService_thenStatsAreNotRecorded() { |
|||
// GIVEN
|
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", false); |
|||
|
|||
var stateMsg = TransportProtos.DeviceStateServiceMsgProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setAdded(true) |
|||
.setUpdated(false) |
|||
.setDeleted(false) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(stateMsg, tbCallbackMock); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(stateMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(statsMock).should(never()).log(stateMsg); |
|||
} |
|||
|
|||
@Test |
|||
public void givenProcessingSuccess_whenForwardingConnectMsgToStateService_thenOnSuccessCallbackIsCalled() { |
|||
// GIVEN
|
|||
var connectMsg = TransportProtos.DeviceConnectProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastConnectTime(time) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(connectMsg, tbCallbackMock); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(connectMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(stateServiceMock).should().onDeviceConnect(tenantId, deviceId, time); |
|||
then(tbCallbackMock).should().onSuccess(); |
|||
then(tbCallbackMock).should(never()).onFailure(any()); |
|||
} |
|||
|
|||
@Test |
|||
public void givenProcessingFailure_whenForwardingConnectMsgToStateService_thenOnFailureCallbackIsCalled() { |
|||
// GIVEN
|
|||
var connectMsg = TransportProtos.DeviceConnectProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastConnectTime(time) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(connectMsg, tbCallbackMock); |
|||
|
|||
var runtimeException = new RuntimeException("Something bad happened!"); |
|||
doThrow(runtimeException).when(stateServiceMock).onDeviceConnect(tenantId, deviceId, time); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(connectMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(tbCallbackMock).should(never()).onSuccess(); |
|||
then(tbCallbackMock).should().onFailure(runtimeException); |
|||
} |
|||
|
|||
@Test |
|||
public void givenStatsEnabled_whenForwardingConnectMsgToStateService_thenStatsAreRecorded() { |
|||
// GIVEN
|
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", true); |
|||
|
|||
var connectMsg = TransportProtos.DeviceConnectProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastConnectTime(time) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(connectMsg, tbCallbackMock); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(connectMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(statsMock).should().log(connectMsg); |
|||
} |
|||
|
|||
@Test |
|||
public void givenStatsDisabled_whenForwardingConnectMsgToStateService_thenStatsAreNotRecorded() { |
|||
// GIVEN
|
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", false); |
|||
|
|||
var connectMsg = TransportProtos.DeviceConnectProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastConnectTime(time) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(connectMsg, tbCallbackMock); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(connectMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(statsMock).should(never()).log(connectMsg); |
|||
} |
|||
|
|||
@Test |
|||
public void givenProcessingSuccess_whenForwardingActivityMsgToStateService_thenOnSuccessCallbackIsCalled() { |
|||
// GIVEN
|
|||
var activityMsg = TransportProtos.DeviceActivityProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastActivityTime(time) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(activityMsg, tbCallbackMock); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(activityMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(stateServiceMock).should().onDeviceActivity(tenantId, deviceId, time); |
|||
then(tbCallbackMock).should().onSuccess(); |
|||
then(tbCallbackMock).should(never()).onFailure(any()); |
|||
} |
|||
|
|||
@Test |
|||
public void givenProcessingFailure_whenForwardingActivityMsgToStateService_thenOnFailureCallbackIsCalled() { |
|||
// GIVEN
|
|||
var activityMsg = TransportProtos.DeviceActivityProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastActivityTime(time) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(activityMsg, tbCallbackMock); |
|||
|
|||
var runtimeException = new RuntimeException("Something bad happened!"); |
|||
doThrow(runtimeException).when(stateServiceMock).onDeviceActivity(tenantId, deviceId, time); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(activityMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(tbCallbackMock).should(never()).onSuccess(); |
|||
|
|||
var exceptionCaptor = ArgumentCaptor.forClass(Throwable.class); |
|||
then(tbCallbackMock).should().onFailure(exceptionCaptor.capture()); |
|||
assertThat(exceptionCaptor.getValue()) |
|||
.isInstanceOf(RuntimeException.class) |
|||
.hasMessage("Failed to update device activity for device [" + deviceId.getId() + "]!") |
|||
.hasCause(runtimeException); |
|||
} |
|||
|
|||
@Test |
|||
public void givenStatsEnabled_whenForwardingActivityMsgToStateService_thenStatsAreRecorded() { |
|||
// GIVEN
|
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", true); |
|||
|
|||
var activityMsg = TransportProtos.DeviceActivityProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastActivityTime(time) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(activityMsg, tbCallbackMock); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(activityMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(statsMock).should().log(activityMsg); |
|||
} |
|||
|
|||
@Test |
|||
public void givenStatsDisabled_whenForwardingActivityMsgToStateService_thenStatsAreNotRecorded() { |
|||
// GIVEN
|
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", false); |
|||
|
|||
var activityMsg = TransportProtos.DeviceActivityProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastActivityTime(time) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(activityMsg, tbCallbackMock); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(activityMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(statsMock).should(never()).log(activityMsg); |
|||
} |
|||
|
|||
@Test |
|||
public void givenProcessingSuccess_whenForwardingDisconnectMsgToStateService_thenOnSuccessCallbackIsCalled() { |
|||
// GIVEN
|
|||
var disconnectMsg = TransportProtos.DeviceDisconnectProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastDisconnectTime(time) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(disconnectMsg, tbCallbackMock); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(disconnectMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(stateServiceMock).should().onDeviceDisconnect(tenantId, deviceId, time); |
|||
then(tbCallbackMock).should().onSuccess(); |
|||
then(tbCallbackMock).should(never()).onFailure(any()); |
|||
} |
|||
|
|||
@Test |
|||
public void givenProcessingFailure_whenForwardingDisconnectMsgToStateService_thenOnFailureCallbackIsCalled() { |
|||
// GIVEN
|
|||
var disconnectMsg = TransportProtos.DeviceDisconnectProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastDisconnectTime(time) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(disconnectMsg, tbCallbackMock); |
|||
|
|||
var runtimeException = new RuntimeException("Something bad happened!"); |
|||
doThrow(runtimeException).when(stateServiceMock).onDeviceDisconnect(tenantId, deviceId, time); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(disconnectMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(tbCallbackMock).should(never()).onSuccess(); |
|||
then(tbCallbackMock).should().onFailure(runtimeException); |
|||
} |
|||
|
|||
@Test |
|||
public void givenStatsEnabled_whenForwardingDisconnectMsgToStateService_thenStatsAreRecorded() { |
|||
// GIVEN
|
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", true); |
|||
|
|||
var disconnectMsg = TransportProtos.DeviceDisconnectProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastDisconnectTime(time) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(disconnectMsg, tbCallbackMock); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(disconnectMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(statsMock).should().log(disconnectMsg); |
|||
} |
|||
|
|||
@Test |
|||
public void givenStatsDisabled_whenForwardingDisconnectMsgToStateService_thenStatsAreNotRecorded() { |
|||
// GIVEN
|
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", false); |
|||
|
|||
var disconnectMsg = TransportProtos.DeviceDisconnectProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastDisconnectTime(time) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(disconnectMsg, tbCallbackMock); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(disconnectMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(statsMock).should(never()).log(disconnectMsg); |
|||
} |
|||
|
|||
@Test |
|||
public void givenProcessingSuccess_whenForwardingInactivityMsgToStateService_thenOnSuccessCallbackIsCalled() { |
|||
// GIVEN
|
|||
var inactivityMsg = TransportProtos.DeviceInactivityProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastInactivityTime(time) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(inactivityMsg, tbCallbackMock); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(inactivityMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(stateServiceMock).should().onDeviceInactivity(tenantId, deviceId, time); |
|||
then(tbCallbackMock).should().onSuccess(); |
|||
then(tbCallbackMock).should(never()).onFailure(any()); |
|||
} |
|||
|
|||
@Test |
|||
public void givenProcessingFailure_whenForwardingInactivityMsgToStateService_thenOnFailureCallbackIsCalled() { |
|||
// GIVEN
|
|||
var inactivityMsg = TransportProtos.DeviceInactivityProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastInactivityTime(time) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(inactivityMsg, tbCallbackMock); |
|||
|
|||
var runtimeException = new RuntimeException("Something bad happened!"); |
|||
doThrow(runtimeException).when(stateServiceMock).onDeviceInactivity(tenantId, deviceId, time); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(inactivityMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(tbCallbackMock).should(never()).onSuccess(); |
|||
then(tbCallbackMock).should().onFailure(runtimeException); |
|||
} |
|||
|
|||
@Test |
|||
public void givenStatsEnabled_whenForwardingInactivityMsgToStateService_thenStatsAreRecorded() { |
|||
// GIVEN
|
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", true); |
|||
|
|||
var inactivityMsg = TransportProtos.DeviceInactivityProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastInactivityTime(time) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(inactivityMsg, tbCallbackMock); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(inactivityMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(statsMock).should().log(inactivityMsg); |
|||
} |
|||
|
|||
@Test |
|||
public void givenStatsDisabled_whenForwardingInactivityMsgToStateService_thenStatsAreNotRecorded() { |
|||
// GIVEN
|
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
|||
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", false); |
|||
|
|||
var inactivityMsg = TransportProtos.DeviceInactivityProto.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.setLastInactivityTime(time) |
|||
.build(); |
|||
|
|||
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(inactivityMsg, tbCallbackMock); |
|||
|
|||
// WHEN
|
|||
defaultTbCoreConsumerServiceMock.forwardToStateService(inactivityMsg, tbCallbackMock); |
|||
|
|||
// THEN
|
|||
then(statsMock).should(never()).log(inactivityMsg); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,266 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.service.state; |
|||
|
|||
import org.junit.jupiter.api.BeforeEach; |
|||
import org.junit.jupiter.api.DisplayName; |
|||
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.Captor; |
|||
import org.mockito.Mock; |
|||
import org.mockito.junit.jupiter.MockitoExtension; |
|||
import org.springframework.test.util.ReflectionTestUtils; |
|||
import org.thingsboard.server.cluster.TbClusterService; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
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.gen.transport.TransportProtos; |
|||
import org.thingsboard.server.queue.TbQueueCallback; |
|||
import org.thingsboard.server.queue.TbQueueMsgMetadata; |
|||
import org.thingsboard.server.queue.discovery.PartitionService; |
|||
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; |
|||
|
|||
import java.util.Optional; |
|||
import java.util.UUID; |
|||
import java.util.stream.Stream; |
|||
|
|||
import static org.mockito.ArgumentMatchers.any; |
|||
import static org.mockito.ArgumentMatchers.eq; |
|||
import static org.mockito.BDDMockito.given; |
|||
import static org.mockito.BDDMockito.then; |
|||
import static org.mockito.Mockito.doThrow; |
|||
import static org.mockito.Mockito.never; |
|||
|
|||
@ExtendWith(MockitoExtension.class) |
|||
public class DefaultRuleEngineDeviceStateManagerTest { |
|||
|
|||
@Mock |
|||
private static DeviceStateService deviceStateServiceMock; |
|||
@Mock |
|||
private static TbCallback tbCallbackMock; |
|||
@Mock |
|||
private static TbClusterService clusterServiceMock; |
|||
@Mock |
|||
private static TbQueueMsgMetadata metadataMock; |
|||
|
|||
@Mock |
|||
private TbServiceInfoProvider serviceInfoProviderMock; |
|||
@Mock |
|||
private PartitionService partitionServiceMock; |
|||
|
|||
@Captor |
|||
private static ArgumentCaptor<TbQueueCallback> queueCallbackCaptor; |
|||
|
|||
private static DefaultRuleEngineDeviceStateManager deviceStateManager; |
|||
|
|||
private static final TenantId TENANT_ID = TenantId.fromUUID(UUID.fromString("57ab2e6c-bc4c-11ee-a506-0242ac120002")); |
|||
private static final DeviceId DEVICE_ID = DeviceId.fromString("74a9053e-bc4c-11ee-a506-0242ac120002"); |
|||
private static final long EVENT_TS = System.currentTimeMillis(); |
|||
private static final RuntimeException RUNTIME_EXCEPTION = new RuntimeException("Something bad happened!"); |
|||
private static final TopicPartitionInfo MY_TPI = TopicPartitionInfo.builder().myPartition(true).build(); |
|||
private static final TopicPartitionInfo EXTERNAL_TPI = TopicPartitionInfo.builder().myPartition(false).build(); |
|||
|
|||
@BeforeEach |
|||
public void setup() { |
|||
deviceStateManager = new DefaultRuleEngineDeviceStateManager(serviceInfoProviderMock, partitionServiceMock, Optional.of(deviceStateServiceMock), clusterServiceMock); |
|||
} |
|||
|
|||
@ParameterizedTest |
|||
@DisplayName("Given event should be routed to local service and event processed has succeeded, " + |
|||
"when onDeviceX() is called, then should route event to local service and call onSuccess() callback.") |
|||
@MethodSource |
|||
public void givenRoutedToLocalAndProcessingSuccess_whenOnDeviceAction_thenShouldCallLocalServiceAndSuccessCallback(Runnable onDeviceAction, Runnable actionVerification) { |
|||
// GIVEN
|
|||
given(serviceInfoProviderMock.isService(ServiceType.TB_CORE)).willReturn(true); |
|||
given(partitionServiceMock.resolve(ServiceType.TB_CORE, TENANT_ID, DEVICE_ID)).willReturn(MY_TPI); |
|||
|
|||
onDeviceAction.run(); |
|||
|
|||
// THEN
|
|||
actionVerification.run(); |
|||
|
|||
then(clusterServiceMock).shouldHaveNoInteractions(); |
|||
then(tbCallbackMock).should().onSuccess(); |
|||
then(tbCallbackMock).should(never()).onFailure(any()); |
|||
} |
|||
|
|||
private static Stream<Arguments> givenRoutedToLocalAndProcessingSuccess_whenOnDeviceAction_thenShouldCallLocalServiceAndSuccessCallback() { |
|||
return Stream.of( |
|||
Arguments.of( |
|||
(Runnable) () -> deviceStateManager.onDeviceConnect(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
|||
(Runnable) () -> then(deviceStateServiceMock).should().onDeviceConnect(TENANT_ID, DEVICE_ID, EVENT_TS) |
|||
), |
|||
Arguments.of( |
|||
(Runnable) () -> deviceStateManager.onDeviceActivity(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
|||
(Runnable) () -> then(deviceStateServiceMock).should().onDeviceActivity(TENANT_ID, DEVICE_ID, EVENT_TS) |
|||
), |
|||
Arguments.of( |
|||
(Runnable) () -> deviceStateManager.onDeviceDisconnect(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
|||
(Runnable) () -> then(deviceStateServiceMock).should().onDeviceDisconnect(TENANT_ID, DEVICE_ID, EVENT_TS) |
|||
), |
|||
Arguments.of( |
|||
(Runnable) () -> deviceStateManager.onDeviceInactivity(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
|||
(Runnable) () -> then(deviceStateServiceMock).should().onDeviceInactivity(TENANT_ID, DEVICE_ID, EVENT_TS) |
|||
) |
|||
); |
|||
} |
|||
|
|||
@ParameterizedTest |
|||
@DisplayName("Given event should be routed to local service and event processed has failed, " + |
|||
"when onDeviceX() is called, then should route event to local service and call onFailure() callback.") |
|||
@MethodSource |
|||
public void givenRoutedToLocalAndProcessingFailure_whenOnDeviceAction_thenShouldCallLocalServiceAndFailureCallback( |
|||
Runnable exceptionThrowSetup, Runnable onDeviceAction, Runnable actionVerification |
|||
) { |
|||
// GIVEN
|
|||
given(serviceInfoProviderMock.isService(ServiceType.TB_CORE)).willReturn(true); |
|||
given(partitionServiceMock.resolve(ServiceType.TB_CORE, TENANT_ID, DEVICE_ID)).willReturn(MY_TPI); |
|||
|
|||
exceptionThrowSetup.run(); |
|||
|
|||
// WHEN
|
|||
onDeviceAction.run(); |
|||
|
|||
// THEN
|
|||
actionVerification.run(); |
|||
|
|||
then(clusterServiceMock).shouldHaveNoInteractions(); |
|||
then(tbCallbackMock).should(never()).onSuccess(); |
|||
then(tbCallbackMock).should().onFailure(RUNTIME_EXCEPTION); |
|||
} |
|||
|
|||
private static Stream<Arguments> givenRoutedToLocalAndProcessingFailure_whenOnDeviceAction_thenShouldCallLocalServiceAndFailureCallback() { |
|||
return Stream.of( |
|||
Arguments.of( |
|||
(Runnable) () -> doThrow(RUNTIME_EXCEPTION).when(deviceStateServiceMock).onDeviceConnect(TENANT_ID, DEVICE_ID, EVENT_TS), |
|||
(Runnable) () -> deviceStateManager.onDeviceConnect(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
|||
(Runnable) () -> then(deviceStateServiceMock).should().onDeviceConnect(TENANT_ID, DEVICE_ID, EVENT_TS) |
|||
), |
|||
Arguments.of( |
|||
(Runnable) () -> doThrow(RUNTIME_EXCEPTION).when(deviceStateServiceMock).onDeviceActivity(TENANT_ID, DEVICE_ID, EVENT_TS), |
|||
(Runnable) () -> deviceStateManager.onDeviceActivity(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
|||
(Runnable) () -> then(deviceStateServiceMock).should().onDeviceActivity(TENANT_ID, DEVICE_ID, EVENT_TS) |
|||
), |
|||
Arguments.of( |
|||
(Runnable) () -> doThrow(RUNTIME_EXCEPTION).when(deviceStateServiceMock).onDeviceDisconnect(TENANT_ID, DEVICE_ID, EVENT_TS), |
|||
(Runnable) () -> deviceStateManager.onDeviceDisconnect(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
|||
(Runnable) () -> then(deviceStateServiceMock).should().onDeviceDisconnect(TENANT_ID, DEVICE_ID, EVENT_TS) |
|||
), |
|||
Arguments.of( |
|||
(Runnable) () -> doThrow(RUNTIME_EXCEPTION).when(deviceStateServiceMock).onDeviceInactivity(TENANT_ID, DEVICE_ID, EVENT_TS), |
|||
(Runnable) () -> deviceStateManager.onDeviceInactivity(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
|||
(Runnable) () -> then(deviceStateServiceMock).should().onDeviceInactivity(TENANT_ID, DEVICE_ID, EVENT_TS) |
|||
) |
|||
); |
|||
} |
|||
|
|||
@ParameterizedTest |
|||
@DisplayName("Given event should be routed to external service, " + |
|||
"when onDeviceX() is called, then should send correct queue message to external service with correct callback object.") |
|||
@MethodSource |
|||
public void givenRoutedToExternal_whenOnDeviceAction_thenShouldSendQueueMsgToExternalServiceWithCorrectCallback(Runnable onDeviceAction, Runnable actionVerification) { |
|||
// WHEN
|
|||
ReflectionTestUtils.setField(deviceStateManager, "deviceStateService", Optional.empty()); |
|||
given(serviceInfoProviderMock.isService(ServiceType.TB_CORE)).willReturn(false); |
|||
given(partitionServiceMock.resolve(ServiceType.TB_CORE, TENANT_ID, DEVICE_ID)).willReturn(EXTERNAL_TPI); |
|||
|
|||
onDeviceAction.run(); |
|||
|
|||
// THEN
|
|||
actionVerification.run(); |
|||
|
|||
TbQueueCallback callback = queueCallbackCaptor.getValue(); |
|||
callback.onSuccess(metadataMock); |
|||
then(tbCallbackMock).should().onSuccess(); |
|||
callback.onFailure(RUNTIME_EXCEPTION); |
|||
then(tbCallbackMock).should().onFailure(RUNTIME_EXCEPTION); |
|||
} |
|||
|
|||
private static Stream<Arguments> givenRoutedToExternal_whenOnDeviceAction_thenShouldSendQueueMsgToExternalServiceWithCorrectCallback() { |
|||
return Stream.of( |
|||
Arguments.of( |
|||
(Runnable) () -> deviceStateManager.onDeviceConnect(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
|||
(Runnable) () -> { |
|||
var deviceConnectMsg = TransportProtos.DeviceConnectProto.newBuilder() |
|||
.setTenantIdMSB(TENANT_ID.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(TENANT_ID.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(DEVICE_ID.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(DEVICE_ID.getId().getLeastSignificantBits()) |
|||
.setLastConnectTime(EVENT_TS) |
|||
.build(); |
|||
var toCoreMsg = TransportProtos.ToCoreMsg.newBuilder() |
|||
.setDeviceConnectMsg(deviceConnectMsg) |
|||
.build(); |
|||
then(clusterServiceMock).should().pushMsgToCore(eq(EXTERNAL_TPI), any(UUID.class), eq(toCoreMsg), queueCallbackCaptor.capture()); |
|||
} |
|||
), |
|||
Arguments.of( |
|||
(Runnable) () -> deviceStateManager.onDeviceActivity(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
|||
(Runnable) () -> { |
|||
var deviceActivityMsg = TransportProtos.DeviceActivityProto.newBuilder() |
|||
.setTenantIdMSB(TENANT_ID.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(TENANT_ID.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(DEVICE_ID.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(DEVICE_ID.getId().getLeastSignificantBits()) |
|||
.setLastActivityTime(EVENT_TS) |
|||
.build(); |
|||
var toCoreMsg = TransportProtos.ToCoreMsg.newBuilder() |
|||
.setDeviceActivityMsg(deviceActivityMsg) |
|||
.build(); |
|||
then(clusterServiceMock).should().pushMsgToCore(eq(EXTERNAL_TPI), any(UUID.class), eq(toCoreMsg), queueCallbackCaptor.capture()); |
|||
} |
|||
), |
|||
Arguments.of( |
|||
(Runnable) () -> deviceStateManager.onDeviceDisconnect(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
|||
(Runnable) () -> { |
|||
var deviceDisconnectMsg = TransportProtos.DeviceDisconnectProto.newBuilder() |
|||
.setTenantIdMSB(TENANT_ID.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(TENANT_ID.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(DEVICE_ID.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(DEVICE_ID.getId().getLeastSignificantBits()) |
|||
.setLastDisconnectTime(EVENT_TS) |
|||
.build(); |
|||
var toCoreMsg = TransportProtos.ToCoreMsg.newBuilder() |
|||
.setDeviceDisconnectMsg(deviceDisconnectMsg) |
|||
.build(); |
|||
then(clusterServiceMock).should().pushMsgToCore(eq(EXTERNAL_TPI), any(UUID.class), eq(toCoreMsg), queueCallbackCaptor.capture()); |
|||
} |
|||
), |
|||
Arguments.of( |
|||
(Runnable) () -> deviceStateManager.onDeviceInactivity(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
|||
(Runnable) () -> { |
|||
var deviceInactivityMsg = TransportProtos.DeviceInactivityProto.newBuilder() |
|||
.setTenantIdMSB(TENANT_ID.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(TENANT_ID.getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(DEVICE_ID.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(DEVICE_ID.getId().getLeastSignificantBits()) |
|||
.setLastInactivityTime(EVENT_TS) |
|||
.build(); |
|||
var toCoreMsg = TransportProtos.ToCoreMsg.newBuilder() |
|||
.setDeviceInactivityMsg(deviceInactivityMsg) |
|||
.build(); |
|||
then(clusterServiceMock).should().pushMsgToCore(eq(EXTERNAL_TPI), any(UUID.class), eq(toCoreMsg), queueCallbackCaptor.capture()); |
|||
} |
|||
) |
|||
); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,25 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.cache.limits; |
|||
|
|||
import org.thingsboard.server.common.data.TenantProfile; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
public interface TenantProfileProvider { |
|||
|
|||
TenantProfile get(TenantId tenantId); |
|||
|
|||
} |
|||
@ -0,0 +1,66 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.common.data.notification.info; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
import org.thingsboard.server.common.data.id.CustomerId; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
import java.util.Map; |
|||
|
|||
import static org.thingsboard.server.common.data.util.CollectionsUtil.mapOf; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
@Builder |
|||
public class EdgeCommunicationFailureNotificationInfo implements RuleOriginatedNotificationInfo { |
|||
|
|||
private TenantId tenantId; |
|||
private CustomerId customerId; |
|||
private EdgeId edgeId; |
|||
private String edgeName; |
|||
private String failureMsg; |
|||
|
|||
@Override |
|||
public Map<String, String> getTemplateData() { |
|||
return mapOf( |
|||
"edgeId", edgeId.toString(), |
|||
"edgeName", edgeName, |
|||
"failureMsg", failureMsg |
|||
); |
|||
} |
|||
|
|||
@Override |
|||
public TenantId getAffectedTenantId() { |
|||
return tenantId; |
|||
} |
|||
|
|||
@Override |
|||
public CustomerId getAffectedCustomerId() { |
|||
return customerId; |
|||
} |
|||
|
|||
@Override |
|||
public EntityId getStateEntityId() { |
|||
return edgeId; |
|||
} |
|||
} |
|||
@ -0,0 +1,66 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.common.data.notification.info; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
import org.thingsboard.server.common.data.id.CustomerId; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
import java.util.Map; |
|||
|
|||
import static org.thingsboard.server.common.data.util.CollectionsUtil.mapOf; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
@Builder |
|||
public class EdgeConnectionNotificationInfo implements RuleOriginatedNotificationInfo { |
|||
|
|||
private String eventType; |
|||
private TenantId tenantId; |
|||
private CustomerId customerId; |
|||
private EdgeId edgeId; |
|||
private String edgeName; |
|||
|
|||
@Override |
|||
public Map<String, String> getTemplateData() { |
|||
return mapOf( |
|||
"eventType", eventType, |
|||
"edgeId", edgeId.toString(), |
|||
"edgeName", edgeName |
|||
); |
|||
} |
|||
|
|||
@Override |
|||
public TenantId getAffectedTenantId() { |
|||
return tenantId; |
|||
} |
|||
|
|||
@Override |
|||
public CustomerId getAffectedCustomerId() { |
|||
return customerId; |
|||
} |
|||
|
|||
@Override |
|||
public EntityId getStateEntityId() { |
|||
return edgeId; |
|||
} |
|||
} |
|||
@ -0,0 +1,63 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.common.data.notification.rule.trigger; |
|||
|
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.id.CustomerId; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.config.NotificationRuleTriggerType; |
|||
|
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
@Data |
|||
@Builder |
|||
public class EdgeCommunicationFailureTrigger implements NotificationRuleTrigger { |
|||
|
|||
private final TenantId tenantId; |
|||
private final CustomerId customerId; |
|||
private final EdgeId edgeId; |
|||
private final String edgeName; |
|||
private final String failureMsg; |
|||
private final String error; |
|||
|
|||
@Override |
|||
public boolean deduplicate() { |
|||
return true; |
|||
} |
|||
|
|||
@Override |
|||
public String getDeduplicationKey() { |
|||
return String.join(":", NotificationRuleTrigger.super.getDeduplicationKey(), error); |
|||
} |
|||
|
|||
@Override |
|||
public long getDefaultDeduplicationDuration() { |
|||
return TimeUnit.MINUTES.toMillis(30); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationRuleTriggerType getType() { |
|||
return NotificationRuleTriggerType.EDGE_COMMUNICATION_FAILURE; |
|||
} |
|||
|
|||
@Override |
|||
public EntityId getOriginatorEntityId() { |
|||
return edgeId; |
|||
} |
|||
} |
|||
@ -0,0 +1,62 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.common.data.notification.rule.trigger; |
|||
|
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.id.CustomerId; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.config.NotificationRuleTriggerType; |
|||
|
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
@Data |
|||
@Builder |
|||
public class EdgeConnectionTrigger implements NotificationRuleTrigger { |
|||
|
|||
private final TenantId tenantId; |
|||
private final CustomerId customerId; |
|||
private final EdgeId edgeId; |
|||
private final boolean connected; |
|||
private final String edgeName; |
|||
|
|||
@Override |
|||
public boolean deduplicate() { |
|||
return true; |
|||
} |
|||
|
|||
@Override |
|||
public String getDeduplicationKey() { |
|||
return String.join(":", NotificationRuleTrigger.super.getDeduplicationKey(), String.valueOf(connected)); |
|||
} |
|||
|
|||
@Override |
|||
public long getDefaultDeduplicationDuration() { |
|||
return TimeUnit.MINUTES.toMillis(1); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationRuleTriggerType getType() { |
|||
return NotificationRuleTriggerType.EDGE_CONNECTION; |
|||
} |
|||
|
|||
@Override |
|||
public EntityId getOriginatorEntityId() { |
|||
return edgeId; |
|||
} |
|||
} |
|||
@ -0,0 +1,39 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.common.data.notification.rule.trigger.config; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
import java.util.Set; |
|||
import java.util.UUID; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
@Builder |
|||
public class EdgeCommunicationFailureNotificationRuleTriggerConfig implements NotificationRuleTriggerConfig { |
|||
|
|||
private Set<UUID> edges; // if empty - all edges
|
|||
|
|||
@Override |
|||
public NotificationRuleTriggerType getTriggerType() { |
|||
return NotificationRuleTriggerType.EDGE_COMMUNICATION_FAILURE; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,44 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.common.data.notification.rule.trigger.config; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
import java.util.Set; |
|||
import java.util.UUID; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
@Builder |
|||
public class EdgeConnectionNotificationRuleTriggerConfig implements NotificationRuleTriggerConfig { |
|||
|
|||
private Set<UUID> edges; // if empty - all edges
|
|||
private Set<EdgeConnectivityEvent> notifyOn; |
|||
|
|||
@Override |
|||
public NotificationRuleTriggerType getTriggerType() { |
|||
return NotificationRuleTriggerType.EDGE_CONNECTION; |
|||
} |
|||
|
|||
public enum EdgeConnectivityEvent { |
|||
CONNECTED, DISCONNECTED |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,47 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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 Consumer<TbQueueMsgMetadata> onSuccess; |
|||
private final Consumer<Throwable> onFailure; |
|||
|
|||
public SimpleTbQueueCallback(Consumer<TbQueueMsgMetadata> onSuccess, Consumer<Throwable> onFailure) { |
|||
this.onSuccess = onSuccess; |
|||
this.onFailure = onFailure; |
|||
} |
|||
|
|||
@Override |
|||
public void onSuccess(TbQueueMsgMetadata metadata) { |
|||
if (onSuccess != null) { |
|||
onSuccess.accept(metadata); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
if (onFailure != null) { |
|||
onFailure.accept(t); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,63 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.transport.lwm2m.server.store.util; |
|||
|
|||
import com.eclipsesource.json.Json; |
|||
import com.eclipsesource.json.JsonObject; |
|||
import org.apache.commons.lang3.NotImplementedException; |
|||
import org.eclipse.leshan.core.request.Identity; |
|||
import org.eclipse.leshan.core.util.Hex; |
|||
|
|||
import java.security.PublicKey; |
|||
|
|||
public class LwM2MIdentitySerDes { |
|||
|
|||
private static final String KEY_ADDRESS = "address"; |
|||
private static final String KEY_PORT = "port"; |
|||
private static final String KEY_ID = "id"; |
|||
private static final String KEY_CN = "cn"; |
|||
private static final String KEY_RPK = "rpk"; |
|||
protected static final String KEY_LWM2MIDENTITY_TYPE = "type"; |
|||
protected static final String LWM2MIDENTITY_TYPE_UNSECURE = "unsecure"; |
|||
protected static final String LWM2MIDENTITY_TYPE_PSK = "psk"; |
|||
protected static final String LWM2MIDENTITY_TYPE_X509 = "x509"; |
|||
protected static final String LWM2MIDENTITY_TYPE_RPK = "rpk"; |
|||
|
|||
public static JsonObject serialize(Identity identity) { |
|||
JsonObject o = Json.object(); |
|||
|
|||
if (identity.isPSK()) { |
|||
o.set(KEY_LWM2MIDENTITY_TYPE, LWM2MIDENTITY_TYPE_PSK); |
|||
o.set(KEY_ID, identity.getPskIdentity()); |
|||
} else if (identity.isRPK()) { |
|||
o.set(KEY_LWM2MIDENTITY_TYPE, LWM2MIDENTITY_TYPE_RPK); |
|||
PublicKey publicKey = identity.getRawPublicKey(); |
|||
o.set(KEY_RPK, Hex.encodeHexString(publicKey.getEncoded())); |
|||
} else if (identity.isX509()) { |
|||
o.set(KEY_LWM2MIDENTITY_TYPE, LWM2MIDENTITY_TYPE_X509); |
|||
o.set(KEY_CN, identity.getX509CommonName()); |
|||
} else { |
|||
o.set(KEY_LWM2MIDENTITY_TYPE, LWM2MIDENTITY_TYPE_UNSECURE); |
|||
o.set(KEY_ADDRESS, identity.getPeerAddress().getHostString()); |
|||
o.set(KEY_PORT, identity.getPeerAddress().getPort()); |
|||
} |
|||
return o; |
|||
} |
|||
|
|||
public static Identity deserialize(JsonObject peer) { |
|||
throw new NotImplementedException(); |
|||
} |
|||
} |
|||
@ -0,0 +1,105 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.transport.lwm2m.bootstrap; |
|||
|
|||
import org.eclipse.californium.core.network.CoapEndpoint; |
|||
import org.eclipse.californium.scandium.config.DtlsConnectorConfig; |
|||
import org.eclipse.leshan.server.californium.LeshanServer; |
|||
import org.eclipse.leshan.server.californium.bootstrap.LeshanBootstrapServer; |
|||
import org.eclipse.leshan.server.californium.registration.CaliforniumRegistrationStore; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.api.extension.ExtendWith; |
|||
import org.mockito.Mock; |
|||
import org.mockito.junit.jupiter.MockitoExtension; |
|||
import org.springframework.test.util.ReflectionTestUtils; |
|||
import org.thingsboard.server.cache.ota.OtaPackageDataCache; |
|||
import org.thingsboard.server.common.transport.TransportService; |
|||
import org.thingsboard.server.transport.lwm2m.bootstrap.secure.TbLwM2MDtlsBootstrapCertificateVerifier; |
|||
import org.thingsboard.server.transport.lwm2m.bootstrap.store.LwM2MBootstrapSecurityStore; |
|||
import org.thingsboard.server.transport.lwm2m.bootstrap.store.LwM2MInMemoryBootstrapConfigStore; |
|||
import org.thingsboard.server.transport.lwm2m.config.LwM2MTransportBootstrapConfig; |
|||
import org.thingsboard.server.transport.lwm2m.config.LwM2MTransportServerConfig; |
|||
import org.thingsboard.server.transport.lwm2m.secure.TbLwM2MAuthorizer; |
|||
import org.thingsboard.server.transport.lwm2m.secure.TbLwM2MDtlsCertificateVerifier; |
|||
import org.thingsboard.server.transport.lwm2m.server.store.TbSecurityStore; |
|||
import org.thingsboard.server.transport.lwm2m.server.uplink.LwM2mUplinkMsgHandler; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.mockito.BDDMockito.when; |
|||
|
|||
@ExtendWith(MockitoExtension.class) |
|||
public class LwM2MTransportBootstrapServiceTest { |
|||
|
|||
@Mock |
|||
private LwM2MTransportServerConfig serverConfig; |
|||
@Mock |
|||
private LwM2MTransportBootstrapConfig bootstrapConfig; |
|||
@Mock |
|||
private LwM2MBootstrapSecurityStore lwM2MBootstrapSecurityStore; |
|||
@Mock |
|||
private LwM2MInMemoryBootstrapConfigStore lwM2MInMemoryBootstrapConfigStore; |
|||
@Mock |
|||
private TransportService transportService; |
|||
@Mock |
|||
private TbLwM2MDtlsBootstrapCertificateVerifier certificateVerifier; |
|||
|
|||
|
|||
@Test |
|||
public void getLHServer_creates_ConnectionIdGenerator_when_connection_id_length_not_null(){ |
|||
final Integer CONNECTION_ID_LENGTH = 6; |
|||
when(serverConfig.getDtlsConnectionIdLength()).thenReturn(CONNECTION_ID_LENGTH); |
|||
var lwM2MBootstrapService = createLwM2MBootstrapService(); |
|||
|
|||
var server = lwM2MBootstrapService.getLhBootstrapServer(); |
|||
var securedEndpoint = (CoapEndpoint) ReflectionTestUtils.getField(server, "securedEndpoint"); |
|||
assertThat(securedEndpoint).isNotNull(); |
|||
|
|||
var config = (DtlsConnectorConfig) ReflectionTestUtils.getField(securedEndpoint.getConnector(), "config"); |
|||
assertThat(config).isNotNull(); |
|||
assertThat(config.getConnectionIdGenerator()).isNotNull(); |
|||
assertThat((Integer) ReflectionTestUtils.getField(config.getConnectionIdGenerator(), "connectionIdLength")) |
|||
.isEqualTo(CONNECTION_ID_LENGTH); |
|||
} |
|||
|
|||
@Test |
|||
public void getLHServer_creates_no_ConnectionIdGenerator_when_connection_id_length_is_null(){ |
|||
when(serverConfig.getDtlsConnectionIdLength()).thenReturn(null); |
|||
var lwM2MBootstrapService = createLwM2MBootstrapService(); |
|||
|
|||
var server = lwM2MBootstrapService.getLhBootstrapServer(); |
|||
var securedEndpoint = (CoapEndpoint) ReflectionTestUtils.getField(server, "securedEndpoint"); |
|||
assertThat(securedEndpoint).isNotNull(); |
|||
|
|||
var config = (DtlsConnectorConfig) ReflectionTestUtils.getField(securedEndpoint.getConnector(), "config"); |
|||
assertThat(config).isNotNull(); |
|||
assertThat(config.getConnectionIdGenerator()).isNull(); |
|||
} |
|||
|
|||
private LwM2MTransportBootstrapService createLwM2MBootstrapService() { |
|||
setDefaultConfigVariables(); |
|||
return new LwM2MTransportBootstrapService(serverConfig, bootstrapConfig, lwM2MBootstrapSecurityStore, |
|||
lwM2MInMemoryBootstrapConfigStore, transportService, certificateVerifier); |
|||
} |
|||
|
|||
private void setDefaultConfigVariables(){ |
|||
when(bootstrapConfig.getPort()).thenReturn(5683); |
|||
when(bootstrapConfig.getSecurePort()).thenReturn(5684); |
|||
when(serverConfig.isRecommendedCiphers()).thenReturn(false); |
|||
when(serverConfig.getDtlsRetransmissionTimeout()).thenReturn(9000); |
|||
} |
|||
|
|||
|
|||
} |
|||
@ -0,0 +1,61 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.transport.lwm2m.config; |
|||
|
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.api.extension.ExtendWith; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.boot.context.properties.EnableConfigurationProperties; |
|||
import org.springframework.boot.test.context.SpringBootContextLoader; |
|||
import org.springframework.boot.test.mock.mockito.MockBean; |
|||
import org.springframework.test.context.ContextConfiguration; |
|||
import org.springframework.test.context.TestPropertySource; |
|||
import org.springframework.test.context.junit.jupiter.SpringExtension; |
|||
import org.thingsboard.server.common.transport.config.ssl.SslCredentialsConfig; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
|
|||
@ExtendWith(SpringExtension.class) |
|||
@EnableConfigurationProperties(value = LwM2MTransportServerConfig.class) |
|||
@ContextConfiguration(classes = {LwM2MTransportServerConfig.class}, loader = SpringBootContextLoader.class) |
|||
@TestPropertySource(properties = { |
|||
"transport.sessions.report_timeout=10", |
|||
"transport.lwm2m.security.recommended_ciphers=true", |
|||
"transport.lwm2m.security.recommended_supported_groups=true", |
|||
"transport.lwm2m.downlink_pool_size=10", |
|||
"transport.lwm2m.uplink_pool_size=10", |
|||
"transport.lwm2m.ota_pool_size=10", |
|||
"transport.lwm2m.clean_period_in_sec=2", |
|||
"transport.lwm2m.dtls.connection_id_length=" |
|||
|
|||
}) |
|||
class LwM2MTransportServerConfigTest { |
|||
|
|||
@MockBean(name = "lwm2mServerCredentials") |
|||
private SslCredentialsConfig credentialsConfig; |
|||
|
|||
@MockBean(name = "lwm2mTrustCredentials") |
|||
private SslCredentialsConfig trustCredentialsConfig; |
|||
|
|||
@Autowired |
|||
private LwM2MTransportServerConfig serverConfig; |
|||
|
|||
@Test |
|||
void getDtlsConnectionIdLength_return_null_is_property_is_empty() { |
|||
// note: transport.lwm2m.dtls.connect_id_length is set in TestPropertySource
|
|||
assertThat(serverConfig.getDtlsConnectionIdLength()).isNull(); |
|||
} |
|||
} |
|||
@ -0,0 +1,109 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.transport.lwm2m.server; |
|||
|
|||
import org.eclipse.californium.core.network.CoapEndpoint; |
|||
import org.eclipse.californium.scandium.config.DtlsConnectorConfig; |
|||
import org.eclipse.leshan.server.californium.LeshanServer; |
|||
import org.eclipse.leshan.server.californium.registration.CaliforniumRegistrationStore; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.api.extension.ExtendWith; |
|||
import org.mockito.Mock; |
|||
import org.mockito.junit.jupiter.MockitoExtension; |
|||
import org.springframework.test.util.ReflectionTestUtils; |
|||
import org.thingsboard.server.cache.ota.OtaPackageDataCache; |
|||
import org.thingsboard.server.transport.lwm2m.config.LwM2MTransportServerConfig; |
|||
import org.thingsboard.server.transport.lwm2m.secure.TbLwM2MAuthorizer; |
|||
import org.thingsboard.server.transport.lwm2m.secure.TbLwM2MDtlsCertificateVerifier; |
|||
import org.thingsboard.server.transport.lwm2m.server.store.TbSecurityStore; |
|||
import org.thingsboard.server.transport.lwm2m.server.uplink.LwM2mUplinkMsgHandler; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.mockito.BDDMockito.when; |
|||
|
|||
@ExtendWith(MockitoExtension.class) |
|||
public class DefaultLwM2mTransportServiceTest { |
|||
|
|||
@Mock |
|||
private LwM2mTransportContext context; |
|||
|
|||
@Mock |
|||
private LwM2MTransportServerConfig config; |
|||
@Mock |
|||
private OtaPackageDataCache otaPackageDataCache; |
|||
@Mock |
|||
private LwM2mUplinkMsgHandler handler; |
|||
@Mock |
|||
private CaliforniumRegistrationStore registrationStore; |
|||
@Mock |
|||
private TbSecurityStore securityStore; |
|||
@Mock |
|||
private TbLwM2MDtlsCertificateVerifier certificateVerifier; |
|||
@Mock |
|||
private TbLwM2MAuthorizer authorizer; |
|||
@Mock |
|||
private LwM2mVersionedModelProvider modelProvider; |
|||
|
|||
|
|||
@Test |
|||
public void getLHServer_creates_ConnectionIdGenerator_when_connection_id_length_not_null(){ |
|||
final Integer CONNECTION_ID_LENGTH = 6; |
|||
when(config.getDtlsConnectionIdLength()).thenReturn(CONNECTION_ID_LENGTH); |
|||
var lwm2mService = createLwM2MService(); |
|||
|
|||
LeshanServer server = ReflectionTestUtils.invokeMethod(lwm2mService, "getLhServer"); |
|||
|
|||
assertThat(server).isNotNull(); |
|||
var securedEndpoint = (CoapEndpoint) ReflectionTestUtils.getField(server, "securedEndpoint"); |
|||
assertThat(securedEndpoint).isNotNull(); |
|||
|
|||
var config = (DtlsConnectorConfig) ReflectionTestUtils.getField(securedEndpoint.getConnector(), "config"); |
|||
assertThat(config).isNotNull(); |
|||
assertThat(config.getConnectionIdGenerator()).isNotNull(); |
|||
assertThat((Integer) ReflectionTestUtils.getField(config.getConnectionIdGenerator(), "connectionIdLength")) |
|||
.isEqualTo(CONNECTION_ID_LENGTH); |
|||
} |
|||
|
|||
@Test |
|||
public void getLHServer_creates_no_ConnectionIdGenerator_when_connection_id_length_is_null(){ |
|||
when(config.getDtlsConnectionIdLength()).thenReturn(null); |
|||
var lwm2mService = createLwM2MService(); |
|||
|
|||
LeshanServer server = ReflectionTestUtils.invokeMethod(lwm2mService, "getLhServer"); |
|||
|
|||
assertThat(server).isNotNull(); |
|||
var securedEndpoint = (CoapEndpoint) ReflectionTestUtils.getField(server, "securedEndpoint"); |
|||
assertThat(securedEndpoint).isNotNull(); |
|||
var config = (DtlsConnectorConfig) ReflectionTestUtils.getField(securedEndpoint.getConnector(), "config"); |
|||
assertThat(config).isNotNull(); |
|||
assertThat(config.getConnectionIdGenerator()).isNull(); |
|||
} |
|||
|
|||
private DefaultLwM2mTransportService createLwM2MService() { |
|||
setDefaultConfigVariables(); |
|||
return new DefaultLwM2mTransportService(context, config, otaPackageDataCache, handler, registrationStore, |
|||
securityStore, certificateVerifier, authorizer, modelProvider); |
|||
} |
|||
|
|||
private void setDefaultConfigVariables(){ |
|||
when(config.getPort()).thenReturn(5683); |
|||
when(config.getSecurePort()).thenReturn(5684); |
|||
when(config.isRecommendedCiphers()).thenReturn(false); |
|||
when(config.getDtlsRetransmissionTimeout()).thenReturn(9000); |
|||
} |
|||
|
|||
|
|||
} |
|||
@ -0,0 +1,265 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.transport.lwm2m.server.store; |
|||
|
|||
import org.eclipse.leshan.core.link.Link; |
|||
import org.eclipse.leshan.core.request.Identity; |
|||
import org.eclipse.leshan.core.util.NamedThreadFactory; |
|||
import org.eclipse.leshan.server.redis.serialization.RegistrationSerDes; |
|||
import org.eclipse.leshan.server.registration.Registration; |
|||
import org.eclipse.leshan.server.registration.RegistrationUpdate; |
|||
import org.junit.jupiter.api.BeforeEach; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.api.extension.ExtendWith; |
|||
import org.mockito.junit.jupiter.MockitoExtension; |
|||
import org.springframework.data.redis.connection.RedisConnection; |
|||
import org.springframework.data.redis.connection.RedisConnectionFactory; |
|||
import org.springframework.integration.redis.util.RedisLockRegistry; |
|||
import org.springframework.test.util.ReflectionTestUtils; |
|||
|
|||
import java.net.InetAddress; |
|||
import java.net.InetSocketAddress; |
|||
import java.net.UnknownHostException; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.locks.Lock; |
|||
|
|||
import static java.nio.charset.StandardCharsets.UTF_8; |
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.mockito.ArgumentMatchers.any; |
|||
import static org.mockito.ArgumentMatchers.eq; |
|||
import static org.mockito.Mockito.lenient; |
|||
import static org.mockito.Mockito.mock; |
|||
import static org.mockito.Mockito.times; |
|||
import static org.mockito.Mockito.verify; |
|||
import static org.thingsboard.server.transport.lwm2m.server.store.TbLwM2mRedisRegistrationStore.DEFAULT_CLEAN_LIMIT; |
|||
import static org.thingsboard.server.transport.lwm2m.server.store.TbLwM2mRedisRegistrationStore.DEFAULT_CLEAN_PERIOD; |
|||
import static org.thingsboard.server.transport.lwm2m.server.store.TbLwM2mRedisRegistrationStore.DEFAULT_GRACE_PERIOD; |
|||
|
|||
|
|||
@ExtendWith(MockitoExtension.class) |
|||
class TbLwM2mRedisRegistrationStoreTest { |
|||
|
|||
RedisConnectionFactory connectionFactory; |
|||
RedisConnection connection; |
|||
RedisLockRegistry lockRegistry; |
|||
|
|||
TbLwM2mRedisRegistrationStore registrationStore; |
|||
|
|||
@BeforeEach |
|||
void setUp() { |
|||
lockRegistry = mock(RedisLockRegistry.class); |
|||
lenient().when(lockRegistry.obtain(any())).thenReturn(mock(Lock.class)); |
|||
connection = mock(RedisConnection.class); |
|||
//when(connection.set(any(byte[].class), any(byte[].class))).
|
|||
connectionFactory = mock(RedisConnectionFactory.class); |
|||
lenient().when(connectionFactory.getConnection()).thenReturn(connection); |
|||
ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1, |
|||
new NamedThreadFactory(String.format("RedisRegistrationStore Cleaner (%ds)", DEFAULT_CLEAN_PERIOD))); |
|||
registrationStore = new TbLwM2mRedisRegistrationStore(connectionFactory, executorService, |
|||
DEFAULT_CLEAN_PERIOD, DEFAULT_GRACE_PERIOD, DEFAULT_CLEAN_LIMIT, lockRegistry); |
|||
} |
|||
|
|||
@Test |
|||
void testAddRegistrationWithNoOldRegistration() { |
|||
setOldRegistration(null); |
|||
Registration registration = buildRegistration(); |
|||
|
|||
assertThat(registrationStore.addRegistration(registration)).isNull(); |
|||
|
|||
byte[] endpoint = registration.getEndpoint().getBytes(UTF_8); |
|||
verify(connection, times(1)).set(getRegIdKey(registration), endpoint); |
|||
verify(connection, times(1)).set(getRegAddrKey(registration), endpoint); |
|||
verify(connection, times(1)).set(getRegIdentityKey(registration), endpoint); |
|||
verify(connection, times(3)).set(any(byte[].class), any(byte[].class)); |
|||
verify(connection, times(0)).del(any(byte[].class)); |
|||
} |
|||
|
|||
@Test |
|||
void testAddRegistrationWithOldRegistrationEqualToCurrent(){ |
|||
var oldRegistration = buildRegistration(); |
|||
setOldRegistration(oldRegistration); |
|||
Registration registration = buildRegistration(); |
|||
|
|||
var deregistration = registrationStore.addRegistration(registration); |
|||
|
|||
assertThat(deregistration.getRegistration()).isEqualTo(oldRegistration); |
|||
|
|||
byte[] endpoint = registration.getEndpoint().getBytes(UTF_8); |
|||
verify(connection, times(1)).set(getRegIdKey(registration), endpoint); |
|||
verify(connection, times(1)).set(getRegAddrKey(registration), endpoint); |
|||
verify(connection, times(1)).set(getRegIdentityKey(registration), endpoint); |
|||
verify(connection, times(3)).set(any(byte[].class), any(byte[].class)); |
|||
verify(connection, times(1)).del(getTknsRegIdKey(oldRegistration)); |
|||
verify(connection, times(1)).del(any(byte[].class)); |
|||
} |
|||
|
|||
@Test |
|||
void testAddRegistrationRemovesIndexes(){ |
|||
var oldRegistration = buildRegistration(Identity.unsecure(getTestAddress(1234))); |
|||
setOldRegistration(oldRegistration); |
|||
var registration = buildRegistration(Identity.unsecure(getTestAddress(2345))); |
|||
|
|||
var deregistration = registrationStore.addRegistration(registration); |
|||
|
|||
assertThat(deregistration.getRegistration()).isEqualTo(oldRegistration); |
|||
byte[] endpoint = registration.getEndpoint().getBytes(UTF_8); |
|||
verify(connection, times(1)).set(getRegIdKey(registration), endpoint); |
|||
verify(connection, times(1)).set(getRegAddrKey(registration), endpoint); |
|||
verify(connection, times(1)).set(getRegIdentityKey(registration), endpoint); |
|||
verify(connection, times(3)).set(any(byte[].class), any(byte[].class)); |
|||
verify(connection, times(1)).del(getRegAddrKey(oldRegistration)); |
|||
verify(connection, times(1)).del(getRegIdentityKey(oldRegistration)); |
|||
verify(connection, times(1)).del(getTknsRegIdKey(oldRegistration)); |
|||
verify(connection, times(3)).del(any(byte[].class)); |
|||
} |
|||
|
|||
@Test |
|||
void testUpdateRegistrationWhenNoRegistrationFound() { |
|||
setOldRegistration(null); |
|||
Registration registration = buildRegistration(); |
|||
RegistrationUpdate update = createUpdateFromRegistration(registration); |
|||
|
|||
assertThat(registrationStore.updateRegistration(update)).isNull(); |
|||
|
|||
verify(connection, times(1)).get(getRegIdKey(registration)); |
|||
verify(connection, times(1)).get(any(byte[].class)); |
|||
verify(connection, times(0)).del(any(byte[].class)); |
|||
} |
|||
|
|||
@Test |
|||
void testUpdateRegistrationWithSameRegistration() { |
|||
Registration registration = buildRegistration(); |
|||
setOldRegistration(registration); |
|||
RegistrationUpdate update = createUpdateFromRegistration(registration); |
|||
|
|||
assertThat(registrationStore.updateRegistration(update)).isNotNull(); |
|||
|
|||
var endpoint = registration.getEndpoint().getBytes(UTF_8); |
|||
// check registration and addressIndex here updated
|
|||
verify(connection, times(1)).set(eq(getEndpointKey(endpoint)), any(byte[].class)); |
|||
verify(connection, times(1)).set(getRegAddrKey(registration), endpoint); |
|||
verify(connection, times(2)).set(any(byte[].class), any(byte[].class)); |
|||
verify(connection, times(0)).del(any(byte[].class)); |
|||
} |
|||
|
|||
@Test |
|||
void testUpdateRegistrationWithRegistrationFromSecureIdentitiesWithDifferentAddress() { |
|||
Registration oldRegistration = buildRegistration(Identity.psk(getTestAddress(1234), "my:psk")); |
|||
setOldRegistration(oldRegistration); |
|||
Registration newRegistration = buildRegistration(Identity.psk(getTestAddress(2345), "my:psk")); |
|||
RegistrationUpdate update = createUpdateFromRegistration(newRegistration); |
|||
assertThat(oldRegistration.getEndpoint()).isEqualTo(newRegistration.getEndpoint()); |
|||
|
|||
assertThat(registrationStore.updateRegistration(update)).isNotNull(); |
|||
|
|||
var endpoint = newRegistration.getEndpoint().getBytes(UTF_8); |
|||
// check registration and addressIndex here updated
|
|||
verify(connection, times(1)).set(eq(getEndpointKey(endpoint)), any(byte[].class)); |
|||
verify(connection, times(1)).set(getRegAddrKey(newRegistration), endpoint); |
|||
// check old AddrIndex has been removed
|
|||
verify(connection, times(1)).del(getRegAddrKey(oldRegistration)); |
|||
// check identityIndex has not been removed
|
|||
verify(connection, times(0)).del(getRegIdentityKey(oldRegistration)); |
|||
// check only one key (AddrIndex) in total was removed
|
|||
verify(connection, times(1)).del(any(byte[].class)); |
|||
} |
|||
|
|||
@Test |
|||
void testGetRegistrationByIdentityReturnsRegistrationForSecureIdentityWithDifferentAddress() { |
|||
Registration registration = buildRegistration(Identity.psk(getTestAddress(1234), "my:psk")); |
|||
setOldRegistration(registration); |
|||
Identity sameIdentityWithDifferentAddress = Identity.psk(getTestAddress(2345), "my:psk"); |
|||
|
|||
Registration retrievedRegistration = registrationStore.getRegistrationByIdentity(sameIdentityWithDifferentAddress); |
|||
|
|||
assertThat(retrievedRegistration).isEqualTo(registration); |
|||
} |
|||
|
|||
private void setOldRegistration(Registration oldRegistration){ |
|||
byte[] serializedRegistration = null; |
|||
if (oldRegistration != null){ |
|||
byte[] endpoint = oldRegistration.getEndpoint().getBytes(UTF_8); |
|||
// set the AddrIndex
|
|||
byte[] regAddrKey = getRegAddrKey(oldRegistration); |
|||
lenient().when(connection.get(eq(regAddrKey))).thenReturn(endpoint); |
|||
// set the IdentityIndex
|
|||
byte[] regIdentityKey = getRegIdentityKey(oldRegistration); |
|||
lenient().when(connection.get(eq(regIdentityKey))).thenReturn(endpoint); |
|||
// set the IdIndex
|
|||
byte[] regIdKey = getRegIdKey(oldRegistration); |
|||
lenient().when(connection.get(eq(regIdKey))).thenReturn(endpoint); |
|||
// set the registration
|
|||
serializedRegistration = RegistrationSerDes.bSerialize(oldRegistration); |
|||
lenient().when(connection.get(eq(getEndpointKey(endpoint)))).thenReturn(serializedRegistration); |
|||
} |
|||
lenient().when(connection.getSet(any(byte[].class), any(byte[].class))).thenReturn(serializedRegistration); |
|||
} |
|||
|
|||
private byte[] getRegAddrKey(Registration registration){ |
|||
return ReflectionTestUtils.invokeMethod(registrationStore, "toRegAddrKey", registration.getSocketAddress()); |
|||
} |
|||
|
|||
private byte[] getRegIdentityKey(Registration registration){ |
|||
return ReflectionTestUtils.invokeMethod(registrationStore, "toRegIdentityKey", registration.getIdentity()); |
|||
} |
|||
|
|||
private byte[] getRegIdKey(Registration registration){ |
|||
return ReflectionTestUtils.invokeMethod(registrationStore, "toRegIdKey", registration.getId()); |
|||
} |
|||
|
|||
private byte[] getEndpointKey(byte[] endpoint){ |
|||
return ReflectionTestUtils.invokeMethod(registrationStore, "toEndpointKey", endpoint); |
|||
} |
|||
|
|||
private byte[] getTknsRegIdKey(Registration registration){ |
|||
return ReflectionTestUtils.invokeMethod(registrationStore, "toKey", "TKNS:REGID:", registration.getId()); |
|||
} |
|||
|
|||
private static Registration buildRegistration() { |
|||
return buildRegistration(Identity.psk(getTestAddress(), "my:psk")); |
|||
} |
|||
|
|||
private static Registration buildRegistration(Identity identity){ |
|||
return new Registration.Builder("my_reg_id", "abcde", identity) |
|||
.objectLinks(new Link[]{}) |
|||
.build(); |
|||
} |
|||
|
|||
private static RegistrationUpdate createUpdateFromRegistration(Registration registration){ |
|||
return new RegistrationUpdate( |
|||
registration.getId(), |
|||
registration.getIdentity(), |
|||
registration.getLifeTimeInSec(), |
|||
registration.getSmsNumber(), |
|||
registration.getBindingMode(), |
|||
registration.getObjectLinks(), |
|||
registration.getAdditionalRegistrationAttributes() |
|||
); |
|||
} |
|||
|
|||
private static InetSocketAddress getTestAddress() { |
|||
return getTestAddress(5684); |
|||
} |
|||
|
|||
private static InetSocketAddress getTestAddress(int port) { |
|||
try { |
|||
return new InetSocketAddress(InetAddress.getByName("1.2.3.4"), port); |
|||
} catch (UnknownHostException e) { |
|||
throw new AssertionError("Cannot create test address"); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,77 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.transport.lwm2m.server.store.util; |
|||
|
|||
import com.eclipsesource.json.JsonObject; |
|||
import org.apache.commons.lang3.NotImplementedException; |
|||
import org.eclipse.leshan.core.request.Identity; |
|||
import org.junit.jupiter.api.Test; |
|||
|
|||
import java.net.InetAddress; |
|||
import java.net.InetSocketAddress; |
|||
import java.net.UnknownHostException; |
|||
import java.security.PublicKey; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.assertj.core.api.Assertions.assertThatThrownBy; |
|||
import static org.mockito.Mockito.mock; |
|||
import static org.mockito.Mockito.when; |
|||
|
|||
class LwM2MIdentitySerDesTest { |
|||
|
|||
@Test |
|||
void serializePskIdentity() { |
|||
assertThat(LwM2MIdentitySerDes.serialize(Identity.psk(getTestAddress(), "my:psk")).toString()) |
|||
.isEqualTo("{\"type\":\"psk\",\"id\":\"my:psk\"}"); |
|||
} |
|||
|
|||
|
|||
@Test |
|||
void serializeRpkIdentity() { |
|||
var public_key = mock(PublicKey.class); |
|||
when(public_key.getEncoded()).thenReturn(new byte[]{1,2,3,4,5,6,7,8,9}); |
|||
|
|||
assertThat(LwM2MIdentitySerDes.serialize(Identity.rpk(getTestAddress(), public_key)).toString()) |
|||
.isEqualTo("{\"type\":\"rpk\",\"rpk\":\"010203040506070809\"}"); |
|||
} |
|||
|
|||
@Test |
|||
void serializeX509Identity() { |
|||
assertThat(LwM2MIdentitySerDes.serialize(Identity.x509(getTestAddress(), "MyCommonName")).toString()) |
|||
.isEqualTo("{\"type\":\"x509\",\"cn\":\"MyCommonName\"}"); |
|||
} |
|||
|
|||
@Test |
|||
void serializeUnsecureIdentity() { |
|||
assertThat(LwM2MIdentitySerDes.serialize(Identity.unsecure(getTestAddress())).toString()) |
|||
.isEqualTo("{\"type\":\"unsecure\",\"address\":\"1.2.3.4\",\"port\":5684}"); |
|||
} |
|||
|
|||
|
|||
@Test |
|||
void deserialize() { |
|||
assertThatThrownBy(() -> LwM2MIdentitySerDes.deserialize(mock(JsonObject.class))) |
|||
.isInstanceOf(NotImplementedException.class); |
|||
} |
|||
|
|||
private static InetSocketAddress getTestAddress() { |
|||
try { |
|||
return new InetSocketAddress(InetAddress.getByName("1.2.3.4"), 5684); |
|||
} catch (UnknownHostException e) { |
|||
throw new AssertionError("Cannot create test address"); |
|||
} |
|||
} |
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue