|
|
|
@ -24,7 +24,6 @@ import org.mockito.Mockito; |
|
|
|
import org.mockito.junit.MockitoJUnitRunner; |
|
|
|
import org.springframework.test.util.ReflectionTestUtils; |
|
|
|
import org.thingsboard.server.cluster.TbClusterService; |
|
|
|
import org.thingsboard.server.common.data.DataConstants; |
|
|
|
import org.thingsboard.server.common.data.DeviceIdInfo; |
|
|
|
import org.thingsboard.server.common.data.id.DeviceId; |
|
|
|
import org.thingsboard.server.common.data.id.TenantId; |
|
|
|
@ -41,14 +40,13 @@ import org.thingsboard.server.dao.device.DeviceService; |
|
|
|
import org.thingsboard.server.dao.sql.query.EntityQueryRepository; |
|
|
|
import org.thingsboard.server.dao.timeseries.TimeseriesService; |
|
|
|
import org.thingsboard.server.queue.discovery.PartitionService; |
|
|
|
import org.thingsboard.server.service.partition.AbstractPartitionBasedService; |
|
|
|
import org.thingsboard.server.queue.discovery.QueueKey; |
|
|
|
import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; |
|
|
|
import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; |
|
|
|
|
|
|
|
import java.lang.reflect.Method; |
|
|
|
import java.util.Collections; |
|
|
|
import java.util.List; |
|
|
|
import java.util.Map; |
|
|
|
import java.util.Set; |
|
|
|
import java.util.UUID; |
|
|
|
|
|
|
|
import static org.hamcrest.CoreMatchers.is; |
|
|
|
@ -78,13 +76,29 @@ public class DefaultDeviceStateServiceTest { |
|
|
|
@Mock |
|
|
|
EntityQueryRepository entityQueryRepository; |
|
|
|
|
|
|
|
TenantId tenantId = new TenantId(UUID.fromString("00797a3b-7aeb-4b5b-b57a-c2a810d0f112")); |
|
|
|
DeviceId deviceId = DeviceId.fromString("00797a3b-7aeb-4b5b-b57a-c2a810d0f112"); |
|
|
|
TopicPartitionInfo tpi; |
|
|
|
|
|
|
|
DefaultDeviceStateService service; |
|
|
|
|
|
|
|
TelemetrySubscriptionService telemetrySubscriptionService; |
|
|
|
|
|
|
|
@Before |
|
|
|
public void setUp() { |
|
|
|
service = spy(new DefaultDeviceStateService(deviceService, attributesService, tsService, clusterService, partitionService, entityQueryRepository, null, null, mock(NotificationRuleProcessor.class))); |
|
|
|
telemetrySubscriptionService = Mockito.mock(TelemetrySubscriptionService.class); |
|
|
|
ReflectionTestUtils.setField(service, "tsSubService", telemetrySubscriptionService); |
|
|
|
ReflectionTestUtils.setField(service, "defaultStateCheckIntervalInSec", 60); |
|
|
|
ReflectionTestUtils.setField(service, "defaultActivityStatsIntervalInSec", 60); |
|
|
|
ReflectionTestUtils.setField(service, "initFetchPackSize", 10); |
|
|
|
|
|
|
|
tpi = TopicPartitionInfo.builder().myPartition(true).build(); |
|
|
|
Mockito.when(partitionService.resolve(ServiceType.TB_CORE, tenantId, deviceId)).thenReturn(tpi); |
|
|
|
Mockito.when(entityQueryRepository.findEntityDataByQueryInternal(Mockito.any())).thenReturn(new PageData<>()); |
|
|
|
var deviceIdInfo = new DeviceIdInfo(tenantId.getId(), null, deviceId.getId()); |
|
|
|
Mockito.when(deviceService.findDeviceIdInfos(Mockito.any())) |
|
|
|
.thenReturn(new PageData<>(List.of(deviceIdInfo), 0, 1, false)); |
|
|
|
} |
|
|
|
|
|
|
|
@Test |
|
|
|
@ -140,35 +154,62 @@ public class DefaultDeviceStateServiceTest { |
|
|
|
Assert.assertEquals(5000L, deviceStateData.getState().getInactivityTimeout()); |
|
|
|
} |
|
|
|
|
|
|
|
private void initStateService(long timeout) throws InterruptedException { |
|
|
|
service.stop(); |
|
|
|
Mockito.reset(service, telemetrySubscriptionService); |
|
|
|
ReflectionTestUtils.setField(service, "defaultInactivityTimeoutMs", timeout); |
|
|
|
service.init(); |
|
|
|
PartitionChangeEvent event = new PartitionChangeEvent(this, new QueueKey(ServiceType.TB_CORE), Collections.singleton(tpi)); |
|
|
|
service.onApplicationEvent(event); |
|
|
|
Thread.sleep(100); |
|
|
|
} |
|
|
|
|
|
|
|
@Test |
|
|
|
public void givenIncreaseInactivityTimeoutAndThenStateIsActive() throws Exception { |
|
|
|
TelemetrySubscriptionService telemetrySubscriptionService = Mockito.mock(TelemetrySubscriptionService.class); |
|
|
|
ReflectionTestUtils.setField(service, "tsSubService", telemetrySubscriptionService); |
|
|
|
ReflectionTestUtils.setField(service, "defaultStateCheckIntervalInSec", 60); |
|
|
|
ReflectionTestUtils.setField(service, "defaultActivityStatsIntervalInSec", 60); |
|
|
|
ReflectionTestUtils.setField(service, "defaultInactivityTimeoutMs", 1); |
|
|
|
ReflectionTestUtils.setField(service, "initFetchPackSize", 10); |
|
|
|
public void increaseInactivityForInactiveDeviceTest() throws Exception { |
|
|
|
final long defaultTimeout = 1; |
|
|
|
initStateService(defaultTimeout); |
|
|
|
DeviceState deviceState = DeviceState.builder().build(); |
|
|
|
DeviceStateData deviceStateData = DeviceStateData.builder() |
|
|
|
.tenantId(tenantId) |
|
|
|
.deviceId(deviceId) |
|
|
|
.state(deviceState) |
|
|
|
.metaData(new TbMsgMetaData()) |
|
|
|
.build(); |
|
|
|
|
|
|
|
Mockito.when(entityQueryRepository.findEntityDataByQueryInternal(Mockito.any())).thenReturn(new PageData<>()); |
|
|
|
service.deviceStates.put(deviceId, deviceStateData); |
|
|
|
service.getPartitionedEntities(tpi).add(deviceId); |
|
|
|
|
|
|
|
service.init(); |
|
|
|
var tenantId = new TenantId(UUID.randomUUID()); |
|
|
|
var tpi = TopicPartitionInfo.builder().myPartition(true).build(); |
|
|
|
Mockito.when(partitionService.resolve(ServiceType.TB_CORE, tenantId, deviceId)).thenReturn(tpi); |
|
|
|
service.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis()); |
|
|
|
activityVerify(true); |
|
|
|
Thread.sleep(defaultTimeout); |
|
|
|
service.checkStates(); |
|
|
|
activityVerify(false); |
|
|
|
|
|
|
|
var deviceIdInfo = new DeviceIdInfo(tenantId.getId(), null, deviceId.getId()); |
|
|
|
Mockito.reset(telemetrySubscriptionService); |
|
|
|
|
|
|
|
Mockito.when(deviceService.findDeviceIdInfos(Mockito.any())) |
|
|
|
.thenReturn(new PageData<>(List.of(deviceIdInfo), 0, 1, false)); |
|
|
|
long increase = 100; |
|
|
|
long newTimeout = System.currentTimeMillis() - deviceState.getLastActivityTime() + increase; |
|
|
|
|
|
|
|
service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, newTimeout); |
|
|
|
activityVerify(true); |
|
|
|
Thread.sleep(increase); |
|
|
|
service.checkStates(); |
|
|
|
activityVerify(false); |
|
|
|
|
|
|
|
Method method = AbstractPartitionBasedService.class.getDeclaredMethod("initStateFromDB", Set.class); |
|
|
|
method.setAccessible(true); |
|
|
|
method.invoke(service, Collections.singleton(tpi)); |
|
|
|
Mockito.reset(telemetrySubscriptionService); |
|
|
|
|
|
|
|
service.onAddedPartitions(Collections.singleton(tpi)); |
|
|
|
service.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis()); |
|
|
|
activityVerify(true); |
|
|
|
Thread.sleep(newTimeout + 5); |
|
|
|
service.checkStates(); |
|
|
|
activityVerify(false); |
|
|
|
} |
|
|
|
|
|
|
|
@Test |
|
|
|
public void increaseInactivityForActiveDeviceTest() throws Exception { |
|
|
|
final long defaultTimeout = 1000; |
|
|
|
initStateService(defaultTimeout); |
|
|
|
DeviceState deviceState = DeviceState.builder().build(); |
|
|
|
|
|
|
|
DeviceStateData deviceStateData = DeviceStateData.builder() |
|
|
|
.tenantId(tenantId) |
|
|
|
.deviceId(deviceId) |
|
|
|
@ -177,44 +218,124 @@ public class DefaultDeviceStateServiceTest { |
|
|
|
.build(); |
|
|
|
|
|
|
|
service.deviceStates.put(deviceId, deviceStateData); |
|
|
|
|
|
|
|
service.getPartitionedEntities(tpi).add(deviceId); |
|
|
|
|
|
|
|
service.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis()); |
|
|
|
activityVerify(true); |
|
|
|
|
|
|
|
Mockito.verify(telemetrySubscriptionService, Mockito.times(1)).saveAttrAndNotify(Mockito.any(), Mockito.eq(deviceId), Mockito.any(), Mockito.eq("active"), Mockito.eq(true), Mockito.any()); |
|
|
|
Mockito.reset(telemetrySubscriptionService); |
|
|
|
|
|
|
|
Thread.sleep(1); |
|
|
|
long increase = 100; |
|
|
|
long newTimeout = System.currentTimeMillis() - deviceState.getLastActivityTime() + increase; |
|
|
|
|
|
|
|
service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, newTimeout); |
|
|
|
Mockito.verify(telemetrySubscriptionService, Mockito.never()).saveAttrAndNotify(Mockito.any(), Mockito.eq(deviceId), Mockito.any(), Mockito.eq("active"), Mockito.any(), Mockito.any()); |
|
|
|
Thread.sleep(defaultTimeout + increase); |
|
|
|
service.checkStates(); |
|
|
|
|
|
|
|
Mockito.verify(telemetrySubscriptionService, Mockito.times(1)).saveAttrAndNotify(Mockito.any(), Mockito.eq(deviceId), Mockito.any(), Mockito.eq("active"), Mockito.eq(false), Mockito.any()); |
|
|
|
activityVerify(false); |
|
|
|
|
|
|
|
Mockito.reset(telemetrySubscriptionService); |
|
|
|
|
|
|
|
service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, System.currentTimeMillis() - deviceState.getLastActivityTime() + 1000); |
|
|
|
service.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis()); |
|
|
|
activityVerify(true); |
|
|
|
Thread.sleep(newTimeout); |
|
|
|
service.checkStates(); |
|
|
|
activityVerify(false); |
|
|
|
} |
|
|
|
|
|
|
|
Mockito.verify(telemetrySubscriptionService, Mockito.times(1)).saveAttrAndNotify(Mockito.any(), Mockito.eq(deviceId), Mockito.any(), Mockito.eq("active"), Mockito.eq(true), Mockito.any()); |
|
|
|
@Test |
|
|
|
public void increaseSmallInactivityForInactiveDeviceTest() throws Exception { |
|
|
|
final long defaultTimeout = 1; |
|
|
|
initStateService(defaultTimeout); |
|
|
|
DeviceState deviceState = DeviceState.builder().build(); |
|
|
|
DeviceStateData deviceStateData = DeviceStateData.builder() |
|
|
|
.tenantId(tenantId) |
|
|
|
.deviceId(deviceId) |
|
|
|
.state(deviceState) |
|
|
|
.metaData(new TbMsgMetaData()) |
|
|
|
.build(); |
|
|
|
|
|
|
|
Thread.sleep(2000); |
|
|
|
service.deviceStates.put(deviceId, deviceStateData); |
|
|
|
service.getPartitionedEntities(tpi).add(deviceId); |
|
|
|
|
|
|
|
service.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis()); |
|
|
|
activityVerify(true); |
|
|
|
Thread.sleep(defaultTimeout); |
|
|
|
service.checkStates(); |
|
|
|
activityVerify(false); |
|
|
|
|
|
|
|
Mockito.verify(telemetrySubscriptionService, Mockito.times(1)).saveAttrAndNotify(Mockito.any(), Mockito.eq(deviceId), Mockito.any(), Mockito.eq("active"), Mockito.eq(false), Mockito.any()); |
|
|
|
Mockito.reset(telemetrySubscriptionService); |
|
|
|
|
|
|
|
long newTimeout = 1; |
|
|
|
Thread.sleep(newTimeout); |
|
|
|
Mockito.verify(telemetrySubscriptionService, Mockito.never()).saveAttrAndNotify(Mockito.any(), Mockito.eq(deviceId), Mockito.any(), Mockito.eq("active"), Mockito.any(), Mockito.any()); |
|
|
|
} |
|
|
|
|
|
|
|
@Test |
|
|
|
public void decreaseInactivityForActiveDeviceTest() throws Exception { |
|
|
|
final long defaultTimeout = 1000; |
|
|
|
initStateService(defaultTimeout); |
|
|
|
DeviceState deviceState = DeviceState.builder().build(); |
|
|
|
DeviceStateData deviceStateData = DeviceStateData.builder() |
|
|
|
.tenantId(tenantId) |
|
|
|
.deviceId(deviceId) |
|
|
|
.state(deviceState) |
|
|
|
.metaData(new TbMsgMetaData()) |
|
|
|
.build(); |
|
|
|
|
|
|
|
service.deviceStates.put(deviceId, deviceStateData); |
|
|
|
service.getPartitionedEntities(tpi).add(deviceId); |
|
|
|
|
|
|
|
service.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis()); |
|
|
|
activityVerify(true); |
|
|
|
|
|
|
|
Mockito.reset(telemetrySubscriptionService); |
|
|
|
|
|
|
|
Mockito.verify(telemetrySubscriptionService, Mockito.never()).saveAttrAndNotify(Mockito.any(), Mockito.eq(deviceId), Mockito.any(), Mockito.eq("active"), Mockito.any(), Mockito.any()); |
|
|
|
|
|
|
|
long newTimeout = 1; |
|
|
|
|
|
|
|
service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, newTimeout); |
|
|
|
activityVerify(false); |
|
|
|
Mockito.reset(telemetrySubscriptionService); |
|
|
|
|
|
|
|
service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, 2000); |
|
|
|
service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, defaultTimeout); |
|
|
|
activityVerify(true); |
|
|
|
Thread.sleep(defaultTimeout); |
|
|
|
service.checkStates(); |
|
|
|
activityVerify(false); |
|
|
|
} |
|
|
|
|
|
|
|
@Test |
|
|
|
public void decreaseInactivityForInactiveDeviceTest() throws Exception { |
|
|
|
final long defaultTimeout = 1000; |
|
|
|
initStateService(defaultTimeout); |
|
|
|
DeviceState deviceState = DeviceState.builder().build(); |
|
|
|
DeviceStateData deviceStateData = DeviceStateData.builder() |
|
|
|
.tenantId(tenantId) |
|
|
|
.deviceId(deviceId) |
|
|
|
.state(deviceState) |
|
|
|
.metaData(new TbMsgMetaData()) |
|
|
|
.build(); |
|
|
|
|
|
|
|
Mockito.verify(telemetrySubscriptionService, Mockito.never()).saveAttrAndNotify(Mockito.any(), Mockito.eq(deviceId), Mockito.any(), Mockito.eq("active"), Mockito.eq(true), Mockito.any()); |
|
|
|
service.deviceStates.put(deviceId, deviceStateData); |
|
|
|
service.getPartitionedEntities(tpi).add(deviceId); |
|
|
|
|
|
|
|
service.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis()); |
|
|
|
activityVerify(true); |
|
|
|
Thread.sleep(defaultTimeout); |
|
|
|
service.checkStates(); |
|
|
|
activityVerify(false); |
|
|
|
Mockito.reset(telemetrySubscriptionService); |
|
|
|
|
|
|
|
Mockito.verify(telemetrySubscriptionService, Mockito.times(1)).saveAttrAndNotify(Mockito.any(), Mockito.eq(deviceId), Mockito.any(), Mockito.eq("active"), Mockito.eq(true), Mockito.any()); |
|
|
|
long newTimeout = 1; |
|
|
|
|
|
|
|
service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, 1); |
|
|
|
service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, newTimeout); |
|
|
|
Mockito.verify(telemetrySubscriptionService, Mockito.never()).saveAttrAndNotify(Mockito.any(), Mockito.eq(deviceId), Mockito.any(), Mockito.eq("active"), Mockito.any(), Mockito.any()); |
|
|
|
} |
|
|
|
|
|
|
|
Mockito.verify(telemetrySubscriptionService, Mockito.times(1)).saveAttrAndNotify(Mockito.any(), Mockito.eq(deviceId), Mockito.any(), Mockito.eq("active"), Mockito.eq(false), Mockito.any()); |
|
|
|
private void activityVerify(boolean isActive) { |
|
|
|
Mockito.verify(telemetrySubscriptionService, Mockito.times(1)).saveAttrAndNotify(Mockito.any(), Mockito.eq(deviceId), Mockito.any(), Mockito.eq("active"), Mockito.eq(isActive), Mockito.any()); |
|
|
|
} |
|
|
|
|
|
|
|
} |