From e8cf3179c53b83b43bdbe777e36b19b1956654d6 Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Tue, 17 Dec 2024 15:45:05 +0200 Subject: [PATCH] Refactor saveAndNotify and saveAndNotifyInternal for attributes --- .../controller/TelemetryController.java | 33 ++- .../device/ClaimDevicesServiceImpl.java | 16 +- .../service/edge/rpc/EdgeGrpcService.java | 22 +- .../telemetry/BaseTelemetryProcessor.java | 43 ++-- .../DefaultTbEntityViewService.java | 66 ++--- .../ota/DefaultOtaPackageStateService.java | 27 +- .../state/DefaultDeviceStateService.java | 42 +-- .../csv/AbstractBulkImportService.java | 39 +-- .../impl/BaseEntityImportService.java | 25 +- .../DefaultTelemetrySubscriptionService.java | 186 +++++--------- .../telemetry/InternalTelemetryService.java | 7 +- .../server/controller/WebsocketApiTest.java | 33 ++- .../state/DefaultDeviceStateServiceTest.java | 242 +++++++++++------- .../engine/api/AttributesSaveRequest.java | 115 +++++++++ .../api/RuleEngineTelemetryService.java | 27 +- .../engine/api/TimeseriesSaveRequest.java | 9 +- .../TbCopyAttributesToEntityViewNode.java | 10 +- .../rule/engine/math/TbMathNode.java | 15 +- .../engine/telemetry/TbMsgAttributesNode.java | 31 ++- .../TbCopyAttributesToEntityViewNodeTest.java | 39 +-- .../rule/engine/math/TbMathNodeTest.java | 11 +- .../telemetry/TbMsgAttributesNodeTest.java | 20 +- .../telemetry/TbMsgTimeseriesNodeTest.java | 11 +- 23 files changed, 619 insertions(+), 450 deletions(-) create mode 100644 rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java diff --git a/application/src/main/java/org/thingsboard/server/controller/TelemetryController.java b/application/src/main/java/org/thingsboard/server/controller/TelemetryController.java index f72b9684e2..b6d854399c 100644 --- a/application/src/main/java/org/thingsboard/server/controller/TelemetryController.java +++ b/application/src/main/java/org/thingsboard/server/controller/TelemetryController.java @@ -47,6 +47,7 @@ import org.springframework.web.bind.annotation.RestController; import org.springframework.web.context.request.async.DeferredResult; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.data.AttributeScope; @@ -625,19 +626,25 @@ public class TelemetryController extends BaseController { } SecurityUser user = getCurrentUser(); return accessValidator.validateEntityAndCallback(getCurrentUser(), Operation.WRITE_ATTRIBUTES, entityIdSrc, (result, tenantId, entityId) -> { - tsSubService.saveAndNotify(tenantId, entityId, scope, attributes, new FutureCallback() { - @Override - public void onSuccess(@Nullable Void tmp) { - logAttributesUpdated(user, entityId, scope, attributes, null); - result.setResult(new ResponseEntity(HttpStatus.OK)); - } - - @Override - public void onFailure(Throwable t) { - logAttributesUpdated(user, entityId, scope, attributes, t); - AccessValidator.handleError(t, result, HttpStatus.INTERNAL_SERVER_ERROR); - } - }); + tsSubService.save(AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(entityId) + .scope(scope) + .entries(attributes) + .callback(new FutureCallback<>() { + @Override + public void onSuccess(@Nullable Void tmp) { + logAttributesUpdated(user, entityId, scope, attributes, null); + result.setResult(new ResponseEntity(HttpStatus.OK)); + } + + @Override + public void onFailure(Throwable t) { + logAttributesUpdated(user, entityId, scope, attributes, t); + AccessValidator.handleError(t, result, HttpStatus.INTERNAL_SERVER_ERROR); + } + }) + .build()); }); } else { return getImmediateDeferredResult("Request is not a JSON object", HttpStatus.BAD_REQUEST); diff --git a/application/src/main/java/org/thingsboard/server/service/device/ClaimDevicesServiceImpl.java b/application/src/main/java/org/thingsboard/server/service/device/ClaimDevicesServiceImpl.java index a77b7d0874..cd3e096ad8 100644 --- a/application/src/main/java/org/thingsboard/server/service/device/ClaimDevicesServiceImpl.java +++ b/application/src/main/java/org/thingsboard/server/service/device/ClaimDevicesServiceImpl.java @@ -28,6 +28,7 @@ import org.springframework.cache.Cache; import org.springframework.cache.CacheManager; import org.springframework.stereotype.Service; import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.Customer; @@ -37,7 +38,6 @@ import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.AttributeKvEntry; -import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; import org.thingsboard.server.common.data.kv.BooleanDataEntry; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.customer.CustomerService; @@ -178,11 +178,12 @@ public class ClaimDevicesServiceImpl implements ClaimDevicesService { return Futures.immediateFuture(new ReclaimResult(unassignedCustomer)); } SettableFuture result = SettableFuture.create(); - telemetryService.saveAndNotify( - tenantId, savedDevice.getId(), AttributeScope.SERVER_SCOPE, List.of( - new BaseAttributeKvEntry(new BooleanDataEntry(CLAIM_ATTRIBUTE_NAME, true), System.currentTimeMillis()) - ), - new FutureCallback<>() { + telemetryService.save(AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(savedDevice.getId()) + .scope(AttributeScope.SERVER_SCOPE) + .entry(new BooleanDataEntry(CLAIM_ATTRIBUTE_NAME, true)) + .callback(new FutureCallback<>() { @Override public void onSuccess(@Nullable Void tmp) { result.set(new ReclaimResult(unassignedCustomer)); @@ -192,7 +193,8 @@ public class ClaimDevicesServiceImpl implements ClaimDevicesService { public void onFailure(Throwable t) { result.setException(t); } - }); + }) + .build()); return result; } cacheEviction(device.getId()); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java index dd2216dd5b..7551302d33 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java @@ -32,6 +32,7 @@ import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.ThingsBoardExecutors; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.cache.TbTransactionalCache; import org.thingsboard.server.cluster.TbClusterService; @@ -42,7 +43,6 @@ import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.kv.BasicTsKvEntry; import org.thingsboard.server.common.data.kv.BooleanDataEntry; import org.thingsboard.server.common.data.kv.LongDataEntry; import org.thingsboard.server.common.data.msg.TbMsgType; @@ -506,11 +506,17 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i tsSubService.save(TimeseriesSaveRequest.builder() .tenantId(tenantId) .entityId(edgeId) - .entry(new BasicTsKvEntry(System.currentTimeMillis(), new LongDataEntry(key, value))) + .entry(new LongDataEntry(key, value)) .callback(new AttributeSaveCallback(tenantId, edgeId, key, value)) .build()); } else { - tsSubService.saveAttrAndNotify(tenantId, edgeId, AttributeScope.SERVER_SCOPE, key, value, new AttributeSaveCallback(tenantId, edgeId, key, value)); + tsSubService.save(AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(edgeId) + .scope(AttributeScope.SERVER_SCOPE) + .entry(new LongDataEntry(key, value)) + .callback(new AttributeSaveCallback(tenantId, edgeId, key, value)) + .build()); } } @@ -520,11 +526,17 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i tsSubService.save(TimeseriesSaveRequest.builder() .tenantId(tenantId) .entityId(edgeId) - .entry(new BasicTsKvEntry(System.currentTimeMillis(), new BooleanDataEntry(key, value))) + .entry(new BooleanDataEntry(key, value)) .callback(new AttributeSaveCallback(tenantId, edgeId, key, value)) .build()); } else { - tsSubService.saveAttrAndNotify(tenantId, edgeId, AttributeScope.SERVER_SCOPE, key, value, new AttributeSaveCallback(tenantId, edgeId, key, value)); + tsSubService.save(AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(edgeId) + .scope(AttributeScope.SERVER_SCOPE) + .entry(new BooleanDataEntry(key, value)) + .callback(new AttributeSaveCallback(tenantId, edgeId, key, value)) + .build()); } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java index d94d4d8939..9ef40280c9 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java @@ -31,6 +31,7 @@ import org.apache.commons.lang3.tuple.Pair; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Lazy; import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.DataConstants; @@ -277,16 +278,29 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor { JsonObject json = JsonUtils.getJsonObject(msg.getKvList()); List attributes = new ArrayList<>(JsonConverter.convertToAttributes(json)); String scope = metaData.getValue("scope"); - tsSubService.saveAndNotify(tenantId, entityId, AttributeScope.valueOf(scope), attributes, new FutureCallback() { - @Override - public void onSuccess(@Nullable Void tmp) { - var defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId); - TbMsg tbMsg = TbMsg.newMsg(defaultQueueAndRuleChain.getKey(), TbMsgType.ATTRIBUTES_UPDATED, entityId, - customerId, metaData, gson.toJson(json), defaultQueueAndRuleChain.getValue(), null); - edgeCtx.getClusterService().pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() { + tsSubService.save(AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(entityId) + .scope(AttributeScope.valueOf(scope)) + .entries(attributes) + .callback(new FutureCallback<>() { @Override - public void onSuccess(TbQueueMsgMetadata metadata) { - futureToSet.set(null); + public void onSuccess(@Nullable Void tmp) { + var defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId); + TbMsg tbMsg = TbMsg.newMsg(defaultQueueAndRuleChain.getKey(), TbMsgType.ATTRIBUTES_UPDATED, entityId, + customerId, metaData, gson.toJson(json), defaultQueueAndRuleChain.getValue(), null); + edgeCtx.getClusterService().pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() { + @Override + public void onSuccess(TbQueueMsgMetadata metadata) { + futureToSet.set(null); + } + + @Override + public void onFailure(Throwable t) { + log.error("[{}] Can't process attributes update [{}]", tenantId, msg, t); + futureToSet.setException(t); + } + }); } @Override @@ -294,15 +308,8 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor { log.error("[{}] Can't process attributes update [{}]", tenantId, msg, t); futureToSet.setException(t); } - }); - } - - @Override - public void onFailure(Throwable t) { - log.error("[{}] Can't process attributes update [{}]", tenantId, msg, t); - futureToSet.setException(t); - } - }); + }) + .build()); return futureToSet; } diff --git a/application/src/main/java/org/thingsboard/server/service/entitiy/entityview/DefaultTbEntityViewService.java b/application/src/main/java/org/thingsboard/server/service/entitiy/entityview/DefaultTbEntityViewService.java index 82877283cd..790b1086b4 100644 --- a/application/src/main/java/org/thingsboard/server/service/entitiy/entityview/DefaultTbEntityViewService.java +++ b/application/src/main/java/org/thingsboard/server/service/entitiy/entityview/DefaultTbEntityViewService.java @@ -25,6 +25,7 @@ import lombok.AllArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import org.springframework.util.ConcurrentReferenceHashMap; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.EntityType; @@ -273,36 +274,41 @@ public class DefaultTbEntityViewService extends AbstractTbEntityService implemen return Futures.transform(getAttrFuture, attributeKvEntries -> { List attributes; if (attributeKvEntries != null && !attributeKvEntries.isEmpty()) { - attributes = - attributeKvEntries.stream() - .filter(attributeKvEntry -> { - long startTime = entityView.getStartTimeMs(); - long endTime = entityView.getEndTimeMs(); - long lastUpdateTs = attributeKvEntry.getLastUpdateTs(); - return startTime == 0 && endTime == 0 || - (endTime == 0 && startTime < lastUpdateTs) || - (startTime == 0 && endTime > lastUpdateTs) || - (startTime < lastUpdateTs && endTime > lastUpdateTs); - }).collect(Collectors.toList()); - tsSubService.saveAndNotify(entityView.getTenantId(), entityId, scope, attributes, new FutureCallback() { - @Override - public void onSuccess(@Nullable Void tmp) { - try { - logAttributesUpdated(entityView.getTenantId(), user, entityId, scope, attributes, null); - } catch (ThingsboardException e) { - log.error("Failed to log attribute updates", e); - } - } - - @Override - public void onFailure(Throwable t) { - try { - logAttributesUpdated(entityView.getTenantId(), user, entityId, scope, attributes, t); - } catch (ThingsboardException e) { - log.error("Failed to log attribute updates", e); - } - } - }); + attributes = attributeKvEntries.stream() + .filter(attributeKvEntry -> { + long startTime = entityView.getStartTimeMs(); + long endTime = entityView.getEndTimeMs(); + long lastUpdateTs = attributeKvEntry.getLastUpdateTs(); + return startTime == 0 && endTime == 0 || + (endTime == 0 && startTime < lastUpdateTs) || + (startTime == 0 && endTime > lastUpdateTs) || + (startTime < lastUpdateTs && endTime > lastUpdateTs); + }).collect(Collectors.toList()); + tsSubService.save(AttributesSaveRequest.builder() + .tenantId(entityView.getTenantId()) + .entityId(entityId) + .scope(scope) + .entries(attributes) + .callback(new FutureCallback<>() { + @Override + public void onSuccess(@Nullable Void tmp) { + try { + logAttributesUpdated(entityView.getTenantId(), user, entityId, scope, attributes, null); + } catch (ThingsboardException e) { + log.error("Failed to log attribute updates", e); + } + } + + @Override + public void onFailure(Throwable t) { + try { + logAttributesUpdated(entityView.getTenantId(), user, entityId, scope, attributes, t); + } catch (ThingsboardException e) { + log.error("Failed to log attribute updates", e); + } + } + }) + .build()); } return null; }, MoreExecutors.directExecutor()); diff --git a/application/src/main/java/org/thingsboard/server/service/ota/DefaultOtaPackageStateService.java b/application/src/main/java/org/thingsboard/server/service/ota/DefaultOtaPackageStateService.java index afae615333..3db4e53379 100644 --- a/application/src/main/java/org/thingsboard/server/service/ota/DefaultOtaPackageStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/ota/DefaultOtaPackageStateService.java @@ -20,6 +20,7 @@ import jakarta.annotation.Nullable; import lombok.extern.slf4j.Slf4j; import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.cluster.TbClusterService; @@ -346,17 +347,23 @@ public class DefaultOtaPackageStateService implements OtaPackageStateService { remove(device, otaPackageType, attrToRemove); - telemetryService.saveAndNotify(tenantId, deviceId, AttributeScope.SHARED_SCOPE, attributes, new FutureCallback<>() { - @Override - public void onSuccess(@Nullable Void tmp) { - log.trace("[{}] Success save attributes with target firmware!", deviceId); - } + telemetryService.save(AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .scope(AttributeScope.SHARED_SCOPE) + .entries(attributes) + .callback(new FutureCallback<>() { + @Override + public void onSuccess(@Nullable Void tmp) { + log.trace("[{}] Success save attributes with target firmware!", deviceId); + } - @Override - public void onFailure(Throwable t) { - log.error("[{}] Failed to save attributes with target firmware!", deviceId, t); - } - }); + @Override + public void onFailure(Throwable t) { + log.error("[{}] Failed to save attributes with target firmware!", deviceId, t); + } + }) + .build()); } private void remove(Device device, OtaPackageType otaPackageType) { diff --git a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java index 4f230f9d6b..8280cf44b8 100644 --- a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java @@ -38,6 +38,7 @@ import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.ThingsBoardExecutors; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.ApiUsageRecordKey; @@ -52,6 +53,7 @@ import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UUIDBased; import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; import org.thingsboard.server.common.data.kv.BasicTsKvEntry; import org.thingsboard.server.common.data.kv.BooleanDataEntry; import org.thingsboard.server.common.data.kv.KvEntry; @@ -866,30 +868,30 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService(deviceId, key, value)) - .build()); - } else { - tsSubService.saveAttrAndNotify(TenantId.SYS_TENANT_ID, deviceId, AttributeScope.SERVER_SCOPE, key, value, new TelemetrySaveCallback<>(deviceId, key, value)); - } + save(deviceId, new LongDataEntry(key, value), getCurrentTimeMillis()); } private void save(DeviceId deviceId, String key, boolean value) { + save(deviceId, new BooleanDataEntry(key, value), getCurrentTimeMillis()); + } + + private void save(DeviceId deviceId, KvEntry kvEntry, long ts) { if (persistToTelemetry) { tsSubService.saveInternal(TimeseriesSaveRequest.builder() .tenantId(TenantId.SYS_TENANT_ID) .entityId(deviceId) - .entry(new BasicTsKvEntry(getCurrentTimeMillis(), new BooleanDataEntry(key, value))) + .entry(new BasicTsKvEntry(ts, kvEntry)) .ttl(telemetryTtl) - .callback(new TelemetrySaveCallback<>(deviceId, key, value)) + .callback(new TelemetrySaveCallback<>(deviceId, kvEntry)) .build()); } else { - tsSubService.saveAttrAndNotify(TenantId.SYS_TENANT_ID, deviceId, AttributeScope.SERVER_SCOPE, key, value, new TelemetrySaveCallback<>(deviceId, key, value)); + tsSubService.save(AttributesSaveRequest.builder() + .tenantId(TenantId.SYS_TENANT_ID) + .entityId(deviceId) + .scope(AttributeScope.SERVER_SCOPE) + .entry(new BaseAttributeKvEntry(ts, kvEntry)) + .callback(new TelemetrySaveCallback<>(deviceId, kvEntry)) + .build()); } } @@ -899,23 +901,21 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService implements FutureCallback { private final DeviceId deviceId; - private final String key; - private final Object value; + private final KvEntry kvEntry; - TelemetrySaveCallback(DeviceId deviceId, String key, Object value) { + TelemetrySaveCallback(DeviceId deviceId, KvEntry kvEntry) { this.deviceId = deviceId; - this.key = key; - this.value = value; + this.kvEntry = kvEntry; } @Override public void onSuccess(@Nullable T result) { - log.trace("[{}] Successfully updated attribute [{}] with value [{}]", deviceId, key, value); + log.trace("[{}] Successfully updated entry {}", deviceId, kvEntry); } @Override public void onFailure(Throwable t) { - log.warn("[{}] Failed to update attribute [{}] with value [{}]", deviceId, key, value, t); + log.warn("[{}] Failed to update entry {}", deviceId, kvEntry, t); } } } diff --git a/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/csv/AbstractBulkImportService.java b/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/csv/AbstractBulkImportService.java index 6d04c3173f..f3e8266ce6 100644 --- a/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/csv/AbstractBulkImportService.java +++ b/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/csv/AbstractBulkImportService.java @@ -32,6 +32,7 @@ import org.springframework.security.core.context.SecurityContextHolder; import org.thingsboard.common.util.DonAsynchron; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.ThingsBoardExecutors; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.data.AttributeScope; @@ -237,23 +238,27 @@ public abstract class AbstractBulkImportService attributes = new ArrayList<>(JsonConverter.convertToAttributes(kvsEntry.getValue())); accessValidator.validateEntityAndCallback(user, Operation.WRITE_ATTRIBUTES, entity.getId(), (result, tenantId, entityId) -> { - tsSubscriptionService.saveAndNotify(tenantId, entityId, AttributeScope.valueOf(scope), attributes, new FutureCallback<>() { - - @Override - public void onSuccess(Void unused) { - entityActionService.logEntityAction(user, (UUIDBased & EntityId) entityId, null, - null, ActionType.ATTRIBUTES_UPDATED, null, AttributeScope.valueOf(scope), attributes); - } - - @Override - public void onFailure(Throwable throwable) { - entityActionService.logEntityAction(user, (UUIDBased & EntityId) entityId, null, - null, ActionType.ATTRIBUTES_UPDATED, BaseController.toException(throwable), - AttributeScope.valueOf(scope), attributes); - throw new RuntimeException(throwable); - } - - }); + tsSubscriptionService.save(AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(entityId) + .scope(AttributeScope.valueOf(scope)) + .entries(attributes) + .callback(new FutureCallback<>() { + @Override + public void onSuccess(Void unused) { + entityActionService.logEntityAction(user, (UUIDBased & EntityId) entityId, null, + null, ActionType.ATTRIBUTES_UPDATED, null, AttributeScope.valueOf(scope), attributes); + } + + @Override + public void onFailure(Throwable throwable) { + entityActionService.logEntityAction(user, (UUIDBased & EntityId) entityId, null, + null, ActionType.ATTRIBUTES_UPDATED, BaseController.toException(throwable), + AttributeScope.valueOf(scope), attributes); + throw new RuntimeException(throwable); + } + }) + .build()); }); } diff --git a/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/impl/BaseEntityImportService.java b/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/impl/BaseEntityImportService.java index f88844bd09..199f54cd12 100644 --- a/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/impl/BaseEntityImportService.java +++ b/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/impl/BaseEntityImportService.java @@ -24,6 +24,7 @@ import org.checkerframework.checker.nullness.qual.Nullable; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Lazy; import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.ExportableEntity; @@ -257,16 +258,22 @@ public abstract class BaseEntityImportService() { - @Override - public void onSuccess(@Nullable Void unused) { - } + tsSubService.save(AttributesSaveRequest.builder() + .tenantId(user.getTenantId()) + .entityId(entity.getId()) + .scope(scope) + .entries(attributeKvEntries) + .callback(new FutureCallback<>() { + @Override + public void onSuccess(@Nullable Void unused) { + } - @Override - public void onFailure(Throwable thr) { - log.error("Failed to import attributes for {} {}", entity.getId().getEntityType(), entity.getId(), thr); - } - }); + @Override + public void onFailure(Throwable thr) { + log.error("Failed to import attributes for {} {}", entity.getId().getEntityType(), entity.getId(), thr); + } + }) + .build()); }); }); } diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java index 1ce0550ee2..bc78139d2f 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java @@ -28,6 +28,7 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.common.data.ApiUsageRecordKey; @@ -38,12 +39,7 @@ import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.AttributeKvEntry; -import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; -import org.thingsboard.server.common.data.kv.BooleanDataEntry; import org.thingsboard.server.common.data.kv.DeleteTsKvQuery; -import org.thingsboard.server.common.data.kv.DoubleDataEntry; -import org.thingsboard.server.common.data.kv.LongDataEntry; -import org.thingsboard.server.common.data.kv.StringDataEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.kv.TsKvLatestRemovingResult; import org.thingsboard.server.common.msg.queue.TbCallback; @@ -57,7 +53,6 @@ import org.thingsboard.server.service.subscription.TbSubscriptionUtils; import java.util.ArrayList; import java.util.Collection; -import java.util.Collections; import java.util.Comparator; import java.util.HashMap; import java.util.List; @@ -159,89 +154,18 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer return saveFuture; } - private void addEntityViewCallback(TenantId tenantId, EntityId entityId, List ts) { - if (EntityType.DEVICE.equals(entityId.getEntityType()) || EntityType.ASSET.equals(entityId.getEntityType())) { - Futures.addCallback(this.tbEntityViewService.findEntityViewsByTenantIdAndEntityIdAsync(tenantId, entityId), - new FutureCallback<>() { - @Override - public void onSuccess(@Nullable List result) { - if (result != null && !result.isEmpty()) { - Map> tsMap = new HashMap<>(); - for (TsKvEntry entry : ts) { - tsMap.computeIfAbsent(entry.getKey(), s -> new ArrayList<>()).add(entry); - } - for (EntityView entityView : result) { - List keys = entityView.getKeys() != null && entityView.getKeys().getTimeseries() != null ? - entityView.getKeys().getTimeseries() : new ArrayList<>(tsMap.keySet()); - List entityViewLatest = new ArrayList<>(); - long startTs = entityView.getStartTimeMs(); - long endTs = entityView.getEndTimeMs() == 0 ? Long.MAX_VALUE : entityView.getEndTimeMs(); - for (String key : keys) { - List entries = tsMap.get(key); - if (entries != null) { - Optional tsKvEntry = entries.stream() - .filter(entry -> entry.getTs() > startTs && entry.getTs() <= endTs) - .max(Comparator.comparingLong(TsKvEntry::getTs)); - tsKvEntry.ifPresent(entityViewLatest::add); - } - } - if (!entityViewLatest.isEmpty()) { - saveLatestAndNotify(tenantId, entityView.getId(), entityViewLatest, new FutureCallback<>() { - @Override - public void onSuccess(@Nullable Void tmp) { - } - - @Override - public void onFailure(Throwable t) { - } - }); - } - } - } - } - - @Override - public void onFailure(Throwable t) { - log.error("Error while finding entity views by tenantId and entityId", t); - } - }, MoreExecutors.directExecutor()); - } - } - - @Override - public void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List attributes, FutureCallback callback) { - saveAndNotify(tenantId, entityId, scope, attributes, true, callback); - } - @Override - public void saveAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List attributes, FutureCallback callback) { - saveAndNotify(tenantId, entityId, scope, attributes, true, callback); + public void save(AttributesSaveRequest request) { + checkInternalEntity(request.getEntityId()); + saveInternal(request); } @Override - public void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List attributes, boolean notifyDevice, FutureCallback callback) { - checkInternalEntity(entityId); - saveAndNotifyInternal(tenantId, entityId, scope, attributes, notifyDevice, callback); - } - - @Override - public void saveAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List attributes, boolean notifyDevice, FutureCallback callback) { - checkInternalEntity(entityId); - saveAndNotifyInternal(tenantId, entityId, scope, attributes, notifyDevice, callback); - } - - @Override - public void saveAndNotifyInternal(TenantId tenantId, EntityId entityId, String scope, List attributes, boolean notifyDevice, FutureCallback callback) { - ListenableFuture> saveFuture = attrService.save(tenantId, entityId, scope, attributes); - addVoidCallback(saveFuture, callback); - addWsCallback(saveFuture, success -> onAttributesUpdate(tenantId, entityId, scope, attributes, notifyDevice)); - } - - @Override - public void saveAndNotifyInternal(TenantId tenantId, EntityId entityId, AttributeScope scope, List attributes, boolean notifyDevice, FutureCallback callback) { - ListenableFuture> saveFuture = attrService.save(tenantId, entityId, scope, attributes); - addVoidCallback(saveFuture, callback); - addWsCallback(saveFuture, success -> onAttributesUpdate(tenantId, entityId, scope.name(), attributes, notifyDevice)); + public void saveInternal(AttributesSaveRequest request) { + log.trace("Executing saveInternal [{}]", request); + ListenableFuture> saveFuture = attrService.save(request.getTenantId(), request.getEntityId(), request.getScope(), request.getEntries()); + addVoidCallback(saveFuture, request.getCallback()); + addWsCallback(saveFuture, success -> onAttributesUpdate(request.getTenantId(), request.getEntityId(), request.getScope().name(), request.getEntries(), request.isNotifyDevice())); } @Override @@ -318,57 +242,61 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer addWsCallback(deleteFuture, list -> onTimeSeriesDelete(tenantId, entityId, keys, list)); } - - @Override - public void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, long value, FutureCallback callback) { - saveAndNotify(tenantId, entityId, scope, Collections.singletonList(new BaseAttributeKvEntry(new LongDataEntry(key, value) - , System.currentTimeMillis())), callback); - } - @Override - public void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, String value, FutureCallback callback) { - saveAndNotify(tenantId, entityId, scope, Collections.singletonList(new BaseAttributeKvEntry(new StringDataEntry(key, value) - , System.currentTimeMillis())), callback); - } - - @Override - public void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, double value, FutureCallback callback) { - saveAndNotify(tenantId, entityId, scope, Collections.singletonList(new BaseAttributeKvEntry(new DoubleDataEntry(key, value) - , System.currentTimeMillis())), callback); - } - - @Override - public void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, boolean value, FutureCallback callback) { - saveAndNotify(tenantId, entityId, scope, Collections.singletonList(new BaseAttributeKvEntry(new BooleanDataEntry(key, value) - , System.currentTimeMillis())), callback); - } - - @Override - public ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, long value) { + public ListenableFuture saveAttrAndNotify(AttributesSaveRequest request) { SettableFuture future = SettableFuture.create(); - saveAttrAndNotify(tenantId, entityId, scope, key, value, new VoidFutureCallback(future)); + request.setCallback(new VoidFutureCallback(future)); + save(request); return future; } - @Override - public ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, String value) { - SettableFuture future = SettableFuture.create(); - saveAttrAndNotify(tenantId, entityId, scope, key, value, new VoidFutureCallback(future)); - return future; - } + private void addEntityViewCallback(TenantId tenantId, EntityId entityId, List ts) { + if (EntityType.DEVICE.equals(entityId.getEntityType()) || EntityType.ASSET.equals(entityId.getEntityType())) { + Futures.addCallback(this.tbEntityViewService.findEntityViewsByTenantIdAndEntityIdAsync(tenantId, entityId), + new FutureCallback<>() { + @Override + public void onSuccess(@Nullable List result) { + if (result != null && !result.isEmpty()) { + Map> tsMap = new HashMap<>(); + for (TsKvEntry entry : ts) { + tsMap.computeIfAbsent(entry.getKey(), s -> new ArrayList<>()).add(entry); + } + for (EntityView entityView : result) { + List keys = entityView.getKeys() != null && entityView.getKeys().getTimeseries() != null ? + entityView.getKeys().getTimeseries() : new ArrayList<>(tsMap.keySet()); + List entityViewLatest = new ArrayList<>(); + long startTs = entityView.getStartTimeMs(); + long endTs = entityView.getEndTimeMs() == 0 ? Long.MAX_VALUE : entityView.getEndTimeMs(); + for (String key : keys) { + List entries = tsMap.get(key); + if (entries != null) { + Optional tsKvEntry = entries.stream() + .filter(entry -> entry.getTs() > startTs && entry.getTs() <= endTs) + .max(Comparator.comparingLong(TsKvEntry::getTs)); + tsKvEntry.ifPresent(entityViewLatest::add); + } + } + if (!entityViewLatest.isEmpty()) { + saveLatestAndNotify(tenantId, entityView.getId(), entityViewLatest, new FutureCallback<>() { + @Override + public void onSuccess(@Nullable Void tmp) { + } - @Override - public ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, double value) { - SettableFuture future = SettableFuture.create(); - saveAttrAndNotify(tenantId, entityId, scope, key, value, new VoidFutureCallback(future)); - return future; - } + @Override + public void onFailure(Throwable t) { + } + }); + } + } + } + } - @Override - public ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, boolean value) { - SettableFuture future = SettableFuture.create(); - saveAttrAndNotify(tenantId, entityId, scope, key, value, new VoidFutureCallback(future)); - return future; + @Override + public void onFailure(Throwable t) { + log.error("Error while finding entity views by tenantId and entityId", t); + } + }, MoreExecutors.directExecutor()); + } } private void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes, boolean notifyDevice) { diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/InternalTelemetryService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/InternalTelemetryService.java index c861efb46f..20eff8b6b5 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/InternalTelemetryService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/InternalTelemetryService.java @@ -17,12 +17,12 @@ package org.thingsboard.server.service.telemetry; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.ListenableFuture; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; import java.util.List; @@ -34,10 +34,7 @@ public interface InternalTelemetryService extends RuleEngineTelemetryService { ListenableFuture saveInternal(TimeseriesSaveRequest request); - @Deprecated(since = "3.7.0") - void saveAndNotifyInternal(TenantId tenantId, EntityId entityId, String scope, List attributes, boolean notifyDevice, FutureCallback callback); - - void saveAndNotifyInternal(TenantId tenantId, EntityId entityId, AttributeScope scope, List attributes, boolean notifyDevice, FutureCallback callback); + void saveInternal(AttributesSaveRequest request); void saveLatestAndNotifyInternal(TenantId tenantId, EntityId entityId, List ts, FutureCallback callback); diff --git a/application/src/test/java/org/thingsboard/server/controller/WebsocketApiTest.java b/application/src/test/java/org/thingsboard/server/controller/WebsocketApiTest.java index 40c7e435ca..05e7132ce8 100644 --- a/application/src/test/java/org/thingsboard/server/controller/WebsocketApiTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/WebsocketApiTest.java @@ -29,6 +29,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.test.context.TestPropertySource; import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils; import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.alarm.Alarm; @@ -832,19 +833,25 @@ public class WebsocketApiTest extends AbstractControllerTest { private void sendAttributes(TenantId tenantId, EntityId entityId, TbAttributeSubscriptionScope scope, List attrData) throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); - tsService.saveAndNotify(tenantId, entityId, scope.getAttributeScope(), attrData, new FutureCallback() { - @Override - public void onSuccess(@Nullable Void result) { - log.debug("sendAttributes callback onSuccess"); - latch.countDown(); - } - - @Override - public void onFailure(Throwable t) { - log.error("Failed to sendAttributes", t); - latch.countDown(); - } - }); + tsService.save(AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(entityId) + .scope(scope.getAttributeScope()) + .entries(attrData) + .callback(new FutureCallback<>() { + @Override + public void onSuccess(@Nullable Void result) { + log.debug("sendAttributes callback onSuccess"); + latch.countDown(); + } + + @Override + public void onFailure(Throwable t) { + log.error("Failed to sendAttributes", t); + latch.countDown(); + } + }) + .build()); assertThat(latch.await(TIMEOUT, TimeUnit.SECONDS)).as("await sendAttributes callback").isTrue(); } diff --git a/application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java b/application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java index 296191d61b..b7207da23a 100644 --- a/application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java @@ -24,9 +24,11 @@ import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.ValueSource; import org.mockito.ArgumentCaptor; +import org.mockito.ArgumentMatcher; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import org.springframework.test.util.ReflectionTestUtils; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.Device; @@ -72,7 +74,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyList; -import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.then; @@ -82,6 +84,7 @@ import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.never; import static org.mockito.Mockito.reset; import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.thingsboard.server.service.state.DefaultDeviceStateService.ACTIVITY_STATE; @@ -208,9 +211,12 @@ public class DefaultDeviceStateServiceTest { service.onDeviceConnect(tenantId, deviceId, lastConnectTime); // THEN - then(telemetrySubscriptionService).should().saveAttrAndNotify( - eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), eq(LAST_CONNECT_TIME), eq(lastConnectTime), any() - ); + then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher) request -> + request.getTenantId().equals(TenantId.SYS_TENANT_ID) && request.getEntityId().equals(deviceId) && + request.getScope().equals(AttributeScope.SERVER_SCOPE) && + request.getEntries().get(0).getKey().equals(LAST_CONNECT_TIME) && + request.getEntries().get(0).getValue().equals(lastConnectTime) + )); var msgCaptor = ArgumentCaptor.forClass(TbMsg.class); then(clusterService).should().pushMsgToRuleEngine(eq(tenantId), eq(deviceId), msgCaptor.capture(), any()); @@ -292,10 +298,12 @@ public class DefaultDeviceStateServiceTest { service.onDeviceDisconnect(tenantId, deviceId, lastDisconnectTime); // THEN - then(telemetrySubscriptionService).should().saveAttrAndNotify( - eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), - eq(LAST_DISCONNECT_TIME), eq(lastDisconnectTime), any() - ); + then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher) request -> + request.getTenantId().equals(TenantId.SYS_TENANT_ID) && request.getEntityId().equals(deviceId) && + request.getScope().equals(AttributeScope.SERVER_SCOPE) && + request.getEntries().get(0).getKey().equals(LAST_DISCONNECT_TIME) && + request.getEntries().get(0).getValue().equals(lastDisconnectTime) + )); var msgCaptor = ArgumentCaptor.forClass(TbMsg.class); then(clusterService).should().pushMsgToRuleEngine(eq(tenantId), eq(deviceId), msgCaptor.capture(), any()); @@ -413,14 +421,18 @@ public class DefaultDeviceStateServiceTest { service.onDeviceInactivity(tenantId, deviceId, lastInactivityTime); // THEN - then(telemetrySubscriptionService).should().saveAttrAndNotify( - eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), - eq(INACTIVITY_ALARM_TIME), eq(lastInactivityTime), any() - ); - then(telemetrySubscriptionService).should().saveAttrAndNotify( - eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), - eq(ACTIVITY_STATE), eq(false), any() - ); + then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher) request -> + request.getTenantId().equals(TenantId.SYS_TENANT_ID) && request.getEntityId().equals(deviceId) && + request.getScope().equals(AttributeScope.SERVER_SCOPE) && + request.getEntries().get(0).getKey().equals(INACTIVITY_ALARM_TIME) && + request.getEntries().get(0).getValue().equals(lastInactivityTime) + )); + then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher) request -> + request.getTenantId().equals(TenantId.SYS_TENANT_ID) && request.getEntityId().equals(deviceId) && + request.getScope().equals(AttributeScope.SERVER_SCOPE) && + request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) && + request.getEntries().get(0).getValue().equals(false) + )); var msgCaptor = ArgumentCaptor.forClass(TbMsg.class); then(clusterService).should() @@ -453,14 +465,17 @@ public class DefaultDeviceStateServiceTest { service.updateInactivityStateIfExpired(System.currentTimeMillis(), deviceId, deviceStateData); // THEN - then(telemetrySubscriptionService).should().saveAttrAndNotify( - eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), - eq(INACTIVITY_ALARM_TIME), anyLong(), any() - ); - then(telemetrySubscriptionService).should().saveAttrAndNotify( - eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), - eq(ACTIVITY_STATE), eq(false), any() - ); + then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher) request -> + request.getTenantId().equals(TenantId.SYS_TENANT_ID) && request.getEntityId().equals(deviceId) && + request.getScope().equals(AttributeScope.SERVER_SCOPE) && + request.getEntries().get(0).getKey().equals(INACTIVITY_ALARM_TIME) + )); + then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher) request -> + request.getTenantId().equals(TenantId.SYS_TENANT_ID) && request.getEntityId().equals(deviceId) && + request.getScope().equals(AttributeScope.SERVER_SCOPE) && + request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) && + request.getEntries().get(0).getValue().equals(false) + )); var msgCaptor = ArgumentCaptor.forClass(TbMsg.class); then(clusterService).should() @@ -612,7 +627,9 @@ public class DefaultDeviceStateServiceTest { long newTimeout = System.currentTimeMillis() - deviceState.getLastActivityTime() + increase; service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, newTimeout); - verify(telemetrySubscriptionService, never()).saveAttrAndNotify(any(), eq(deviceId), any(AttributeScope.class), eq(ACTIVITY_STATE), any(), any()); + verify(telemetrySubscriptionService, never()).save(argThat((ArgumentMatcher) request -> + request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) + )); Thread.sleep(defaultTimeout + increase); service.checkStates(); activityVerify(false); @@ -651,7 +668,9 @@ public class DefaultDeviceStateServiceTest { long newTimeout = 1; Thread.sleep(newTimeout); - verify(telemetrySubscriptionService, never()).saveAttrAndNotify(any(), eq(deviceId), any(AttributeScope.class), eq(ACTIVITY_STATE), any(), any()); + verify(telemetrySubscriptionService, never()).save(argThat((ArgumentMatcher) request -> + request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) + )); } @Test @@ -672,8 +691,6 @@ public class DefaultDeviceStateServiceTest { service.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis()); activityVerify(true); - verify(telemetrySubscriptionService, never()).saveAttrAndNotify(any(), eq(deviceId), any(AttributeScope.class), eq(ACTIVITY_STATE), any(), any()); - long newTimeout = 1; Thread.sleep(newTimeout); @@ -713,11 +730,17 @@ public class DefaultDeviceStateServiceTest { long newTimeout = 1; service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, newTimeout); - verify(telemetrySubscriptionService, never()).saveAttrAndNotify(any(), eq(deviceId), any(AttributeScope.class), eq(ACTIVITY_STATE), any(), any()); + verify(telemetrySubscriptionService, never()).save(argThat((ArgumentMatcher) request -> + request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) + )); } private void activityVerify(boolean isActive) { - verify(telemetrySubscriptionService).saveAttrAndNotify(any(), eq(deviceId), any(AttributeScope.class), eq(ACTIVITY_STATE), eq(isActive), any()); + verify(telemetrySubscriptionService).save(argThat((ArgumentMatcher) request -> + request.getEntityId().equals(deviceId) && + request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) && + request.getEntries().get(0).getValue().equals(isActive) + )); } @Test @@ -763,21 +786,27 @@ public class DefaultDeviceStateServiceTest { // THEN assertThat(deviceState.isActive()).isEqualTo(true); assertThat(deviceState.getLastActivityTime()).isEqualTo(lastReportedActivity); - then(telemetrySubscriptionService).should().saveAttrAndNotify( - any(), eq(deviceId), any(AttributeScope.class), eq(LAST_ACTIVITY_TIME), eq(lastReportedActivity), any() - ); + then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher) request -> + request.getEntityId().equals(deviceId) && + request.getEntries().get(0).getKey().equals(LAST_ACTIVITY_TIME) && + request.getEntries().get(0).getValue().equals(lastReportedActivity) + )); assertThat(deviceState.getLastInactivityAlarmTime()).isEqualTo(expectedInactivityAlarmTime); if (shouldSetInactivityAlarmTimeToZero) { - then(telemetrySubscriptionService).should().saveAttrAndNotify( - any(), eq(deviceId), any(AttributeScope.class), eq(INACTIVITY_ALARM_TIME), eq(0L), any() - ); + then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher) request -> + request.getEntityId().equals(deviceId) && + request.getEntries().get(0).getKey().equals(INACTIVITY_ALARM_TIME) && + request.getEntries().get(0).getValue().equals(0L) + )); } if (shouldUpdateActivityStateToActive) { - then(telemetrySubscriptionService).should().saveAttrAndNotify( - eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), eq(ACTIVITY_STATE), eq(true), any() - ); + then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher) request -> + request.getEntityId().equals(deviceId) && + request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) && + request.getEntries().get(0).getValue().equals(true) + )); var msgCaptor = ArgumentCaptor.forClass(TbMsg.class); then(clusterService).should().pushMsgToRuleEngine(eq(tenantId), eq(deviceId), msgCaptor.capture(), any()); @@ -796,28 +825,28 @@ public class DefaultDeviceStateServiceTest { private static Stream provideParametersForUpdateActivityState() { return Stream.of( - Arguments.of(true, 100, 120, 80, 80, false, false), + Arguments.of(true, 100, 120, 80, 80, false, false), - Arguments.of(true, 100, 120, 100, 100, false, false), + Arguments.of(true, 100, 120, 100, 100, false, false), Arguments.of(false, 100, 120, 110, 110, false, true), - Arguments.of(true, 100, 100, 80, 80, false, false), + Arguments.of(true, 100, 100, 80, 80, false, false), - Arguments.of(true, 100, 100, 100, 100, false, false), + Arguments.of(true, 100, 100, 100, 100, false, false), - Arguments.of(false, 100, 100, 110, 0, true, true), + Arguments.of(false, 100, 100, 110, 0, true, true), - Arguments.of(false, 100, 110, 110, 0, true, true), + Arguments.of(false, 100, 110, 110, 0, true, true), - Arguments.of(false, 100, 110, 120, 0, true, true), + Arguments.of(false, 100, 110, 120, 0, true, true), - Arguments.of(true, 0, 0, 0, 0, false, false), + Arguments.of(true, 0, 0, 0, 0, false, false), - Arguments.of(false, 0, 0, 0, 0, true, true) + Arguments.of(false, 0, 0, 0, 0, true, true) ); } @@ -857,9 +886,10 @@ public class DefaultDeviceStateServiceTest { assertThat(deviceState.getInactivityTimeout()).isEqualTo(newInactivityTimeout); assertThat(deviceState.isActive()).isEqualTo(expectedActivityState); if (activityState && !expectedActivityState) { - then(telemetrySubscriptionService).should().saveAttrAndNotify( - any(), eq(deviceId), any(AttributeScope.class), eq(ACTIVITY_STATE), eq(false), any() - ); + then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher) request -> + request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) && + request.getEntries().get(0).getValue().equals(false) + )); } } @@ -954,9 +984,10 @@ public class DefaultDeviceStateServiceTest { assertThat(state.getLastInactivityAlarmTime()).isEqualTo(expectedLastInactivityAlarmTime); if (shouldUpdateActivityStateToInactive) { - then(telemetrySubscriptionService).should().saveAttrAndNotify( - eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), eq(ACTIVITY_STATE), eq(false), any() - ); + then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher) request -> + request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) && + request.getEntries().get(0).getValue().equals(false) + )); var msgCaptor = ArgumentCaptor.forClass(TbMsg.class); then(clusterService).should().pushMsgToRuleEngine(eq(tenantId), eq(deviceId), msgCaptor.capture(), any()); @@ -971,72 +1002,74 @@ public class DefaultDeviceStateServiceTest { assertThat(actualNotification.getDeviceId()).isEqualTo(deviceId); assertThat(actualNotification.isActive()).isFalse(); - then(telemetrySubscriptionService).should().saveAttrAndNotify( - eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), - eq(INACTIVITY_ALARM_TIME), eq(expectedLastInactivityAlarmTime), any() - ); + then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher) request -> + request.getTenantId().equals(TenantId.SYS_TENANT_ID) && request.getEntityId().equals(deviceId) && + request.getScope().equals(AttributeScope.SERVER_SCOPE) && + request.getEntries().get(0).getKey().equals(INACTIVITY_ALARM_TIME) && + request.getEntries().get(0).getValue().equals(expectedLastInactivityAlarmTime) + )); } } private static Stream provideParametersForUpdateInactivityStateIfExpired() { return Stream.of( - Arguments.of(false, 100, 70, 90, 70, 60, false, 90, false), + Arguments.of(false, 100, 70, 90, 70, 60, false, 90, false), - Arguments.of(false, 100, 40, 50, 70, 10, false, 50, false), + Arguments.of(false, 100, 40, 50, 70, 10, false, 50, false), - Arguments.of(false, 100, 25, 60, 75, 25, false, 60, false), + Arguments.of(false, 100, 25, 60, 75, 25, false, 60, false), - Arguments.of(false, 100, 60, 70, 10, 50, false, 70, false), + Arguments.of(false, 100, 60, 70, 10, 50, false, 70, false), - Arguments.of(false, 100, 10, 15, 90, 10, false, 15, false), + Arguments.of(false, 100, 10, 15, 90, 10, false, 15, false), - Arguments.of(false, 100, 0, 40, 75, 0, false, 40, false), + Arguments.of(false, 100, 0, 40, 75, 0, false, 40, false), - Arguments.of(true, 100, 90, 80, 80, 50, true, 80, false), + Arguments.of(true, 100, 90, 80, 80, 50, true, 80, false), - Arguments.of(true, 100, 95, 90, 10, 50, true, 90, false), + Arguments.of(true, 100, 95, 90, 10, 50, true, 90, false), - Arguments.of(true, 100, 10, 10, 90, 10, false, 100, true), + Arguments.of(true, 100, 10, 10, 90, 10, false, 100, true), - Arguments.of(true, 100, 10, 10, 90, 11, true, 10, false), + Arguments.of(true, 100, 10, 10, 90, 11, true, 10, false), - Arguments.of(true, 100, 15, 10, 85, 5, false, 100, true), + Arguments.of(true, 100, 15, 10, 85, 5, false, 100, true), - Arguments.of(true, 100, 15, 10, 75, 5, false, 100, true), + Arguments.of(true, 100, 15, 10, 75, 5, false, 100, true), - Arguments.of(true, 100, 95, 90, 5, 50, false, 100, true), + Arguments.of(true, 100, 95, 90, 5, 50, false, 100, true), - Arguments.of(true, 100, 0, 0, 101, 0, true, 0, false), + Arguments.of(true, 100, 0, 0, 101, 0, true, 0, false), - Arguments.of(true, 100, 0, 0, 100, 0, false, 100, true), + Arguments.of(true, 100, 0, 0, 100, 0, false, 100, true), - Arguments.of(true, 100, 0, 0, 99, 0, false, 100, true), + Arguments.of(true, 100, 0, 0, 99, 0, false, 100, true), - Arguments.of(true, 100, 0, 0, 120, 10, true, 0, false), + Arguments.of(true, 100, 0, 0, 120, 10, true, 0, false), - Arguments.of(true, 100, 50, 0, 100, 0, true, 0, false), + Arguments.of(true, 100, 50, 0, 100, 0, true, 0, false), - Arguments.of(true, 100, 10, 0, 91, 0, true, 0, false), + Arguments.of(true, 100, 10, 0, 91, 0, true, 0, false), - Arguments.of(true, 100, 90, 0, 10, 0, false, 100, true), + Arguments.of(true, 100, 90, 0, 10, 0, false, 100, true), - Arguments.of(true, 100, 100, 100, 1, 0, true, 100, false), + Arguments.of(true, 100, 100, 100, 1, 0, true, 100, false), - Arguments.of(true, 100, 100, 100, 100, 100, true, 100, false), + Arguments.of(true, 100, 100, 100, 100, 100, true, 100, false), - Arguments.of(false, 100, 59, 60, 30, 10, false, 60, false), + Arguments.of(false, 100, 59, 60, 30, 10, false, 60, false), - Arguments.of(true, 100, 60, 60, 30, 10, false, 100, true), + Arguments.of(true, 100, 60, 60, 30, 10, false, 100, true), - Arguments.of(true, 100, 61, 60, 30, 10, false, 100, true), + Arguments.of(true, 100, 61, 60, 30, 10, false, 100, true), - Arguments.of(true, 0, 0, 0, 1, 0, true, 0, false), + Arguments.of(true, 0, 0, 0, 1, 0, true, 0, false), - Arguments.of(true, 0, 0, 0, 0, 0, false, 0, true), + Arguments.of(true, 0, 0, 0, 0, 0, false, 0, true), - Arguments.of(true, 100, 90, 80, 20, 70, true, 80, false), + Arguments.of(true, 100, 90, 80, 20, 70, true, 80, false), - Arguments.of(true, 100, 80, 90, 30, 70, true, 90, false) + Arguments.of(true, 100, 80, 90, 30, 70, true, 90, false) ); } @@ -1100,7 +1133,10 @@ public class DefaultDeviceStateServiceTest { // THEN await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { assertThat(service.deviceStates.get(deviceId).getState().isActive()).isEqualTo(false); - then(telemetrySubscriptionService).should().saveAttrAndNotify(eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), eq(ACTIVITY_STATE), eq(false), any()); + then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher) request -> + request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) && + request.getEntries().get(0).getValue().equals(false) + )); }); } @@ -1127,10 +1163,31 @@ public class DefaultDeviceStateServiceTest { service.onDeviceActivity(tenantId, deviceId, currentTime); // THEN + ArgumentCaptor attributeRequestCaptor = ArgumentCaptor.forClass(AttributesSaveRequest.class); + then(telemetrySubscriptionService).should(times(2)).save(attributeRequestCaptor.capture()); + await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { assertThat(service.deviceStates.get(deviceId).getState().isActive()).isEqualTo(true); - then(telemetrySubscriptionService).should().saveAttrAndNotify(eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), eq(LAST_ACTIVITY_TIME), eq(currentTime), any()); - then(telemetrySubscriptionService).should().saveAttrAndNotify(eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), eq(ACTIVITY_STATE), eq(true), any()); + + assertThat(attributeRequestCaptor.getAllValues()).hasSize(2) + .anySatisfy(request -> { + assertThat(request.getTenantId()).isEqualTo(TenantId.SYS_TENANT_ID); + assertThat(request.getEntityId()).isEqualTo(deviceId); + assertThat(request.getScope()).isEqualTo(AttributeScope.SERVER_SCOPE); + assertThat(request.getEntries()).singleElement().satisfies(attributeKvEntry -> { + assertThat(attributeKvEntry.getKey()).isEqualTo(LAST_ACTIVITY_TIME); + assertThat(attributeKvEntry.getLongValue()).hasValue(currentTime); + }); + }) + .anySatisfy(request -> { + assertThat(request.getTenantId()).isEqualTo(TenantId.SYS_TENANT_ID); + assertThat(request.getEntityId()).isEqualTo(deviceId); + assertThat(request.getScope()).isEqualTo(AttributeScope.SERVER_SCOPE); + assertThat(request.getEntries()).singleElement().satisfies(attributeKvEntry -> { + assertThat(attributeKvEntry.getKey()).isEqualTo(ACTIVITY_STATE); + assertThat(attributeKvEntry.getBooleanValue()).hasValue(true); + }); + }); }); } @@ -1174,7 +1231,10 @@ public class DefaultDeviceStateServiceTest { // THEN await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { assertThat(service.deviceStates.get(deviceId).getState().isActive()).isEqualTo(true); - then(telemetrySubscriptionService).should().saveAttrAndNotify(eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), eq(ACTIVITY_STATE), eq(true), any()); + then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher) request -> + request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) && + request.getEntries().get(0).getValue().equals(true) + )); }); } diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java new file mode 100644 index 0000000000..c1b12081f4 --- /dev/null +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java @@ -0,0 +1,115 @@ +/** + * 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 com.google.common.util.concurrent.FutureCallback; +import lombok.AccessLevel; +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.Setter; +import lombok.ToString; +import org.thingsboard.server.common.data.AttributeScope; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; +import org.thingsboard.server.common.data.kv.KvEntry; + +import java.util.List; + +@Getter +@ToString +@AllArgsConstructor(access = AccessLevel.PRIVATE) +public class AttributesSaveRequest { + + private final TenantId tenantId; + private final EntityId entityId; + private final AttributeScope scope; + private final List entries; // todo: rename to attributes? same with timeseries + private final boolean notifyDevice; + @Setter + private FutureCallback callback; + + public static Builder builder() { + return new Builder(); + } + + public static class Builder { + + private TenantId tenantId; + private EntityId entityId; + private AttributeScope scope; + private List entries; + private boolean notifyDevice = true; + private FutureCallback callback; + + Builder() {} + + public Builder tenantId(TenantId tenantId) { + this.tenantId = tenantId; + return this; + } + + public Builder entityId(EntityId entityId) { + this.entityId = entityId; + return this; + } + + public Builder scope(AttributeScope scope) { + this.scope = scope; + return this; + } + + @Deprecated + public Builder scope(String scope) { + try { + this.scope = AttributeScope.valueOf(scope); + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException("Invalid attribute scope '" + scope + "'"); + } + return this; + } + + public Builder entries(List entries) { + this.entries = entries; + return this; + } + + public Builder entry(AttributeKvEntry entry) { + return entries(List.of(entry)); + } + + public Builder entry(KvEntry kvEntry) { + return entry(new BaseAttributeKvEntry(kvEntry, System.currentTimeMillis())); + } + + public Builder notifyDevice(boolean notifyDevice) { + this.notifyDevice = notifyDevice; + return this; + } + + public Builder callback(FutureCallback callback) { + this.callback = callback; + return this; + } + + public AttributesSaveRequest build() { + return new AttributesSaveRequest(tenantId, entityId, scope, entries, notifyDevice, callback); + } + + } + +} diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java index 9f9a928839..bdc966ad88 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java @@ -20,7 +20,6 @@ import com.google.common.util.concurrent.ListenableFuture; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.DeleteTsKvQuery; import org.thingsboard.server.common.data.kv.TsKvEntry; @@ -36,33 +35,11 @@ public interface RuleEngineTelemetryService { ListenableFuture saveAndNotify(TimeseriesSaveRequest request); - @Deprecated(since = "3.7.0") - void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List attributes, FutureCallback callback); - - void saveAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List attributes, FutureCallback callback); - - @Deprecated(since = "3.7.0") - void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List attributes, boolean notifyDevice, FutureCallback callback); - - void saveAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List attributes, boolean notifyDevice, FutureCallback callback); + void save(AttributesSaveRequest request); void saveLatestAndNotify(TenantId tenantId, EntityId entityId, List ts, FutureCallback callback); - ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, long value); - - ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, String value); - - ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, double value); - - ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, boolean value); - - void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, long value, FutureCallback callback); - - void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, String value, FutureCallback callback); - - void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, double value, FutureCallback callback); - - void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, boolean value, FutureCallback callback); + ListenableFuture saveAttrAndNotify(AttributesSaveRequest request); void deleteAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List keys, FutureCallback callback); diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java index bcf021d3ca..de19c2b88e 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java @@ -23,6 +23,8 @@ import lombok.Setter; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.kv.BasicTsKvEntry; +import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; import java.util.List; @@ -77,8 +79,11 @@ public class TimeseriesSaveRequest { } public Builder entry(TsKvEntry entry) { - this.entries = List.of(entry); - return this; + return entries(List.of(entry)); + } + + public Builder entry(KvEntry kvEntry) { + return entry(new BasicTsKvEntry(System.currentTimeMillis(), kvEntry)); } public Builder ttl(long ttl) { diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java index 26bb6d7a4c..e18ced7ee0 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java @@ -23,6 +23,7 @@ import com.google.gson.JsonPrimitive; import jakarta.annotation.Nullable; import lombok.extern.slf4j.Slf4j; import org.thingsboard.common.util.DonAsynchron; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.EmptyNodeConfiguration; import org.thingsboard.rule.engine.api.RuleNode; import org.thingsboard.rule.engine.api.TbContext; @@ -112,8 +113,13 @@ public class TbCopyAttributesToEntityViewNode implements TbNode { Set attributes = JsonConverter.convertToAttributes(JsonParser.parseString(msg.getData())); List filteredAttributes = attributes.stream().filter(attr -> attributeContainsInEntityView(scope, attr.getKey(), entityView)).collect(Collectors.toList()); - ctx.getTelemetryService().saveAndNotify(ctx.getTenantId(), entityView.getId(), scope, filteredAttributes, - getFutureCallback(ctx, msg, entityView)); + ctx.getTelemetryService().save(AttributesSaveRequest.builder() + .tenantId(ctx.getTenantId()) + .entityId(entityView.getId()) + .scope(scope) + .entries(filteredAttributes) + .callback(getFutureCallback(ctx, msg, entityView)) + .build()); } } } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java index 893b06bfec..6f61de3fd3 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java @@ -24,6 +24,7 @@ import net.objecthunter.exp4j.Expression; import net.objecthunter.exp4j.ExpressionBuilder; import org.springframework.util.ConcurrentReferenceHashMap; import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.RuleNode; import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbNode; @@ -38,6 +39,7 @@ import org.thingsboard.server.common.data.id.EntityId; 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.LongDataEntry; import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.common.msg.TbMsg; @@ -150,15 +152,20 @@ public class TbMathNode implements TbNode { private ListenableFuture saveAttribute(TbContext ctx, TbMsg msg, double result, TbMathResult mathResultDef) { AttributeScope attributeScope = getAttributeScope(mathResultDef.getAttributeScope()); + KvEntry kvEntry; if (isIntegerResult(mathResultDef, config.getOperation())) { var value = toIntValue(result); - return ctx.getTelemetryService().saveAttrAndNotify( - ctx.getTenantId(), msg.getOriginator(), attributeScope, mathResultDef.getKey(), value); + kvEntry = new LongDataEntry(mathResultDef.getKey(), value); } else { var value = toDoubleValue(mathResultDef, result); - return ctx.getTelemetryService().saveAttrAndNotify( - ctx.getTenantId(), msg.getOriginator(), attributeScope, mathResultDef.getKey(), value); + kvEntry = new DoubleDataEntry(mathResultDef.getKey(), value); } + return ctx.getTelemetryService().saveAttrAndNotify(AttributesSaveRequest.builder() + .tenantId(ctx.getTenantId()) + .entityId(msg.getOriginator()) + .scope(attributeScope) + .entry(kvEntry) + .build()); } private boolean isIntegerResult(TbMathResult mathResultDef, TbRuleNodeMathFunctionType function) { diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java index 40b82ba150..2250e0bbdd 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java @@ -17,11 +17,13 @@ package org.thingsboard.rule.engine.telemetry; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.node.ObjectNode; +import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; import com.google.gson.JsonParser; import lombok.extern.slf4j.Slf4j; import org.thingsboard.common.util.DonAsynchron; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.RuleNode; import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbNode; @@ -56,10 +58,10 @@ import static org.thingsboard.server.common.data.msg.TbMsgType.POST_ATTRIBUTES_R version = 2, nodeDescription = "Saves attributes data", nodeDetails = "Saves entity attributes based on configurable scope parameter. Expects messages with 'POST_ATTRIBUTES_REQUEST' message type. " + - "If upsert(update/insert) operation is completed successfully rule node will send the incoming message via Success chain, otherwise, Failure chain is used. " + - "Additionally if checkbox Send attributes updated notification is set to true, rule node will put the \"Attributes Updated\" " + - "event for SHARED_SCOPE and SERVER_SCOPE attributes updates to the corresponding rule engine queue." + - "Performance checkbox 'Save attributes only if the value changes' will skip attributes overwrites for values with no changes (avoid concurrent writes because this check is not transactional; will not update 'Last updated time' for skipped attributes).", + "If upsert(update/insert) operation is completed successfully rule node will send the incoming message via Success chain, otherwise, Failure chain is used. " + + "Additionally if checkbox Send attributes updated notification is set to true, rule node will put the \"Attributes Updated\" " + + "event for SHARED_SCOPE and SERVER_SCOPE attributes updates to the corresponding rule engine queue." + + "Performance checkbox 'Save attributes only if the value changes' will skip attributes overwrites for values with no changes (avoid concurrent writes because this check is not transactional; will not update 'Last updated time' for skipped attributes).", uiResources = {"static/rulenode/rulenode-core-config.js"}, configDirective = "tbActionNodeAttributesConfig", icon = "file_upload" @@ -114,16 +116,17 @@ public class TbMsgAttributesNode implements TbNode { ctx.tellSuccess(msg); return; } - ctx.getTelemetryService().saveAndNotify( - ctx.getTenantId(), - msg.getOriginator(), - scope, - attributes, - config.isNotifyDevice() || checkNotifyDeviceMdValue(msg.getMetaData().getValue(NOTIFY_DEVICE_METADATA_KEY)), - sendAttributesUpdateNotification ? - new AttributesUpdateNodeCallback(ctx, msg, scope.name(), attributes) : - new TelemetryNodeCallback(ctx, msg) - ); + FutureCallback callback = sendAttributesUpdateNotification ? + new AttributesUpdateNodeCallback(ctx, msg, scope.name(), attributes) : + new TelemetryNodeCallback(ctx, msg); + ctx.getTelemetryService().save(AttributesSaveRequest.builder() + .tenantId(ctx.getTenantId()) + .entityId(msg.getOriginator()) + .scope(scope) + .entries(attributes) + .notifyDevice(config.isNotifyDevice() || checkNotifyDeviceMdValue(msg.getMetaData().getValue(NOTIFY_DEVICE_METADATA_KEY))) + .callback(callback) + .build()); } List filterChangedAttr(List currentAttributes, List newAttributes) { diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNodeTest.java index 4166c05a7a..fcd6083eb2 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNodeTest.java @@ -24,8 +24,10 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.EnumSource; import org.mockito.ArgumentCaptor; import org.mockito.Mock; +import org.mockito.ThrowingConsumer; import org.mockito.junit.jupiter.MockitoExtension; import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.EmptyNodeConfiguration; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.TbContext; @@ -37,7 +39,6 @@ import org.thingsboard.server.common.data.EntityView; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EntityViewId; import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.data.msg.TbNodeConnectionType; import org.thingsboard.server.common.data.objects.AttributesEntityView; @@ -56,6 +57,7 @@ import java.util.UUID; import static org.assertj.core.api.AssertionsForClassTypes.assertThat; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.assertArg; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.verify; @@ -113,10 +115,10 @@ public class TbCopyAttributesToEntityViewNodeTest { mockEntityViewLookup(entityView); when(ctxMock.getTelemetryService()).thenReturn(telemetryServiceMock); doAnswer(invocation -> { - FutureCallback callback = invocation.getArgument(4); - callback.onSuccess(null); + AttributesSaveRequest request = invocation.getArgument(0); + request.getCallback().onSuccess(null); return null; - }).when(telemetryServiceMock).saveAndNotify(any(), any(), any(AttributeScope.class), anyList(), any(FutureCallback.class)); + }).when(telemetryServiceMock).save(any(AttributesSaveRequest.class)); TbMsg newMsg = TbMsg.newMsg(msg, msg.getQueueName(), msg.getRuleChainId(), msg.getRuleNodeId()); // TODO: use newMsg() with any(TbMsgType.class), replace in other tests as well. doAnswer(invocation -> newMsg).when(ctxMock).newMsg(any(), any(String.class), any(), any(), any(), any()); @@ -124,13 +126,15 @@ public class TbCopyAttributesToEntityViewNodeTest { node.onMsg(ctxMock, msg); verify(entityViewServiceMock).findEntityViewsByTenantIdAndEntityIdAsync(eq(TENANT_ID), eq(DEVICE_ID)); - ArgumentCaptor> filteredAttributesCaptor = ArgumentCaptor.forClass(List.class); - verify(telemetryServiceMock).saveAndNotify(eq(TENANT_ID), eq(ENTITY_VIEW_ID), eq(AttributeScope.CLIENT_SCOPE), - filteredAttributesCaptor.capture(), any(FutureCallback.class)); - List filteredAttributesCaptorValue = filteredAttributesCaptor.getValue(); - assertThat(filteredAttributesCaptorValue.size()).isEqualTo(1); - assertThat(filteredAttributesCaptorValue.get(0).getKey()).isEqualTo("clientAttribute1"); - assertThat(filteredAttributesCaptorValue.get(0).getValue()).isEqualTo(100L); + verify(telemetryServiceMock).save(assertArg((ThrowingConsumer) request -> { + assertThat(request.getTenantId()).isEqualTo(TENANT_ID); + assertThat(request.getEntityId()).isEqualTo(ENTITY_VIEW_ID); + assertThat(request.getScope()).isEqualTo(AttributeScope.CLIENT_SCOPE); + + assertThat(request.getEntries().size()).isEqualTo(1); + assertThat(request.getEntries().get(0).getKey()).isEqualTo("clientAttribute1"); + assertThat(request.getEntries().get(0).getValue()).isEqualTo(100L); + })); verify(ctxMock).ack(eq(msg)); verify(ctxMock).enqueueForTellNext(eq(newMsg), eq(TbNodeConnectionType.SUCCESS)); verifyNoMoreInteractions(ctxMock, entityViewServiceMock, telemetryServiceMock); @@ -195,17 +199,22 @@ public class TbCopyAttributesToEntityViewNodeTest { mockEntityViewLookup(entityView); when(ctxMock.getTelemetryService()).thenReturn(telemetryServiceMock); doAnswer(invocation -> { - FutureCallback callback = invocation.getArgument(4); - callback.onSuccess(null); + AttributesSaveRequest request = invocation.getArgument(0); + request.getCallback().onSuccess(null); return null; - }).when(telemetryServiceMock).saveAndNotify(any(), any(), any(AttributeScope.class), anyList(), any(FutureCallback.class)); + }).when(telemetryServiceMock).save(any(AttributesSaveRequest.class)); TbMsg newMsg = TbMsg.newMsg(msg, msg.getQueueName(), msg.getRuleChainId(), msg.getRuleNodeId()); doAnswer(invocation -> newMsg).when(ctxMock).newMsg(any(), any(String.class), any(), any(), any(), any()); node.onMsg(ctxMock, msg); verify(entityViewServiceMock).findEntityViewsByTenantIdAndEntityIdAsync(eq(TENANT_ID), eq(DEVICE_ID)); - verify(telemetryServiceMock).saveAndNotify(eq(TENANT_ID), eq(ENTITY_VIEW_ID), eq(AttributeScope.CLIENT_SCOPE), eq(Collections.emptyList()), any(FutureCallback.class)); + verify(telemetryServiceMock).save(assertArg((ThrowingConsumer) request -> { + assertThat(request.getTenantId()).isEqualTo(TENANT_ID); + assertThat(request.getEntityId()).isEqualTo(ENTITY_VIEW_ID); + assertThat(request.getScope()).isEqualTo(AttributeScope.CLIENT_SCOPE); + assertThat(request.getEntries().isEmpty()).isTrue(); + })); verify(ctxMock).ack(eq(msg)); verify(ctxMock).enqueueForTellNext(eq(newMsg), eq(TbNodeConnectionType.SUCCESS)); verifyNoMoreInteractions(ctxMock, entityViewServiceMock, telemetryServiceMock); diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/math/TbMathNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/math/TbMathNodeTest.java index 137a51d394..9104236be0 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/math/TbMathNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/math/TbMathNodeTest.java @@ -46,6 +46,7 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; 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.LongDataEntry; import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.msg.TbMsg; @@ -69,8 +70,6 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyDouble; -import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.ArgumentMatchers.assertArg; import static org.mockito.ArgumentMatchers.eq; @@ -435,14 +434,16 @@ public class TbMathNodeTest { TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, originator, TbMsgMetaData.EMPTY, JacksonUtil.newObjectNode().put("a", 5).toString()); - when(telemetryService.saveAttrAndNotify(any(), any(), any(AttributeScope.class), anyString(), anyDouble())) + when(telemetryService.saveAttrAndNotify(any())) .thenReturn(Futures.immediateFuture(null)); node.onMsg(ctx, msg); ArgumentCaptor msgCaptor = ArgumentCaptor.forClass(TbMsg.class); verify(ctx, timeout(TIMEOUT)).tellSuccess(msgCaptor.capture()); - verify(telemetryService, times(1)).saveAttrAndNotify(any(), any(), any(AttributeScope.class), anyString(), anyDouble()); + verify(telemetryService, times(1)).saveAttrAndNotify(assertArg(request -> { + assertThat(request.getEntries()).singleElement().extracting(KvEntry::getValue).isInstanceOf(Double.class); + })); TbMsg resultMsg = msgCaptor.getValue(); assertNotNull(resultMsg); @@ -554,7 +555,7 @@ public class TbMathNodeTest { new TbMathArgument(TbMathArgumentType.MESSAGE_BODY, "a") ); - TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, originator, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_ARRAY); + TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, originator, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_ARRAY); node.onMsg(ctx, msg); ArgumentCaptor tCaptor = ArgumentCaptor.forClass(Throwable.class); diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeTest.java index b5cf5533b8..74aec7b566 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeTest.java @@ -22,9 +22,10 @@ import org.junit.jupiter.api.Test; 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.ThrowingConsumer; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.rule.engine.AbstractRuleNodeUpgradeTest; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbNode; @@ -53,6 +54,7 @@ import java.util.stream.Stream; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.assertArg; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.BDDMockito.willCallRealMethod; import static org.mockito.Mockito.mock; @@ -169,17 +171,15 @@ class TbMsgAttributesNodeTest extends AbstractRuleNodeUpgradeTest { node.saveAttr(testAttrList, ctxMock, testTbMsg, AttributeScope.SHARED_SCOPE, false); - ArgumentCaptor notifyDeviceCaptor = ArgumentCaptor.forClass(Boolean.class); - - verify(telemetryServiceMock, times(1)).saveAndNotify( - eq(tenantId), eq(deviceId), eq(AttributeScope.SHARED_SCOPE), - eq(testAttrList), notifyDeviceCaptor.capture(), any() - ); - boolean notifyDevice = notifyDeviceCaptor.getValue(); - assertThat(notifyDevice).isEqualTo(expectedArgumentValue); + verify(telemetryServiceMock, times(1)).save(assertArg((ThrowingConsumer) request -> { + assertThat(request.getTenantId()).isEqualTo(tenantId); + assertThat(request.getEntityId()).isEqualTo(deviceId); + assertThat(request.getScope()).isEqualTo(AttributeScope.SHARED_SCOPE); + assertThat(request.getEntries()).isEqualTo(testAttrList); + assertThat(request.isNotifyDevice()).isEqualTo(expectedArgumentValue); + })); } - // Rule nodes upgrade private static Stream givenFromVersionAndConfig_whenUpgrade_thenVerifyHasChangesAndConfig() { return Stream.of( diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeTest.java index d80a0bd340..a66c27e06a 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeTest.java @@ -25,6 +25,7 @@ import org.junit.jupiter.params.provider.EnumSource; import org.junit.jupiter.params.provider.MethodSource; import org.mockito.ArgumentCaptor; import org.mockito.Mock; +import org.mockito.ThrowingConsumer; import org.mockito.junit.jupiter.MockitoExtension; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; @@ -129,12 +130,12 @@ public class TbMsgTimeseriesNodeTest { TimeseriesSaveRequest request = invocation.getArgument(0); request.getCallback().onSuccess(null); return null; - }).when(telemetryServiceMock).save(any()); + }).when(telemetryServiceMock).save(any(TimeseriesSaveRequest.class)); node.onMsg(ctxMock, msg); List expectedList = getTsKvEntriesListWithTs(data, System.currentTimeMillis()); - verify(telemetryServiceMock).save(assertArg(request -> { + verify(telemetryServiceMock).save(assertArg((ThrowingConsumer) request -> { assertThat(request.getTenantId()).isEqualTo(TENANT_ID); assertThat(request.getCustomerId()).isNull(); assertThat(request.getEntityId()).isEqualTo(DEVICE_ID); @@ -170,12 +171,12 @@ public class TbMsgTimeseriesNodeTest { TimeseriesSaveRequest request = invocation.getArgument(0); request.getCallback().onSuccess(null); return null; - }).when(telemetryServiceMock).save(any()); + }).when(telemetryServiceMock).save(any(TimeseriesSaveRequest.class)); node.onMsg(ctxMock, msg); List expectedList = getTsKvEntriesListWithTs(data, ts); - verify(telemetryServiceMock).save(assertArg(request -> { + verify(telemetryServiceMock).save(assertArg((ThrowingConsumer) request -> { assertThat(request.getTenantId()).isEqualTo(TENANT_ID); assertThat(request.getCustomerId()).isNull(); assertThat(request.getEntityId()).isEqualTo(DEVICE_ID); @@ -208,7 +209,7 @@ public class TbMsgTimeseriesNodeTest { TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, metadata, data); node.onMsg(ctxMock, msg); - verify(telemetryServiceMock).save(assertArg(request -> { + verify(telemetryServiceMock).save(assertArg((ThrowingConsumer) request -> { assertThat(request.getTenantId()).isEqualTo(TENANT_ID); assertThat(request.getCustomerId()).isNull(); assertThat(request.getEntityId()).isEqualTo(DEVICE_ID);