committed by
GitHub
45 changed files with 2726 additions and 419 deletions
@ -0,0 +1,36 @@ |
|||||
|
/** |
||||
|
* 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.install; |
||||
|
|
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import org.springframework.boot.info.BuildProperties; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
|
||||
|
@Component |
||||
|
@RequiredArgsConstructor |
||||
|
public class ProjectInfo { |
||||
|
|
||||
|
private final BuildProperties buildProperties; |
||||
|
|
||||
|
public String getProjectVersion() { |
||||
|
return buildProperties.getVersion().replaceAll("[^\\d.]", ""); |
||||
|
} |
||||
|
|
||||
|
public String getProductType() { |
||||
|
return "CE"; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,107 @@ |
|||||
|
/** |
||||
|
* 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.entitiy.entityview; |
||||
|
|
||||
|
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.thingsboard.rule.engine.api.TimeseriesSaveRequest; |
||||
|
import org.thingsboard.server.common.data.EntityView; |
||||
|
import org.thingsboard.server.common.data.id.DeviceId; |
||||
|
import org.thingsboard.server.common.data.id.EntityId; |
||||
|
import org.thingsboard.server.common.data.id.EntityViewId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.kv.BasicTsKvEntry; |
||||
|
import org.thingsboard.server.common.data.kv.DoubleDataEntry; |
||||
|
import org.thingsboard.server.common.data.kv.TsKvEntry; |
||||
|
import org.thingsboard.server.common.data.objects.AttributesEntityView; |
||||
|
import org.thingsboard.server.common.data.objects.TelemetryEntityView; |
||||
|
import org.thingsboard.server.dao.attributes.AttributesService; |
||||
|
import org.thingsboard.server.dao.entityview.EntityViewService; |
||||
|
import org.thingsboard.server.dao.timeseries.TimeseriesService; |
||||
|
import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; |
||||
|
|
||||
|
import java.util.List; |
||||
|
import java.util.UUID; |
||||
|
|
||||
|
import static com.google.common.util.concurrent.Futures.immediateFuture; |
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
import static org.mockito.ArgumentMatchers.anyList; |
||||
|
import static org.mockito.ArgumentMatchers.eq; |
||||
|
import static org.mockito.BDDMockito.given; |
||||
|
import static org.mockito.BDDMockito.then; |
||||
|
|
||||
|
@ExtendWith(MockitoExtension.class) |
||||
|
class DefaultTbEntityViewServiceTest { |
||||
|
|
||||
|
final TenantId tenantId = TenantId.fromUUID(UUID.fromString("f09c8180-686c-11ef-9471-a71d33080e9c")); |
||||
|
final EntityId entityId = DeviceId.fromString("782aaab0-c7a8-11ef-a668-79582e785d5f"); |
||||
|
|
||||
|
@Mock |
||||
|
EntityViewService entityViewService; |
||||
|
@Mock |
||||
|
AttributesService attributesService; |
||||
|
@Mock |
||||
|
TelemetrySubscriptionService tsSubService; |
||||
|
@Mock |
||||
|
TimeseriesService tsService; |
||||
|
|
||||
|
DefaultTbEntityViewService defaultTbEntityViewService; |
||||
|
|
||||
|
@BeforeEach |
||||
|
void setup() { |
||||
|
defaultTbEntityViewService = new DefaultTbEntityViewService(entityViewService, attributesService, tsSubService, tsService); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldNotSaveTimeseriesWhenCopyingLatestToEntityView() throws Exception { |
||||
|
// GIVEN
|
||||
|
var entityView = new EntityView(new EntityViewId(UUID.randomUUID())); |
||||
|
entityView.setTenantId(tenantId); |
||||
|
entityView.setEntityId(entityId); |
||||
|
entityView.setKeys(new TelemetryEntityView(List.of("temperature"), new AttributesEntityView())); |
||||
|
|
||||
|
List<TsKvEntry> latest = List.of(new BasicTsKvEntry(123L, new DoubleDataEntry("temperature", 22.3))); |
||||
|
|
||||
|
given(tsService.findAll(eq(tenantId), eq(entityId), anyList())).willReturn(immediateFuture(latest)); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbEntityViewService.updateEntityViewAttributes(tenantId, entityView, null, null); |
||||
|
|
||||
|
// THEN
|
||||
|
var captor = ArgumentCaptor.forClass(TimeseriesSaveRequest.class); |
||||
|
then(tsSubService).should().saveTimeseries(captor.capture()); |
||||
|
|
||||
|
var expectedCopyLatestRequest = TimeseriesSaveRequest.builder() |
||||
|
.tenantId(tenantId) |
||||
|
.entityId(entityView.getId()) |
||||
|
.entries(latest) |
||||
|
.ttl(0L) |
||||
|
.strategy(TimeseriesSaveRequest.Strategy.LATEST_AND_WS) |
||||
|
.build(); |
||||
|
|
||||
|
var actualCopyLatestRequest = captor.getValue(); |
||||
|
|
||||
|
assertThat(actualCopyLatestRequest) |
||||
|
.usingRecursiveComparison() |
||||
|
.ignoringFields("callback") |
||||
|
.isEqualTo(expectedCopyLatestRequest); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,373 @@ |
|||||
|
/** |
||||
|
* 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.telemetry; |
||||
|
|
||||
|
import com.google.common.util.concurrent.FutureCallback; |
||||
|
import com.google.common.util.concurrent.MoreExecutors; |
||||
|
import com.google.common.util.concurrent.SettableFuture; |
||||
|
import org.checkerframework.checker.nullness.qual.NonNull; |
||||
|
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.junit.jupiter.params.ParameterizedTest; |
||||
|
import org.junit.jupiter.params.provider.Arguments; |
||||
|
import org.junit.jupiter.params.provider.MethodSource; |
||||
|
import org.mockito.Mock; |
||||
|
import org.mockito.junit.jupiter.MockitoExtension; |
||||
|
import org.springframework.test.util.ReflectionTestUtils; |
||||
|
import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; |
||||
|
import org.thingsboard.server.cluster.TbClusterService; |
||||
|
import org.thingsboard.server.common.data.ApiUsageRecordKey; |
||||
|
import org.thingsboard.server.common.data.ApiUsageState; |
||||
|
import org.thingsboard.server.common.data.ApiUsageStateValue; |
||||
|
import org.thingsboard.server.common.data.EntityView; |
||||
|
import org.thingsboard.server.common.data.id.CustomerId; |
||||
|
import org.thingsboard.server.common.data.id.DeviceId; |
||||
|
import org.thingsboard.server.common.data.id.EntityId; |
||||
|
import org.thingsboard.server.common.data.id.EntityViewId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.kv.BasicTsKvEntry; |
||||
|
import org.thingsboard.server.common.data.kv.DoubleDataEntry; |
||||
|
import org.thingsboard.server.common.data.kv.KvEntry; |
||||
|
import org.thingsboard.server.common.data.kv.TsKvEntry; |
||||
|
import org.thingsboard.server.common.data.objects.AttributesEntityView; |
||||
|
import org.thingsboard.server.common.data.objects.TelemetryEntityView; |
||||
|
import org.thingsboard.server.common.msg.queue.ServiceType; |
||||
|
import org.thingsboard.server.common.msg.queue.TbCallback; |
||||
|
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
||||
|
import org.thingsboard.server.common.stats.TbApiUsageReportClient; |
||||
|
import org.thingsboard.server.dao.attributes.AttributesService; |
||||
|
import org.thingsboard.server.dao.timeseries.TimeseriesService; |
||||
|
import org.thingsboard.server.queue.discovery.PartitionService; |
||||
|
import org.thingsboard.server.queue.discovery.QueueKey; |
||||
|
import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; |
||||
|
import org.thingsboard.server.service.apiusage.TbApiUsageStateService; |
||||
|
import org.thingsboard.server.service.entitiy.entityview.TbEntityViewService; |
||||
|
import org.thingsboard.server.service.subscription.SubscriptionManagerService; |
||||
|
|
||||
|
import java.time.Duration; |
||||
|
import java.util.Collections; |
||||
|
import java.util.List; |
||||
|
import java.util.Map; |
||||
|
import java.util.Optional; |
||||
|
import java.util.Set; |
||||
|
import java.util.UUID; |
||||
|
import java.util.concurrent.ExecutionException; |
||||
|
import java.util.concurrent.ExecutorService; |
||||
|
import java.util.stream.LongStream; |
||||
|
import java.util.stream.Stream; |
||||
|
|
||||
|
import static com.google.common.util.concurrent.Futures.immediateFuture; |
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
import static org.mockito.BDDMockito.given; |
||||
|
import static org.mockito.BDDMockito.then; |
||||
|
import static org.mockito.Mockito.lenient; |
||||
|
|
||||
|
@ExtendWith(MockitoExtension.class) |
||||
|
class DefaultTelemetrySubscriptionServiceTest { |
||||
|
|
||||
|
final TenantId tenantId = TenantId.fromUUID(UUID.fromString("a00ec470-c6b4-11ef-8c88-63b5533fb5bc")); |
||||
|
final CustomerId customerId = new CustomerId(UUID.fromString("7bdc9750-c775-11ef-8e03-ff69ed8da327")); |
||||
|
final EntityId entityId = DeviceId.fromString("cc51e450-53e1-11ee-883e-e56b48fd2088"); |
||||
|
|
||||
|
final long sampleTtl = 10_000L; |
||||
|
|
||||
|
final List<TsKvEntry> sampleTelemetry = List.of( |
||||
|
new BasicTsKvEntry(100L, new DoubleDataEntry("temperature", 65.2)), |
||||
|
new BasicTsKvEntry(100L, new DoubleDataEntry("humidity", 33.1)) |
||||
|
); |
||||
|
|
||||
|
ApiUsageState apiUsageState; |
||||
|
|
||||
|
final TopicPartitionInfo tpi = TopicPartitionInfo.builder() |
||||
|
.tenantId(tenantId) |
||||
|
.myPartition(true) |
||||
|
.build(); |
||||
|
|
||||
|
final FutureCallback<Void> emptyCallback = new FutureCallback<>() { |
||||
|
@Override |
||||
|
public void onSuccess(Void result) {} |
||||
|
|
||||
|
@Override |
||||
|
public void onFailure(@NonNull Throwable t) {} |
||||
|
}; |
||||
|
|
||||
|
ExecutorService wsCallBackExecutor; |
||||
|
ExecutorService tsCallBackExecutor; |
||||
|
|
||||
|
@Mock |
||||
|
TbClusterService clusterService; |
||||
|
@Mock |
||||
|
PartitionService partitionService; |
||||
|
@Mock |
||||
|
SubscriptionManagerService subscriptionManagerService; |
||||
|
@Mock |
||||
|
AttributesService attrService; |
||||
|
@Mock |
||||
|
TimeseriesService tsService; |
||||
|
@Mock |
||||
|
TbEntityViewService tbEntityViewService; |
||||
|
@Mock |
||||
|
TbApiUsageReportClient apiUsageClient; |
||||
|
@Mock |
||||
|
TbApiUsageStateService apiUsageStateService; |
||||
|
|
||||
|
DefaultTelemetrySubscriptionService telemetryService; |
||||
|
|
||||
|
@BeforeEach |
||||
|
void setup() { |
||||
|
telemetryService = new DefaultTelemetrySubscriptionService(attrService, tsService, tbEntityViewService, apiUsageClient, apiUsageStateService); |
||||
|
ReflectionTestUtils.setField(telemetryService, "clusterService", clusterService); |
||||
|
ReflectionTestUtils.setField(telemetryService, "partitionService", partitionService); |
||||
|
ReflectionTestUtils.setField(telemetryService, "subscriptionManagerService", Optional.of(subscriptionManagerService)); |
||||
|
|
||||
|
wsCallBackExecutor = MoreExecutors.newDirectExecutorService(); |
||||
|
ReflectionTestUtils.setField(telemetryService, "wsCallBackExecutor", wsCallBackExecutor); |
||||
|
|
||||
|
tsCallBackExecutor = MoreExecutors.newDirectExecutorService(); |
||||
|
ReflectionTestUtils.setField(telemetryService, "tsCallBackExecutor", tsCallBackExecutor); |
||||
|
|
||||
|
apiUsageState = new ApiUsageState(); |
||||
|
apiUsageState.setDbStorageState(ApiUsageStateValue.ENABLED); |
||||
|
lenient().when(apiUsageStateService.getApiUsageState(tenantId)).thenReturn(apiUsageState); |
||||
|
|
||||
|
lenient().when(partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId)).thenReturn(tpi); |
||||
|
|
||||
|
lenient().when(tsService.save(tenantId, entityId, sampleTelemetry, sampleTtl)).thenReturn(immediateFuture(sampleTelemetry.size())); |
||||
|
lenient().when(tsService.saveWithoutLatest(tenantId, entityId, sampleTelemetry, sampleTtl)).thenReturn(immediateFuture(sampleTelemetry.size())); |
||||
|
lenient().when(tsService.saveLatest(tenantId, entityId, sampleTelemetry)).thenReturn(immediateFuture(listOfNNumbers(sampleTelemetry.size()))); |
||||
|
|
||||
|
// mock no entity views
|
||||
|
lenient().when(tbEntityViewService.findEntityViewsByTenantIdAndEntityIdAsync(tenantId, entityId)).thenReturn(immediateFuture(Collections.emptyList())); |
||||
|
|
||||
|
// send partition change event so currentPartitions set is populated
|
||||
|
telemetryService.onTbApplicationEvent(new PartitionChangeEvent(this, ServiceType.TB_CORE, Map.of(new QueueKey(ServiceType.TB_CORE), Set.of(tpi)))); |
||||
|
} |
||||
|
|
||||
|
@AfterEach |
||||
|
void cleanup() { |
||||
|
wsCallBackExecutor.shutdownNow(); |
||||
|
tsCallBackExecutor.shutdownNow(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldReportStorageDataPointsApiUsageWhenTimeSeriesIsSaved() { |
||||
|
// GIVEN
|
||||
|
var request = TimeseriesSaveRequest.builder() |
||||
|
.tenantId(tenantId) |
||||
|
.customerId(customerId) |
||||
|
.entityId(entityId) |
||||
|
.entries(sampleTelemetry) |
||||
|
.ttl(sampleTtl) |
||||
|
.strategy(new TimeseriesSaveRequest.Strategy(true, false, false)) |
||||
|
.callback(emptyCallback) |
||||
|
.build(); |
||||
|
|
||||
|
// WHEN
|
||||
|
telemetryService.saveTimeseries(request); |
||||
|
|
||||
|
// THEN
|
||||
|
then(apiUsageClient).should().report(tenantId, customerId, ApiUsageRecordKey.STORAGE_DP_COUNT, sampleTelemetry.size()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldNotReportStorageDataPointsApiUsageWhenTimeSeriesIsNotSaved() { |
||||
|
// GIVEN
|
||||
|
var request = TimeseriesSaveRequest.builder() |
||||
|
.tenantId(tenantId) |
||||
|
.customerId(customerId) |
||||
|
.entityId(entityId) |
||||
|
.entries(sampleTelemetry) |
||||
|
.ttl(sampleTtl) |
||||
|
.strategy(TimeseriesSaveRequest.Strategy.LATEST_AND_WS) |
||||
|
.callback(emptyCallback) |
||||
|
.build(); |
||||
|
|
||||
|
// WHEN
|
||||
|
telemetryService.saveTimeseries(request); |
||||
|
|
||||
|
// THEN
|
||||
|
then(apiUsageClient).shouldHaveNoInteractions(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldThrowStorageDisabledWhenTimeSeriesIsSavedAndStorageIsDisabled() { |
||||
|
// GIVEN
|
||||
|
apiUsageState.setDbStorageState(ApiUsageStateValue.DISABLED); |
||||
|
|
||||
|
SettableFuture<Void> future = SettableFuture.create(); |
||||
|
var request = TimeseriesSaveRequest.builder() |
||||
|
.tenantId(tenantId) |
||||
|
.customerId(customerId) |
||||
|
.entityId(entityId) |
||||
|
.entries(sampleTelemetry) |
||||
|
.ttl(sampleTtl) |
||||
|
.strategy(TimeseriesSaveRequest.Strategy.SAVE_ALL) |
||||
|
.future(future) |
||||
|
.build(); |
||||
|
|
||||
|
// WHEN
|
||||
|
telemetryService.saveTimeseries(request); |
||||
|
|
||||
|
// THEN
|
||||
|
assertThat(future).failsWithin(Duration.ofSeconds(5)) |
||||
|
.withThrowableOfType(ExecutionException.class) |
||||
|
.withCauseInstanceOf(RuntimeException.class) |
||||
|
.withMessageContaining("DB storage writes are disabled due to API limits!"); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldNotThrowStorageDisabledWhenTimeSeriesIsNotSavedAndStorageIsDisabled() { |
||||
|
// GIVEN
|
||||
|
apiUsageState.setDbStorageState(ApiUsageStateValue.DISABLED); |
||||
|
|
||||
|
SettableFuture<Void> future = SettableFuture.create(); |
||||
|
var request = TimeseriesSaveRequest.builder() |
||||
|
.tenantId(tenantId) |
||||
|
.customerId(customerId) |
||||
|
.entityId(entityId) |
||||
|
.entries(sampleTelemetry) |
||||
|
.ttl(sampleTtl) |
||||
|
.strategy(TimeseriesSaveRequest.Strategy.LATEST_AND_WS) |
||||
|
.future(future) |
||||
|
.build(); |
||||
|
|
||||
|
// WHEN
|
||||
|
telemetryService.saveTimeseries(request); |
||||
|
|
||||
|
// THEN
|
||||
|
assertThat(future).succeedsWithin(Duration.ofSeconds(5)); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldCopyLatestToEntityViewWhenLatestIsSavedOnMainEntity() { |
||||
|
// GIVEN
|
||||
|
var entityView = new EntityView(new EntityViewId(UUID.randomUUID())); |
||||
|
entityView.setTenantId(tenantId); |
||||
|
entityView.setCustomerId(customerId); |
||||
|
entityView.setEntityId(entityId); |
||||
|
entityView.setKeys(new TelemetryEntityView(sampleTelemetry.stream().map(KvEntry::getKey).toList(), new AttributesEntityView())); |
||||
|
|
||||
|
// mock that there is one entity view
|
||||
|
given(tbEntityViewService.findEntityViewsByTenantIdAndEntityIdAsync(tenantId, entityId)).willReturn(immediateFuture(List.of(entityView))); |
||||
|
// mock that save latest call for entity view is successful
|
||||
|
given(tsService.saveLatest(tenantId, entityView.getId(), sampleTelemetry)).willReturn(immediateFuture(listOfNNumbers(sampleTelemetry.size()))); |
||||
|
// mock TPI for entity view
|
||||
|
given(partitionService.resolve(ServiceType.TB_CORE, tenantId, entityView.getId())).willReturn(tpi); |
||||
|
|
||||
|
var request = TimeseriesSaveRequest.builder() |
||||
|
.tenantId(tenantId) |
||||
|
.customerId(customerId) |
||||
|
.entityId(entityId) |
||||
|
.entries(sampleTelemetry) |
||||
|
.ttl(sampleTtl) |
||||
|
.strategy(new TimeseriesSaveRequest.Strategy(false, true, false)) |
||||
|
.callback(emptyCallback) |
||||
|
.build(); |
||||
|
|
||||
|
// WHEN
|
||||
|
telemetryService.saveTimeseries(request); |
||||
|
|
||||
|
// THEN
|
||||
|
// should save latest to both the main entity and it's entity view
|
||||
|
then(tsService).should().saveLatest(tenantId, entityId, sampleTelemetry); |
||||
|
then(tsService).should().saveLatest(tenantId, entityView.getId(), sampleTelemetry); |
||||
|
then(tsService).shouldHaveNoMoreInteractions(); |
||||
|
|
||||
|
// should send WS update only for entity view (WS update for the main entity is disabled in the save request)
|
||||
|
then(subscriptionManagerService).should().onTimeSeriesUpdate(tenantId, entityView.getId(), sampleTelemetry, TbCallback.EMPTY); |
||||
|
then(subscriptionManagerService).shouldHaveNoMoreInteractions(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldNotCopyLatestToEntityViewWhenLatestIsNotSavedOnMainEntity() { |
||||
|
// GIVEN
|
||||
|
var request = TimeseriesSaveRequest.builder() |
||||
|
.tenantId(tenantId) |
||||
|
.customerId(customerId) |
||||
|
.entityId(entityId) |
||||
|
.entries(sampleTelemetry) |
||||
|
.ttl(sampleTtl) |
||||
|
.strategy(new TimeseriesSaveRequest.Strategy(true, false, false)) |
||||
|
.callback(emptyCallback) |
||||
|
.build(); |
||||
|
|
||||
|
// WHEN
|
||||
|
telemetryService.saveTimeseries(request); |
||||
|
|
||||
|
// THEN
|
||||
|
// should save only time series for the main entity
|
||||
|
then(tsService).should().saveWithoutLatest(tenantId, entityId, sampleTelemetry, sampleTtl); |
||||
|
then(tsService).shouldHaveNoMoreInteractions(); |
||||
|
|
||||
|
// should not send any WS updates
|
||||
|
then(subscriptionManagerService).shouldHaveNoInteractions(); |
||||
|
} |
||||
|
|
||||
|
@ParameterizedTest |
||||
|
@MethodSource("booleanCombinations") |
||||
|
void shouldCallCorrectApiBasedOnBooleanFlagsInTheSaveRequest(boolean saveTimeseries, boolean saveLatest, boolean sendWsUpdate) { |
||||
|
// GIVEN
|
||||
|
var request = TimeseriesSaveRequest.builder() |
||||
|
.tenantId(tenantId) |
||||
|
.customerId(customerId) |
||||
|
.entityId(entityId) |
||||
|
.entries(sampleTelemetry) |
||||
|
.ttl(sampleTtl) |
||||
|
.strategy(new TimeseriesSaveRequest.Strategy(saveTimeseries, saveLatest, sendWsUpdate)) |
||||
|
.callback(emptyCallback) |
||||
|
.build(); |
||||
|
|
||||
|
// WHEN
|
||||
|
telemetryService.saveTimeseries(request); |
||||
|
|
||||
|
// THEN
|
||||
|
if (saveTimeseries && saveLatest) { |
||||
|
then(tsService).should().save(tenantId, entityId, sampleTelemetry, sampleTtl); |
||||
|
} else if (saveLatest) { |
||||
|
then(tsService).should().saveLatest(tenantId, entityId, sampleTelemetry); |
||||
|
} else if (saveTimeseries) { |
||||
|
then(tsService).should().saveWithoutLatest(tenantId, entityId, sampleTelemetry, sampleTtl); |
||||
|
} |
||||
|
then(tsService).shouldHaveNoMoreInteractions(); |
||||
|
|
||||
|
if (sendWsUpdate) { |
||||
|
then(subscriptionManagerService).should().onTimeSeriesUpdate(tenantId, entityId, sampleTelemetry, TbCallback.EMPTY); |
||||
|
} else { |
||||
|
then(subscriptionManagerService).shouldHaveNoInteractions(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private static Stream<Arguments> booleanCombinations() { |
||||
|
return Stream.of( |
||||
|
Arguments.of(true, true, true), |
||||
|
Arguments.of(true, true, false), |
||||
|
Arguments.of(true, false, true), |
||||
|
Arguments.of(true, false, false), |
||||
|
Arguments.of(false, true, true), |
||||
|
Arguments.of(false, true, false), |
||||
|
Arguments.of(false, false, true), |
||||
|
Arguments.of(false, false, false) |
||||
|
); |
||||
|
} |
||||
|
|
||||
|
// used to emulate sequence numbers returned by save latest API
|
||||
|
private static List<Long> listOfNNumbers(int N) { |
||||
|
return LongStream.range(0, N).boxed().toList(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,51 @@ |
|||||
|
/** |
||||
|
* 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.rule.engine.api; |
||||
|
|
||||
|
import org.junit.jupiter.api.Test; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
|
||||
|
class TimeseriesSaveRequestTest { |
||||
|
|
||||
|
@Test |
||||
|
void testDefaultSaveStrategyIsSaveAll() { |
||||
|
var request = TimeseriesSaveRequest.builder().build(); |
||||
|
|
||||
|
assertThat(request.getStrategy()).isEqualTo(TimeseriesSaveRequest.Strategy.SAVE_ALL); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void testSaveAllStrategy() { |
||||
|
assertThat(TimeseriesSaveRequest.Strategy.SAVE_ALL).isEqualTo(new TimeseriesSaveRequest.Strategy(true, true, true)); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void testWsOnlyStrategy() { |
||||
|
assertThat(TimeseriesSaveRequest.Strategy.WS_ONLY).isEqualTo(new TimeseriesSaveRequest.Strategy(false, false, true)); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void testLatestAndWsStrategy() { |
||||
|
assertThat(TimeseriesSaveRequest.Strategy.LATEST_AND_WS).isEqualTo(new TimeseriesSaveRequest.Strategy(false, true, true)); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void testSkipAllStrategy() { |
||||
|
assertThat(TimeseriesSaveRequest.Strategy.SKIP_ALL).isEqualTo(new TimeseriesSaveRequest.Strategy(false, false, false)); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,89 @@ |
|||||
|
/** |
||||
|
* 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.rule.engine.telemetry.strategy; |
||||
|
|
||||
|
import com.fasterxml.jackson.annotation.JsonCreator; |
||||
|
import com.fasterxml.jackson.annotation.JsonProperty; |
||||
|
import com.github.benmanes.caffeine.cache.Caffeine; |
||||
|
import com.github.benmanes.caffeine.cache.LoadingCache; |
||||
|
import com.google.common.collect.Sets; |
||||
|
import com.google.common.primitives.Longs; |
||||
|
|
||||
|
import java.time.Duration; |
||||
|
import java.util.Set; |
||||
|
import java.util.UUID; |
||||
|
|
||||
|
final class DeduplicatePersistenceStrategy implements PersistenceStrategy { |
||||
|
|
||||
|
private static final int MIN_DEDUPLICATION_INTERVAL_SECS = 1; |
||||
|
private static final int MAX_DEDUPLICATION_INTERVAL_SECS = (int) Duration.ofDays(1L).toSeconds(); |
||||
|
|
||||
|
private static final long MIN_INTERVAL_EXPIRY_MILLIS = Duration.ofMinutes(10L).toMillis(); |
||||
|
private static final int INTERVAL_EXPIRY_FACTOR = 10; |
||||
|
private static final long MAX_INTERVAL_EXPIRY_MILLIS = Duration.ofDays(2L).toMillis(); |
||||
|
|
||||
|
private static final int MAX_TOTAL_INTERVALS_DURATION_SECS = (int) Duration.ofDays(2L).toSeconds(); |
||||
|
private static final int MAX_NUMBER_OF_INTERVALS = 100; |
||||
|
|
||||
|
private final long deduplicationIntervalMillis; |
||||
|
private final LoadingCache<Long, Set<UUID>> deduplicationCache; |
||||
|
|
||||
|
@JsonCreator |
||||
|
public DeduplicatePersistenceStrategy(@JsonProperty("deduplicationIntervalSecs") int deduplicationIntervalSecs) { |
||||
|
if (deduplicationIntervalSecs < MIN_DEDUPLICATION_INTERVAL_SECS || deduplicationIntervalSecs > MAX_DEDUPLICATION_INTERVAL_SECS) { |
||||
|
throw new IllegalArgumentException("Deduplication interval must be at least " + MIN_DEDUPLICATION_INTERVAL_SECS + " second(s) " + |
||||
|
"and at most " + MAX_DEDUPLICATION_INTERVAL_SECS + " second(s), was " + deduplicationIntervalSecs + " second(s)"); |
||||
|
} |
||||
|
deduplicationIntervalMillis = Duration.ofSeconds(deduplicationIntervalSecs).toMillis(); |
||||
|
deduplicationCache = Caffeine.newBuilder() |
||||
|
.softValues() |
||||
|
.expireAfterAccess(calculateExpireAfterAccess(deduplicationIntervalSecs)) |
||||
|
.maximumSize(calculateMaxNumberOfDeduplicationIntervals(deduplicationIntervalSecs)) |
||||
|
.build(__ -> Sets.newConcurrentHashSet()); |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* Calculates the expire-after-access duration. By default, we keep each deduplication interval |
||||
|
* alive for 10 “iterations” (interval duration × 10). However, we never let this drop below |
||||
|
* 10 minutes to ensure adequate retention for small intervals, nor exceed 48 hours to prevent |
||||
|
* storing stale data in memory. |
||||
|
*/ |
||||
|
private static Duration calculateExpireAfterAccess(int deduplicationIntervalSecs) { |
||||
|
long desiredExpiryMillis = Duration.ofSeconds(deduplicationIntervalSecs).toMillis() * INTERVAL_EXPIRY_FACTOR; |
||||
|
return Duration.ofMillis(Longs.constrainToRange(desiredExpiryMillis, MIN_INTERVAL_EXPIRY_MILLIS, MAX_INTERVAL_EXPIRY_MILLIS)); |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* Calculates the maximum number of deduplication intervals we will store in the cache. |
||||
|
* We limit retention to two days to avoid stale data and cap it at 100 intervals to manage memory usage. |
||||
|
*/ |
||||
|
private static long calculateMaxNumberOfDeduplicationIntervals(int deduplicationIntervalSecs) { |
||||
|
int numberOfDeduplicationIntervals = MAX_TOTAL_INTERVALS_DURATION_SECS / deduplicationIntervalSecs; |
||||
|
return Math.min(numberOfDeduplicationIntervals, MAX_NUMBER_OF_INTERVALS); |
||||
|
} |
||||
|
|
||||
|
@JsonProperty("deduplicationIntervalSecs") |
||||
|
public long getDeduplicationIntervalSecs() { |
||||
|
return Duration.ofMillis(deduplicationIntervalMillis).toSeconds(); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public boolean shouldPersist(long ts, UUID originatorUuid) { |
||||
|
long intervalNumber = ts / deduplicationIntervalMillis; |
||||
|
return deduplicationCache.get(intervalNumber).add(originatorUuid); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -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.rule.engine.telemetry.strategy; |
||||
|
|
||||
|
import com.fasterxml.jackson.annotation.JsonCreator; |
||||
|
|
||||
|
import java.util.UUID; |
||||
|
|
||||
|
final class OnEveryMessagePersistenceStrategy implements PersistenceStrategy { |
||||
|
|
||||
|
private static final OnEveryMessagePersistenceStrategy INSTANCE = new OnEveryMessagePersistenceStrategy(); |
||||
|
|
||||
|
private OnEveryMessagePersistenceStrategy() {} |
||||
|
|
||||
|
@JsonCreator |
||||
|
public static OnEveryMessagePersistenceStrategy getInstance() { |
||||
|
return INSTANCE; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public boolean shouldPersist(long ts, UUID originatorUuid) { |
||||
|
return true; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,49 @@ |
|||||
|
/** |
||||
|
* 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.rule.engine.telemetry.strategy; |
||||
|
|
||||
|
import com.fasterxml.jackson.annotation.JsonSubTypes; |
||||
|
import com.fasterxml.jackson.annotation.JsonTypeInfo; |
||||
|
|
||||
|
import java.util.UUID; |
||||
|
|
||||
|
@JsonTypeInfo( |
||||
|
use = JsonTypeInfo.Id.NAME, |
||||
|
include = JsonTypeInfo.As.PROPERTY, |
||||
|
property = "type" |
||||
|
) |
||||
|
@JsonSubTypes({ |
||||
|
@JsonSubTypes.Type(value = OnEveryMessagePersistenceStrategy.class, name = "ON_EVERY_MESSAGE"), |
||||
|
@JsonSubTypes.Type(value = DeduplicatePersistenceStrategy.class, name = "DEDUPLICATE"), |
||||
|
@JsonSubTypes.Type(value = SkipPersistenceStrategy.class, name = "SKIP") |
||||
|
}) |
||||
|
public sealed interface PersistenceStrategy permits OnEveryMessagePersistenceStrategy, DeduplicatePersistenceStrategy, SkipPersistenceStrategy { |
||||
|
|
||||
|
static PersistenceStrategy onEveryMessage() { |
||||
|
return OnEveryMessagePersistenceStrategy.getInstance(); |
||||
|
} |
||||
|
|
||||
|
static PersistenceStrategy deduplicate(int deduplicationIntervalSecs) { |
||||
|
return new DeduplicatePersistenceStrategy(deduplicationIntervalSecs); |
||||
|
} |
||||
|
|
||||
|
static PersistenceStrategy skip() { |
||||
|
return SkipPersistenceStrategy.getInstance(); |
||||
|
} |
||||
|
|
||||
|
boolean shouldPersist(long ts, UUID originatorUuid); |
||||
|
|
||||
|
} |
||||
@ -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.rule.engine.telemetry.strategy; |
||||
|
|
||||
|
import com.fasterxml.jackson.annotation.JsonCreator; |
||||
|
|
||||
|
import java.util.UUID; |
||||
|
|
||||
|
final class SkipPersistenceStrategy implements PersistenceStrategy { |
||||
|
|
||||
|
private static final SkipPersistenceStrategy INSTANCE = new SkipPersistenceStrategy(); |
||||
|
|
||||
|
private SkipPersistenceStrategy() {} |
||||
|
|
||||
|
@JsonCreator |
||||
|
public static SkipPersistenceStrategy getInstance() { |
||||
|
return INSTANCE; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public boolean shouldPersist(long ts, UUID originatorUuid) { |
||||
|
return false; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,260 @@ |
|||||
|
/** |
||||
|
* 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.rule.engine.telemetry.strategy; |
||||
|
|
||||
|
import com.github.benmanes.caffeine.cache.LoadingCache; |
||||
|
import com.github.benmanes.caffeine.cache.Policy; |
||||
|
import org.junit.jupiter.api.BeforeEach; |
||||
|
import org.junit.jupiter.api.Test; |
||||
|
import org.springframework.test.util.ReflectionTestUtils; |
||||
|
|
||||
|
import java.time.Duration; |
||||
|
import java.util.Set; |
||||
|
import java.util.UUID; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
import static org.assertj.core.api.Assertions.assertThatThrownBy; |
||||
|
|
||||
|
class DeduplicatePersistenceStrategyTest { |
||||
|
|
||||
|
final int deduplicationIntervalSecs = 10; |
||||
|
|
||||
|
DeduplicatePersistenceStrategy strategy; |
||||
|
|
||||
|
@BeforeEach |
||||
|
void setup() { |
||||
|
strategy = new DeduplicatePersistenceStrategy(deduplicationIntervalSecs); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldThrowWhenDeduplicationIntervalIsLessThanOneSecond() { |
||||
|
assertThatThrownBy(() -> new DeduplicatePersistenceStrategy(0)) |
||||
|
.isInstanceOf(IllegalArgumentException.class) |
||||
|
.hasMessageContaining("Deduplication interval must be at least 1 second(s) and at most 86400 second(s), was 0 second(s)"); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldThrowWhenDeduplicationIntervalIsMoreThan24Hours() { |
||||
|
assertThatThrownBy(() -> new DeduplicatePersistenceStrategy(86401)) |
||||
|
.isInstanceOf(IllegalArgumentException.class) |
||||
|
.hasMessageContaining("Deduplication interval must be at least 1 second(s) and at most 86400 second(s), was 86401 second(s)"); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldUseAtLeastTenMinutesForExpireAfterAccess() { |
||||
|
// GIVEN
|
||||
|
int deduplicationIntervalSecs = 1; // min deduplication interval duration
|
||||
|
|
||||
|
// WHEN
|
||||
|
strategy = new DeduplicatePersistenceStrategy(deduplicationIntervalSecs); |
||||
|
|
||||
|
// THEN
|
||||
|
var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache"); |
||||
|
|
||||
|
assertThat(deduplicationCache.policy().expireAfterAccess()) |
||||
|
.isPresent() |
||||
|
.map(Policy.FixedExpiration::getExpiresAfter) |
||||
|
.hasValue(Duration.ofMinutes(10L)); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldCalculateExpireAfterAccessAsIntervalDurationMultipliedByTen() { |
||||
|
// GIVEN
|
||||
|
int deduplicationIntervalSecs = (int) Duration.ofHours(1L).toSeconds(); // max deduplication interval duration
|
||||
|
|
||||
|
// WHEN
|
||||
|
strategy = new DeduplicatePersistenceStrategy(deduplicationIntervalSecs); |
||||
|
|
||||
|
// THEN
|
||||
|
var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache"); |
||||
|
|
||||
|
assertThat(deduplicationCache.policy().expireAfterAccess()) |
||||
|
.isPresent() |
||||
|
.map(Policy.FixedExpiration::getExpiresAfter) |
||||
|
.hasValue(Duration.ofHours(10L)); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldUseAtMostTwoDaysForExpireAfterAccess() { |
||||
|
// GIVEN
|
||||
|
int deduplicationIntervalSecs = (int) Duration.ofDays(1L).toSeconds(); // max deduplication interval duration
|
||||
|
|
||||
|
// WHEN
|
||||
|
strategy = new DeduplicatePersistenceStrategy(deduplicationIntervalSecs); |
||||
|
|
||||
|
// THEN
|
||||
|
var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache"); |
||||
|
|
||||
|
assertThat(deduplicationCache.policy().expireAfterAccess()) |
||||
|
.isPresent() |
||||
|
.map(Policy.FixedExpiration::getExpiresAfter) |
||||
|
.hasValue(Duration.ofDays(2L)); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldNotAllowMoreThan100DeduplicationIntervals() { |
||||
|
// GIVEN
|
||||
|
int deduplicationIntervalSecs = 1; // min deduplication interval duration
|
||||
|
|
||||
|
// WHEN
|
||||
|
strategy = new DeduplicatePersistenceStrategy(deduplicationIntervalSecs); |
||||
|
|
||||
|
// THEN
|
||||
|
var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache"); |
||||
|
|
||||
|
assertThat(deduplicationCache.policy().eviction()) |
||||
|
.isPresent() |
||||
|
.map(Policy.Eviction::getMaximum) |
||||
|
.hasValue(100L); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldCalculateMaxIntervalsAsTwoDaysDividedByIntervalDuration() { |
||||
|
// GIVEN
|
||||
|
int deduplicationIntervalSecs = (int) Duration.ofHours(1L).toSeconds(); |
||||
|
|
||||
|
// WHEN
|
||||
|
strategy = new DeduplicatePersistenceStrategy(deduplicationIntervalSecs); |
||||
|
|
||||
|
// THEN
|
||||
|
var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache"); |
||||
|
|
||||
|
assertThat(deduplicationCache.policy().eviction()) |
||||
|
.isPresent() |
||||
|
.map(Policy.Eviction::getMaximum) |
||||
|
.hasValue(48L); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldKeepAtLeastTwoDeduplicationIntervals() { |
||||
|
// GIVEN
|
||||
|
int deduplicationIntervalSecs = (int) Duration.ofDays(1L).toSeconds(); // max deduplication interval duration
|
||||
|
|
||||
|
// WHEN
|
||||
|
strategy = new DeduplicatePersistenceStrategy(deduplicationIntervalSecs); |
||||
|
|
||||
|
// THEN
|
||||
|
var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache"); |
||||
|
|
||||
|
assertThat(deduplicationCache.policy().eviction()) |
||||
|
.isPresent() |
||||
|
.map(Policy.Eviction::getMaximum) |
||||
|
.hasValue(2L); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldReturnTrueForFirstCallInInterval() { |
||||
|
long ts = 1_000_000L; |
||||
|
UUID originator = UUID.randomUUID(); |
||||
|
|
||||
|
assertThat(strategy.shouldPersist(ts, originator)).isTrue(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldReturnFalseForSubsequentCallsInInterval() { |
||||
|
long baseTs = 1_000_000L; |
||||
|
UUID originator = UUID.randomUUID(); |
||||
|
|
||||
|
// Initial call should return true
|
||||
|
assertThat(strategy.shouldPersist(baseTs, originator)).isTrue(); |
||||
|
|
||||
|
// Subsequent call within the same interval should return false for the same originator
|
||||
|
long withinSameIntervalTs = baseTs + 1000L; |
||||
|
assertThat(strategy.shouldPersist(withinSameIntervalTs, originator)).isFalse(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldHandleMultipleOriginatorsIndependently() { |
||||
|
long baseTs = 1_000_000L; |
||||
|
UUID originator1 = UUID.randomUUID(); |
||||
|
UUID originator2 = UUID.randomUUID(); |
||||
|
|
||||
|
// First call for different originators in the same interval should return true independently
|
||||
|
assertThat(strategy.shouldPersist(baseTs, originator1)).isTrue(); |
||||
|
assertThat(strategy.shouldPersist(baseTs, originator2)).isTrue(); |
||||
|
|
||||
|
// Subsequent calls for the same originators within the same interval should return false
|
||||
|
assertThat(strategy.shouldPersist(baseTs + 500L, originator1)).isFalse(); |
||||
|
assertThat(strategy.shouldPersist(baseTs + 500L, originator2)).isFalse(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldHandleEdgeCaseTimestamps() { |
||||
|
long minTs = Long.MIN_VALUE; |
||||
|
long maxTs = Long.MAX_VALUE; |
||||
|
UUID originator = UUID.randomUUID(); |
||||
|
|
||||
|
assertThat(strategy.shouldPersist(minTs, originator)).isTrue(); |
||||
|
assertThat(strategy.shouldPersist(minTs + 1L, originator)).isFalse(); |
||||
|
|
||||
|
assertThat(strategy.shouldPersist(maxTs, originator)).isTrue(); |
||||
|
assertThat(strategy.shouldPersist(maxTs - 1L, originator)).isFalse(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldResetDeduplicationAtIntervalBoundaries() { |
||||
|
UUID originator = UUID.randomUUID(); |
||||
|
|
||||
|
// check 1st interval
|
||||
|
long firstIntervalStart = 0L; |
||||
|
long firstIntervalEnd = firstIntervalStart + Duration.ofSeconds(deduplicationIntervalSecs).toMillis() - 1L; |
||||
|
long firstIntervalMiddle = calculateMiddle(firstIntervalStart, firstIntervalEnd); |
||||
|
|
||||
|
assertThat(strategy.shouldPersist(firstIntervalStart, originator)).isTrue(); |
||||
|
assertThat(strategy.shouldPersist(firstIntervalStart + 1, originator)).isFalse(); |
||||
|
assertThat(strategy.shouldPersist(firstIntervalMiddle, originator)).isFalse(); |
||||
|
assertThat(strategy.shouldPersist(firstIntervalEnd - 1, originator)).isFalse(); |
||||
|
assertThat(strategy.shouldPersist(firstIntervalEnd, originator)).isFalse(); |
||||
|
|
||||
|
// check 2nd interval
|
||||
|
long secondIntervalStart = firstIntervalEnd + 1L; |
||||
|
long secondIntervalEnd = secondIntervalStart + Duration.ofSeconds(deduplicationIntervalSecs).toMillis() - 1L; |
||||
|
long secondIntervalMiddle = calculateMiddle(secondIntervalStart, secondIntervalEnd); |
||||
|
|
||||
|
assertThat(strategy.shouldPersist(secondIntervalStart, originator)).isTrue(); |
||||
|
assertThat(strategy.shouldPersist(secondIntervalStart + 1, originator)).isFalse(); |
||||
|
assertThat(strategy.shouldPersist(secondIntervalMiddle, originator)).isFalse(); |
||||
|
assertThat(strategy.shouldPersist(secondIntervalEnd - 1, originator)).isFalse(); |
||||
|
assertThat(strategy.shouldPersist(secondIntervalEnd, originator)).isFalse(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void shouldHandleMultipleOriginatorsOverMultipleIntervals() { |
||||
|
UUID originator1 = UUID.randomUUID(); |
||||
|
UUID originator2 = UUID.randomUUID(); |
||||
|
long baseTs = 0L; |
||||
|
|
||||
|
// First interval for both originators
|
||||
|
assertThat(strategy.shouldPersist(baseTs, originator1)).isTrue(); |
||||
|
assertThat(strategy.shouldPersist(baseTs, originator2)).isTrue(); |
||||
|
|
||||
|
// Move to the next interval
|
||||
|
long nextIntervalTs = baseTs + Duration.ofSeconds(10).toMillis(); |
||||
|
|
||||
|
// Each originator should be allowed again in the new interval
|
||||
|
assertThat(strategy.shouldPersist(nextIntervalTs, originator1)).isTrue(); |
||||
|
assertThat(strategy.shouldPersist(nextIntervalTs, originator2)).isTrue(); |
||||
|
|
||||
|
// Subsequent calls in the same new interval should return false
|
||||
|
assertThat(strategy.shouldPersist(nextIntervalTs + 500L, originator1)).isFalse(); |
||||
|
assertThat(strategy.shouldPersist(nextIntervalTs + 500L, originator2)).isFalse(); |
||||
|
} |
||||
|
|
||||
|
private static long calculateMiddle(long start, long end) { |
||||
|
return start + (end - start) / 2; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,48 @@ |
|||||
|
/** |
||||
|
* 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.rule.engine.telemetry.strategy; |
||||
|
|
||||
|
import org.junit.jupiter.params.ParameterizedTest; |
||||
|
import org.junit.jupiter.params.provider.Arguments; |
||||
|
import org.junit.jupiter.params.provider.MethodSource; |
||||
|
|
||||
|
import java.util.UUID; |
||||
|
import java.util.stream.Stream; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
|
||||
|
class OnEveryMessagePersistenceStrategyTest { |
||||
|
|
||||
|
@ParameterizedTest |
||||
|
@MethodSource("edgeCaseProvider") |
||||
|
void shouldAlwaysReturnTrueForAnyInput(long timestamp, UUID originator) { |
||||
|
var onEveryMessage = OnEveryMessagePersistenceStrategy.getInstance(); |
||||
|
assertThat(onEveryMessage.shouldPersist(timestamp, originator)).isTrue(); |
||||
|
} |
||||
|
|
||||
|
private static Stream<Arguments> edgeCaseProvider() { |
||||
|
return Stream.of( |
||||
|
Arguments.of(Long.MIN_VALUE, new UUID(0L, 0L)), |
||||
|
Arguments.of(Long.MAX_VALUE, new UUID(Long.MAX_VALUE, Long.MAX_VALUE)), |
||||
|
Arguments.of(0L, new UUID(0L, 0L)), |
||||
|
Arguments.of(-1L, new UUID(-1L, -1L)), |
||||
|
Arguments.of(1L, new UUID(1L, 1L)), |
||||
|
Arguments.of(42L, UUID.randomUUID()), |
||||
|
Arguments.of(1000L, null) |
||||
|
); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,55 @@ |
|||||
|
/** |
||||
|
* 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.rule.engine.telemetry.strategy; |
||||
|
|
||||
|
import org.junit.jupiter.api.Test; |
||||
|
import org.springframework.test.util.ReflectionTestUtils; |
||||
|
|
||||
|
import java.time.Duration; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
|
||||
|
class PersistenceStrategyTest { |
||||
|
|
||||
|
@Test |
||||
|
void testOnEveryMessageReturnsCorrectInstance() { |
||||
|
PersistenceStrategy strategy = PersistenceStrategy.onEveryMessage(); |
||||
|
assertThat(strategy) |
||||
|
.isNotNull() |
||||
|
.isInstanceOf(OnEveryMessagePersistenceStrategy.class); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void testDeduplicateReturnsCorrectInstance() { |
||||
|
int validDeduplicationIntervalSecs = 5; |
||||
|
PersistenceStrategy strategy = PersistenceStrategy.deduplicate(validDeduplicationIntervalSecs); |
||||
|
assertThat(strategy) |
||||
|
.isNotNull() |
||||
|
.isInstanceOf(DeduplicatePersistenceStrategy.class); |
||||
|
|
||||
|
long actualDeduplicationIntervalMillis = (long) ReflectionTestUtils.getField(strategy, "deduplicationIntervalMillis"); |
||||
|
assertThat(actualDeduplicationIntervalMillis).isEqualTo(Duration.ofSeconds(validDeduplicationIntervalSecs).toMillis()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void testSkipReturnsCorrectInstance() { |
||||
|
PersistenceStrategy strategy = PersistenceStrategy.skip(); |
||||
|
assertThat(strategy) |
||||
|
.isNotNull() |
||||
|
.isInstanceOf(SkipPersistenceStrategy.class); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,48 @@ |
|||||
|
/** |
||||
|
* 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.rule.engine.telemetry.strategy; |
||||
|
|
||||
|
import org.junit.jupiter.params.ParameterizedTest; |
||||
|
import org.junit.jupiter.params.provider.Arguments; |
||||
|
import org.junit.jupiter.params.provider.MethodSource; |
||||
|
|
||||
|
import java.util.UUID; |
||||
|
import java.util.stream.Stream; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
|
||||
|
class SkipPersistenceStrategyTest { |
||||
|
|
||||
|
@ParameterizedTest |
||||
|
@MethodSource("edgeCaseProvider") |
||||
|
void shouldAlwaysReturnFalseForAnyInput(long timestamp, UUID originator) { |
||||
|
var skipStrategy = SkipPersistenceStrategy.getInstance(); |
||||
|
assertThat(skipStrategy.shouldPersist(timestamp, originator)).isFalse(); |
||||
|
} |
||||
|
|
||||
|
private static Stream<Arguments> edgeCaseProvider() { |
||||
|
return Stream.of( |
||||
|
Arguments.of(Long.MIN_VALUE, new UUID(0L, 0L)), |
||||
|
Arguments.of(Long.MAX_VALUE, new UUID(Long.MAX_VALUE, Long.MAX_VALUE)), |
||||
|
Arguments.of(0L, new UUID(0L, 0L)), |
||||
|
Arguments.of(-1L, new UUID(-1L, -1L)), |
||||
|
Arguments.of(1L, new UUID(1L, 1L)), |
||||
|
Arguments.of(42L, UUID.randomUUID()), |
||||
|
Arguments.of(1000L, null) |
||||
|
); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,40 @@ |
|||||
|
<!-- |
||||
|
|
||||
|
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. |
||||
|
|
||||
|
--> |
||||
|
<section [formGroup]="persistenceSettingRowForm" class="tb-form-panel stroked no-gap no-padding-bottom"> |
||||
|
<div class="tb-form-panel-title mb-4">{{ title }}</div> |
||||
|
<mat-form-field> |
||||
|
<mat-label translate>rule-node-config.save-time-series.strategy</mat-label> |
||||
|
<mat-select formControlName="type"> |
||||
|
@for (strategy of persistenceStrategies; track strategy) { |
||||
|
<mat-option [value]="strategy">{{ PersistenceTypeTranslationMap.get(strategy) | translate }}</mat-option> |
||||
|
} |
||||
|
</mat-select> |
||||
|
</mat-form-field> |
||||
|
@if(persistenceSettingRowForm.get('type').value === PersistenceType.DEDUPLICATE) { |
||||
|
<tb-time-unit-input |
||||
|
required |
||||
|
labelText="{{ 'rule-node-config.save-time-series.deduplication-interval' | translate }}" |
||||
|
requiredText="{{ 'rule-node-config.save-time-series.deduplication-interval-required' | translate }}" |
||||
|
minErrorText="{{ 'rule-node-config.save-time-series.deduplication-interval-min-max-range' | translate }}" |
||||
|
maxErrorText="{{ 'rule-node-config.save-time-series.deduplication-interval-min-max-range' | translate }}" |
||||
|
[minTime]="1" |
||||
|
[maxTime]="maxDeduplicateTime" |
||||
|
formControlName="deduplicationIntervalSecs"> |
||||
|
</tb-time-unit-input> |
||||
|
} |
||||
|
</section> |
||||
@ -0,0 +1,114 @@ |
|||||
|
///
|
||||
|
/// 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.
|
||||
|
///
|
||||
|
|
||||
|
import { Component, forwardRef, Input } from '@angular/core'; |
||||
|
import { |
||||
|
ControlValueAccessor, |
||||
|
FormBuilder, |
||||
|
NG_VALIDATORS, |
||||
|
NG_VALUE_ACCESSOR, |
||||
|
ValidationErrors, |
||||
|
Validator |
||||
|
} from '@angular/forms'; |
||||
|
import { |
||||
|
AdvancedPersistenceConfig, |
||||
|
defaultAdvancedPersistenceConfig, |
||||
|
maxDeduplicateTimeSecs, |
||||
|
PersistenceType, |
||||
|
PersistenceTypeTranslationMap |
||||
|
} from '@home/components/rule-node/action/timeseries-config.models'; |
||||
|
import { isDefinedAndNotNull } from '@core/utils'; |
||||
|
import { takeUntilDestroyed } from '@angular/core/rxjs-interop'; |
||||
|
|
||||
|
@Component({ |
||||
|
selector: 'tb-advanced-persistence-setting-row', |
||||
|
templateUrl: './advanced-persistence-setting-row.component.html', |
||||
|
providers: [{ |
||||
|
provide: NG_VALUE_ACCESSOR, |
||||
|
useExisting: forwardRef(() => AdvancedPersistenceSettingRowComponent), |
||||
|
multi: true |
||||
|
},{ |
||||
|
provide: NG_VALIDATORS, |
||||
|
useExisting: forwardRef(() => AdvancedPersistenceSettingRowComponent), |
||||
|
multi: true |
||||
|
}] |
||||
|
}) |
||||
|
export class AdvancedPersistenceSettingRowComponent implements ControlValueAccessor, Validator { |
||||
|
|
||||
|
@Input() |
||||
|
title: string; |
||||
|
|
||||
|
persistenceSettingRowForm = this.fb.group({ |
||||
|
type: [defaultAdvancedPersistenceConfig.type], |
||||
|
deduplicationIntervalSecs: [{value: 60, disabled: true}] |
||||
|
}); |
||||
|
|
||||
|
PersistenceType = PersistenceType; |
||||
|
persistenceStrategies = [PersistenceType.ON_EVERY_MESSAGE, PersistenceType.DEDUPLICATE, PersistenceType.SKIP]; |
||||
|
PersistenceTypeTranslationMap = PersistenceTypeTranslationMap; |
||||
|
|
||||
|
maxDeduplicateTime = maxDeduplicateTimeSecs; |
||||
|
|
||||
|
private propagateChange: (value: any) => void = () => {}; |
||||
|
|
||||
|
constructor(private fb: FormBuilder) { |
||||
|
this.persistenceSettingRowForm.get('type').valueChanges.pipe( |
||||
|
takeUntilDestroyed() |
||||
|
).subscribe(() => this.updatedValidation()); |
||||
|
|
||||
|
this.persistenceSettingRowForm.valueChanges.pipe( |
||||
|
takeUntilDestroyed() |
||||
|
).subscribe((value) => this.propagateChange(value)); |
||||
|
} |
||||
|
|
||||
|
registerOnChange(fn: any) { |
||||
|
this.propagateChange = fn; |
||||
|
} |
||||
|
|
||||
|
registerOnTouched(_fn: any) { |
||||
|
} |
||||
|
|
||||
|
setDisabledState(isDisabled: boolean) { |
||||
|
if (isDisabled) { |
||||
|
this.persistenceSettingRowForm.disable({emitEvent: false}); |
||||
|
} else { |
||||
|
this.persistenceSettingRowForm.enable({emitEvent: false}); |
||||
|
this.updatedValidation(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
validate(): ValidationErrors | null { |
||||
|
return this.persistenceSettingRowForm.valid ? null : { |
||||
|
persistenceSettingRow: false |
||||
|
}; |
||||
|
} |
||||
|
|
||||
|
writeValue(value: AdvancedPersistenceConfig) { |
||||
|
if (isDefinedAndNotNull(value)) { |
||||
|
this.persistenceSettingRowForm.patchValue(value, {emitEvent: false}); |
||||
|
} else { |
||||
|
this.persistenceSettingRowForm.patchValue(defaultAdvancedPersistenceConfig); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private updatedValidation() { |
||||
|
if (this.persistenceSettingRowForm.get('type').value === PersistenceType.DEDUPLICATE) { |
||||
|
this.persistenceSettingRowForm.get('deduplicationIntervalSecs').enable({emitEvent: false}); |
||||
|
} else { |
||||
|
this.persistenceSettingRowForm.get('deduplicationIntervalSecs').disable({emitEvent: false}) |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,31 @@ |
|||||
|
<!-- |
||||
|
|
||||
|
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. |
||||
|
|
||||
|
--> |
||||
|
<section [formGroup]="persistenceForm" class="tb-form-panel no-border no-padding"> |
||||
|
<tb-advanced-persistence-setting-row |
||||
|
formControlName="timeseries" |
||||
|
title="{{ 'rule-node-config.save-time-series.time-series' | translate }}" |
||||
|
></tb-advanced-persistence-setting-row> |
||||
|
<tb-advanced-persistence-setting-row |
||||
|
formControlName="latest" |
||||
|
title="{{ 'rule-node-config.save-time-series.latest' | translate }}" |
||||
|
></tb-advanced-persistence-setting-row> |
||||
|
<tb-advanced-persistence-setting-row |
||||
|
formControlName="webSockets" |
||||
|
title="{{ 'rule-node-config.save-time-series.web-sockets' | translate }}" |
||||
|
></tb-advanced-persistence-setting-row> |
||||
|
</section> |
||||
@ -0,0 +1,83 @@ |
|||||
|
///
|
||||
|
/// 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.
|
||||
|
///
|
||||
|
|
||||
|
import { |
||||
|
ControlValueAccessor, |
||||
|
FormBuilder, |
||||
|
NG_VALIDATORS, |
||||
|
NG_VALUE_ACCESSOR, |
||||
|
ValidationErrors, |
||||
|
Validator |
||||
|
} from '@angular/forms'; |
||||
|
import { Component, forwardRef } from '@angular/core'; |
||||
|
import { takeUntilDestroyed } from '@angular/core/rxjs-interop'; |
||||
|
import { AdvancedPersistenceStrategy } from '@home/components/rule-node/action/timeseries-config.models'; |
||||
|
|
||||
|
@Component({ |
||||
|
selector: 'tb-advanced-persistence-settings', |
||||
|
templateUrl: './advanced-persistence-setting.component.html', |
||||
|
providers: [{ |
||||
|
provide: NG_VALUE_ACCESSOR, |
||||
|
useExisting: forwardRef(() => AdvancedPersistenceSettingComponent), |
||||
|
multi: true |
||||
|
},{ |
||||
|
provide: NG_VALIDATORS, |
||||
|
useExisting: forwardRef(() => AdvancedPersistenceSettingComponent), |
||||
|
multi: true |
||||
|
}] |
||||
|
}) |
||||
|
export class AdvancedPersistenceSettingComponent implements ControlValueAccessor, Validator { |
||||
|
|
||||
|
persistenceForm = this.fb.group({ |
||||
|
timeseries: [null], |
||||
|
latest: [null], |
||||
|
webSockets: [null] |
||||
|
}); |
||||
|
|
||||
|
private propagateChange: (value: any) => void = () => {}; |
||||
|
|
||||
|
constructor(private fb: FormBuilder) { |
||||
|
this.persistenceForm.valueChanges.pipe( |
||||
|
takeUntilDestroyed() |
||||
|
).subscribe(value => this.propagateChange(value)); |
||||
|
} |
||||
|
|
||||
|
registerOnChange(fn: any) { |
||||
|
this.propagateChange = fn; |
||||
|
} |
||||
|
|
||||
|
registerOnTouched(_fn: any) { |
||||
|
} |
||||
|
|
||||
|
setDisabledState(isDisabled: boolean) { |
||||
|
if (isDisabled) { |
||||
|
this.persistenceForm.disable({emitEvent: false}); |
||||
|
} else { |
||||
|
this.persistenceForm.enable({emitEvent: false}); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
validate(): ValidationErrors | null { |
||||
|
return this.persistenceForm.valid ? null : { |
||||
|
persistenceForm: false |
||||
|
}; |
||||
|
} |
||||
|
|
||||
|
writeValue(value: AdvancedPersistenceStrategy) { |
||||
|
this.persistenceForm.patchValue(value, {emitEvent: false}); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,78 @@ |
|||||
|
///
|
||||
|
/// 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.
|
||||
|
///
|
||||
|
|
||||
|
import { DAY, SECOND } from '@shared/models/time/time.models'; |
||||
|
|
||||
|
export const maxDeduplicateTimeSecs = DAY / SECOND; |
||||
|
|
||||
|
export interface TimeseriesNodeConfiguration { |
||||
|
persistenceSettings: PersistenceSettings; |
||||
|
defaultTTL: number; |
||||
|
useServerTs: boolean; |
||||
|
} |
||||
|
|
||||
|
export interface TimeseriesNodeConfigurationForm extends Omit<TimeseriesNodeConfiguration, 'persistenceSettings'> { |
||||
|
persistenceSettings: PersistenceSettingsForm |
||||
|
} |
||||
|
|
||||
|
export type PersistenceSettings = BasicPersistenceSettings & Partial<DeduplicatePersistenceStrategy> & Partial<AdvancedPersistenceStrategy>; |
||||
|
|
||||
|
export type PersistenceSettingsForm = Omit<PersistenceSettings, keyof AdvancedPersistenceStrategy> & { |
||||
|
isAdvanced: boolean; |
||||
|
advanced?: Partial<AdvancedPersistenceStrategy>; |
||||
|
type: PersistenceType; |
||||
|
}; |
||||
|
|
||||
|
export enum PersistenceType { |
||||
|
ON_EVERY_MESSAGE = 'ON_EVERY_MESSAGE', |
||||
|
DEDUPLICATE = 'DEDUPLICATE', |
||||
|
WEBSOCKETS_ONLY = 'WEBSOCKETS_ONLY', |
||||
|
ADVANCED = 'ADVANCED', |
||||
|
SKIP = 'SKIP' |
||||
|
} |
||||
|
|
||||
|
export const PersistenceTypeTranslationMap = new Map<PersistenceType, string>([ |
||||
|
[PersistenceType.ON_EVERY_MESSAGE, 'rule-node-config.save-time-series.strategy-type.every-message'], |
||||
|
[PersistenceType.DEDUPLICATE, 'rule-node-config.save-time-series.strategy-type.deduplicate'], |
||||
|
[PersistenceType.WEBSOCKETS_ONLY, 'rule-node-config.save-time-series.strategy-type.web-sockets-only'], |
||||
|
[PersistenceType.SKIP, 'rule-node-config.save-time-series.strategy-type.skip'], |
||||
|
]) |
||||
|
|
||||
|
export interface BasicPersistenceSettings { |
||||
|
type: PersistenceType; |
||||
|
} |
||||
|
|
||||
|
export interface DeduplicatePersistenceStrategy extends BasicPersistenceSettings{ |
||||
|
deduplicationIntervalSecs: number; |
||||
|
} |
||||
|
|
||||
|
export interface AdvancedPersistenceStrategy extends BasicPersistenceSettings{ |
||||
|
timeseries: AdvancedPersistenceConfig; |
||||
|
latest: AdvancedPersistenceConfig; |
||||
|
webSockets: AdvancedPersistenceConfig; |
||||
|
} |
||||
|
|
||||
|
export type AdvancedPersistenceConfig = WithOptional<DeduplicatePersistenceStrategy, 'deduplicationIntervalSecs'>; |
||||
|
|
||||
|
export const defaultAdvancedPersistenceConfig: AdvancedPersistenceConfig = { |
||||
|
type: PersistenceType.ON_EVERY_MESSAGE |
||||
|
} |
||||
|
|
||||
|
export const defaultAdvancedPersistenceStrategy: Omit<AdvancedPersistenceStrategy, 'type'> = { |
||||
|
timeseries: defaultAdvancedPersistenceConfig, |
||||
|
latest: defaultAdvancedPersistenceConfig, |
||||
|
webSockets: defaultAdvancedPersistenceConfig, |
||||
|
} |
||||
@ -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. |
||||
|
|
||||
|
--> |
||||
|
<section [formGroup]="timeInputForm" class="flex gap-4"> |
||||
|
<mat-form-field class="max-w-66% flex-full" subscriptSizing="dynamic"> |
||||
|
<mat-label *ngIf="labelText">{{ labelText }}</mat-label> |
||||
|
<input type="number" min="0" step="1" matInput formControlName="time"> |
||||
|
<div matSuffix> |
||||
|
<ng-content select="[matSuffix]"></ng-content> |
||||
|
</div> |
||||
|
<mat-hint *ngIf="subscriptSizing === 'fixed'"></mat-hint> |
||||
|
<mat-error *ngIf="timeInputForm.get('time').hasError('required') && requiredText"> |
||||
|
{{ requiredText }} |
||||
|
</mat-error> |
||||
|
<mat-error *ngIf="timeInputForm.get('time').hasError('min') && minErrorText"> |
||||
|
{{ minErrorText }} |
||||
|
</mat-error> |
||||
|
<mat-error *ngIf="timeInputForm.get('time').hasError('max') && maxErrorText"> |
||||
|
{{ maxErrorText }} |
||||
|
</mat-error> |
||||
|
</mat-form-field> |
||||
|
<mat-form-field class="h-fit max-w-33% flex-full" [subscriptSizing]="subscriptSizing"> |
||||
|
<mat-label translate>rule-node-config.units</mat-label> |
||||
|
<mat-select formControlName="timeUnit"> |
||||
|
@for (timeUnit of timeUnits; track timeUnit) { |
||||
|
<mat-option [value]="timeUnit">{{ timeUnitTranslations.get(timeUnit) | translate }}</mat-option> |
||||
|
} |
||||
|
</mat-select> |
||||
|
</mat-form-field> |
||||
|
</section> |
||||
@ -0,0 +1,199 @@ |
|||||
|
///
|
||||
|
/// 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.
|
||||
|
///
|
||||
|
|
||||
|
import { Component, DestroyRef, forwardRef, Input, OnInit } from '@angular/core'; |
||||
|
import { |
||||
|
AbstractControl, |
||||
|
ControlValueAccessor, |
||||
|
FormBuilder, |
||||
|
NG_VALIDATORS, |
||||
|
NG_VALUE_ACCESSOR, |
||||
|
ValidationErrors, |
||||
|
Validator, |
||||
|
Validators |
||||
|
} from '@angular/forms'; |
||||
|
import { TimeUnit, timeUnitTranslations } from '../rule-node-config.models'; |
||||
|
import { isDefinedAndNotNull, isNumeric } from '@core/utils'; |
||||
|
import { takeUntilDestroyed } from '@angular/core/rxjs-interop'; |
||||
|
import { coerceBoolean, coerceNumber } from '@shared/decorators/coercion'; |
||||
|
import { DAY, HOUR, MINUTE, SECOND } from '@shared/models/time/time.models'; |
||||
|
import { SubscriptSizing } from '@angular/material/form-field'; |
||||
|
|
||||
|
interface TimeUnitInputModel { |
||||
|
time: number; |
||||
|
timeUnit: TimeUnit |
||||
|
} |
||||
|
|
||||
|
@Component({ |
||||
|
selector: 'tb-time-unit-input', |
||||
|
templateUrl: './time-unit-input.component.html', |
||||
|
providers: [{ |
||||
|
provide: NG_VALUE_ACCESSOR, |
||||
|
useExisting: forwardRef(() => TimeUnitInputComponent), |
||||
|
multi: true |
||||
|
},{ |
||||
|
provide: NG_VALIDATORS, |
||||
|
useExisting: forwardRef(() => TimeUnitInputComponent), |
||||
|
multi: true |
||||
|
}] |
||||
|
}) |
||||
|
export class TimeUnitInputComponent implements ControlValueAccessor, Validator, OnInit { |
||||
|
|
||||
|
@Input() |
||||
|
labelText: string; |
||||
|
|
||||
|
@Input() |
||||
|
@coerceBoolean() |
||||
|
required: boolean; |
||||
|
|
||||
|
@Input() |
||||
|
requiredText: string; |
||||
|
|
||||
|
@Input() |
||||
|
@coerceNumber() |
||||
|
minTime = 0; |
||||
|
|
||||
|
@Input() |
||||
|
minErrorText: string; |
||||
|
|
||||
|
@Input() |
||||
|
@coerceNumber() |
||||
|
maxTime: number; |
||||
|
|
||||
|
@Input() |
||||
|
maxErrorText: string; |
||||
|
|
||||
|
@Input() |
||||
|
subscriptSizing: SubscriptSizing = 'fixed'; |
||||
|
|
||||
|
timeUnits = Object.values(TimeUnit).filter(item => item !== TimeUnit.MILLISECONDS) as TimeUnit[]; |
||||
|
|
||||
|
timeUnitTranslations = timeUnitTranslations; |
||||
|
|
||||
|
timeInputForm = this.fb.group({ |
||||
|
time: [0], |
||||
|
timeUnit: [TimeUnit.SECONDS] |
||||
|
}); |
||||
|
|
||||
|
private timeIntervalsInSec = new Map<TimeUnit, number>([ |
||||
|
[TimeUnit.DAYS, DAY/SECOND], |
||||
|
[TimeUnit.HOURS, HOUR/SECOND], |
||||
|
[TimeUnit.MINUTES, MINUTE/SECOND], |
||||
|
[TimeUnit.SECONDS, SECOND/SECOND], |
||||
|
]); |
||||
|
|
||||
|
private modelValue: number; |
||||
|
|
||||
|
private propagateChange: (value: any) => void = () => {}; |
||||
|
|
||||
|
constructor(private fb: FormBuilder, |
||||
|
private destroyRef: DestroyRef) { |
||||
|
} |
||||
|
|
||||
|
ngOnInit() { |
||||
|
if(this.required || this.maxTime) { |
||||
|
const timeControl = this.timeInputForm.get('time'); |
||||
|
const validators = [Validators.pattern(/^\d*$/)]; |
||||
|
if (this.required) { |
||||
|
validators.push(Validators.required); |
||||
|
} |
||||
|
if (this.maxTime) { |
||||
|
validators.push((control: AbstractControl) => |
||||
|
Validators.max(Math.floor(this.maxTime / this.timeIntervalsInSec.get(this.timeInputForm.get('timeUnit').value)))(control) |
||||
|
); |
||||
|
} |
||||
|
if (isDefinedAndNotNull(this.minTime)) { |
||||
|
validators.push(Validators.min(this.minTime)); |
||||
|
} |
||||
|
|
||||
|
timeControl.setValidators(validators); |
||||
|
timeControl.updateValueAndValidity({ emitEvent: false }); |
||||
|
} |
||||
|
|
||||
|
this.timeInputForm.get('timeUnit').valueChanges.pipe( |
||||
|
takeUntilDestroyed(this.destroyRef) |
||||
|
).subscribe(() => { |
||||
|
this.timeInputForm.get('time').updateValueAndValidity({onlySelf: true}); |
||||
|
this.timeInputForm.get('time').markAsTouched({onlySelf: true}); |
||||
|
}); |
||||
|
|
||||
|
this.timeInputForm.valueChanges.pipe( |
||||
|
takeUntilDestroyed(this.destroyRef) |
||||
|
).subscribe(value => { |
||||
|
this.updatedModel(value); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
registerOnChange(fn: any) { |
||||
|
this.propagateChange = fn; |
||||
|
} |
||||
|
|
||||
|
registerOnTouched(_fn: any) { |
||||
|
} |
||||
|
|
||||
|
setDisabledState(isDisabled: boolean) { |
||||
|
if (isDisabled) { |
||||
|
this.timeInputForm.disable({emitEvent: false}); |
||||
|
} else { |
||||
|
this.timeInputForm.enable({emitEvent: false}); |
||||
|
if(this.timeInputForm.invalid) { |
||||
|
setTimeout(() => this.updatedModel(this.timeInputForm.value, true)) |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
writeValue(sec: number) { |
||||
|
if (sec !== this.modelValue) { |
||||
|
if (isDefinedAndNotNull(sec) && isNumeric(sec) && Number(sec) !== 0) { |
||||
|
this.timeInputForm.patchValue(this.parseTime(sec), {emitEvent: false}); |
||||
|
this.modelValue = sec; |
||||
|
} else { |
||||
|
this.timeInputForm.patchValue({ |
||||
|
time: 0, |
||||
|
timeUnit: TimeUnit.SECONDS |
||||
|
}, {emitEvent: false}); |
||||
|
this.modelValue = 0; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
validate(): ValidationErrors | null { |
||||
|
return this.timeInputForm.valid ? null : { |
||||
|
timeInput: false |
||||
|
}; |
||||
|
} |
||||
|
|
||||
|
private updatedModel(value: Partial<TimeUnitInputModel>, forceUpdated = false) { |
||||
|
const time = value.time * this.timeIntervalsInSec.get(value.timeUnit); |
||||
|
if (this.modelValue !== time || forceUpdated) { |
||||
|
this.modelValue = time; |
||||
|
this.propagateChange(time); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private parseTime(value: number): TimeUnitInputModel { |
||||
|
for (const [timeUnit, timeValue] of this.timeIntervalsInSec) { |
||||
|
const calc = value / timeValue; |
||||
|
if (Number.isInteger(calc)) { |
||||
|
return { |
||||
|
time: calc, |
||||
|
timeUnit: timeUnit |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
} |
||||
Loading…
Reference in new issue