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 b1733a9af0..8c63be6de9 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.TimeseriesSaveRequest; import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.EntityType; @@ -399,7 +400,7 @@ public class TelemetryController extends BaseController { public DeferredResult saveEntityAttributesV2( @Parameter(description = ENTITY_TYPE_PARAM_DESCRIPTION, required = true, schema = @Schema(defaultValue = "DEVICE")) @PathVariable("entityType") String entityType, @Parameter(description = ENTITY_ID_PARAM_DESCRIPTION, required = true) @PathVariable("entityId") String entityIdStr, - @Parameter(description = ATTRIBUTES_SCOPE_DESCRIPTION, schema = @Schema(allowableValues = {"SERVER_SCOPE", "SHARED_SCOPE"}, requiredMode = Schema.RequiredMode.REQUIRED)) @PathVariable("scope")AttributeScope scope, + @Parameter(description = ATTRIBUTES_SCOPE_DESCRIPTION, schema = @Schema(allowableValues = {"SERVER_SCOPE", "SHARED_SCOPE"}, requiredMode = Schema.RequiredMode.REQUIRED)) @PathVariable("scope") AttributeScope scope, @io.swagger.v3.oas.annotations.parameters.RequestBody(description = ATTRIBUTES_JSON_REQUEST_DESCRIPTION, required = true) @RequestBody JsonNode request) throws ThingsboardException { EntityId entityId = EntityIdFactory.getByTypeAndId(entityType, entityIdStr); return saveAttributes(getTenantId(), entityId, scope, request); @@ -423,8 +424,8 @@ public class TelemetryController extends BaseController { public DeferredResult saveEntityTelemetry( @Parameter(description = ENTITY_TYPE_PARAM_DESCRIPTION, required = true, schema = @Schema(defaultValue = "DEVICE")) @PathVariable("entityType") String entityType, @Parameter(description = ENTITY_ID_PARAM_DESCRIPTION, required = true) @PathVariable("entityId") String entityIdStr, - @Parameter(description = TELEMETRY_SCOPE_DESCRIPTION, required = true, schema = @Schema(allowableValues = "ANY")) @PathVariable("scope")String scope, - @io.swagger.v3.oas.annotations.parameters.RequestBody(description = TELEMETRY_JSON_REQUEST_DESCRIPTION, required = true)@RequestBody String requestBody) throws ThingsboardException { + @Parameter(description = TELEMETRY_SCOPE_DESCRIPTION, required = true, schema = @Schema(allowableValues = "ANY")) @PathVariable("scope") String scope, + @io.swagger.v3.oas.annotations.parameters.RequestBody(description = TELEMETRY_JSON_REQUEST_DESCRIPTION, required = true) @RequestBody String requestBody) throws ThingsboardException { EntityId entityId = EntityIdFactory.getByTypeAndId(entityType, entityIdStr); return saveTelemetry(getTenantId(), entityId, requestBody, 0L); } @@ -447,9 +448,9 @@ public class TelemetryController extends BaseController { public DeferredResult saveEntityTelemetryWithTTL( @Parameter(description = ENTITY_TYPE_PARAM_DESCRIPTION, required = true, schema = @Schema(defaultValue = "DEVICE")) @PathVariable("entityType") String entityType, @Parameter(description = ENTITY_ID_PARAM_DESCRIPTION, required = true) @PathVariable("entityId") String entityIdStr, - @Parameter(description = TELEMETRY_SCOPE_DESCRIPTION, required = true, schema = @Schema(allowableValues = "ANY")) @PathVariable("scope")String scope, - @Parameter(description = "A long value representing TTL (Time to Live) parameter.", required = true)@PathVariable("ttl")Long ttl, - @io.swagger.v3.oas.annotations.parameters.RequestBody(description = TELEMETRY_JSON_REQUEST_DESCRIPTION, required = true)@RequestBody String requestBody) throws ThingsboardException { + @Parameter(description = TELEMETRY_SCOPE_DESCRIPTION, required = true, schema = @Schema(allowableValues = "ANY")) @PathVariable("scope") String scope, + @Parameter(description = "A long value representing TTL (Time to Live) parameter.", required = true) @PathVariable("ttl") Long ttl, + @io.swagger.v3.oas.annotations.parameters.RequestBody(description = TELEMETRY_JSON_REQUEST_DESCRIPTION, required = true) @RequestBody String requestBody) throws ThingsboardException { EntityId entityId = EntityIdFactory.getByTypeAndId(entityType, entityIdStr); return saveTelemetry(getTenantId(), entityId, requestBody, ttl); } @@ -550,8 +551,8 @@ public class TelemetryController extends BaseController { @ResponseBody public DeferredResult deleteDeviceAttributes( @Parameter(description = DEVICE_ID_PARAM_DESCRIPTION, required = true) @PathVariable(DEVICE_ID) String deviceIdStr, - @Parameter(description = ATTRIBUTES_SCOPE_DESCRIPTION, schema = @Schema(allowableValues = {"SERVER_SCOPE", "SHARED_SCOPE", "CLIENT_SCOPE"}, requiredMode = Schema.RequiredMode.REQUIRED)) @PathVariable("scope")AttributeScope scope, - @Parameter(description = ATTRIBUTES_KEYS_DESCRIPTION, required = true)@RequestParam(name = "keys")String keysStr) throws ThingsboardException { + @Parameter(description = ATTRIBUTES_SCOPE_DESCRIPTION, schema = @Schema(allowableValues = {"SERVER_SCOPE", "SHARED_SCOPE", "CLIENT_SCOPE"}, requiredMode = Schema.RequiredMode.REQUIRED)) @PathVariable("scope") AttributeScope scope, + @Parameter(description = ATTRIBUTES_KEYS_DESCRIPTION, required = true) @RequestParam(name = "keys") String keysStr) throws ThingsboardException { EntityId entityId = EntityIdFactory.getByTypeAndUuid(EntityType.DEVICE, deviceIdStr); return deleteAttributes(entityId, scope, keysStr); } @@ -573,8 +574,8 @@ public class TelemetryController extends BaseController { public DeferredResult deleteEntityAttributes( @Parameter(description = ENTITY_TYPE_PARAM_DESCRIPTION, required = true, schema = @Schema(defaultValue = "DEVICE")) @PathVariable("entityType") String entityType, @Parameter(description = ENTITY_ID_PARAM_DESCRIPTION, required = true) @PathVariable("entityId") String entityIdStr, - @Parameter(description = ATTRIBUTES_SCOPE_DESCRIPTION, required = true, schema = @Schema(allowableValues = {"SERVER_SCOPE", "SHARED_SCOPE", "CLIENT_SCOPE"})) @PathVariable("scope")AttributeScope scope, - @Parameter(description = ATTRIBUTES_KEYS_DESCRIPTION, required = true)@RequestParam(name = "keys")String keysStr) throws ThingsboardException { + @Parameter(description = ATTRIBUTES_SCOPE_DESCRIPTION, required = true, schema = @Schema(allowableValues = {"SERVER_SCOPE", "SHARED_SCOPE", "CLIENT_SCOPE"})) @PathVariable("scope") AttributeScope scope, + @Parameter(description = ATTRIBUTES_KEYS_DESCRIPTION, required = true) @RequestParam(name = "keys") String keysStr) throws ThingsboardException { EntityId entityId = EntityIdFactory.getByTypeAndId(entityType, entityIdStr); return deleteAttributes(entityId, scope, keysStr); } @@ -587,7 +588,7 @@ public class TelemetryController extends BaseController { SecurityUser user = getCurrentUser(); return accessValidator.validateEntityAndCallback(getCurrentUser(), Operation.WRITE_ATTRIBUTES, entityIdSrc, (result, tenantId, entityId) -> { - tsSubService.deleteAndNotify(tenantId, entityId, scope.name(), keys, new FutureCallback() { + tsSubService.deleteAndNotify(tenantId, entityId, scope, keys, new FutureCallback() { @Override public void onSuccess(@Nullable Void tmp) { logAttributesDeleted(user, entityId, scope, keys, null); @@ -672,19 +673,27 @@ public class TelemetryController extends BaseController { TenantProfile tenantProfile = tenantProfileCache.get(tenantId); tenantTtl = TimeUnit.DAYS.toSeconds(((DefaultTenantProfileConfiguration) tenantProfile.getProfileData().getConfiguration()).getDefaultStorageTtlDays()); } - tsSubService.saveAndNotify(tenantId, user.getCustomerId(), entityId, entries, tenantTtl, new FutureCallback() { - @Override - public void onSuccess(@Nullable Void tmp) { - logTelemetryUpdated(user, entityId, entries, null); - result.setResult(new ResponseEntity(HttpStatus.OK)); - } - - @Override - public void onFailure(Throwable t) { - logTelemetryUpdated(user, entityId, entries, t); - AccessValidator.handleError(t, result, HttpStatus.INTERNAL_SERVER_ERROR); - } - }); + tsSubService.save(TimeseriesSaveRequest.builder() + .tenantId(tenantId) + .customerId(user.getCustomerId()) + .entityId(entityId) + .entries(entries) + .ttl(tenantTtl) + .saveLatest(true) + .callback(new FutureCallback() { + @Override + public void onSuccess(@Nullable Void tmp) { + logTelemetryUpdated(user, entityId, entries, null); + result.setResult(new ResponseEntity(HttpStatus.OK)); + } + + @Override + public void onFailure(Throwable t) { + logTelemetryUpdated(user, entityId, entries, t); + AccessValidator.handleError(t, result, HttpStatus.INTERNAL_SERVER_ERROR); + } + }) + .build()); }); } 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 71387aa542..d451bb36b3 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.TimeseriesSaveRequest; import org.thingsboard.server.cache.TbTransactionalCache; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.AttributeScope; @@ -503,10 +504,13 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i private void save(TenantId tenantId, EdgeId edgeId, String key, long value) { log.debug("[{}][{}] Updating long edge telemetry [{}] [{}]", tenantId, edgeId, key, value); if (persistToTelemetry) { - tsSubService.saveAndNotify( - tenantId, edgeId, - Collections.singletonList(new BasicTsKvEntry(System.currentTimeMillis(), new LongDataEntry(key, value))), - new AttributeSaveCallback(tenantId, edgeId, key, value)); + tsSubService.save(TimeseriesSaveRequest.builder() + .tenantId(tenantId) + .entityId(edgeId) + .entries(Collections.singletonList(new BasicTsKvEntry(System.currentTimeMillis(), new LongDataEntry(key, value)))) + .saveLatest(true) + .callback(new AttributeSaveCallback(tenantId, edgeId, key, value)) + .build()); } else { tsSubService.saveAttrAndNotify(tenantId, edgeId, AttributeScope.SERVER_SCOPE, key, value, new AttributeSaveCallback(tenantId, edgeId, key, value)); } @@ -515,10 +519,13 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i private void save(TenantId tenantId, EdgeId edgeId, String key, boolean value) { log.debug("[{}][{}] Updating boolean edge telemetry [{}] [{}]", tenantId, edgeId, key, value); if (persistToTelemetry) { - tsSubService.saveAndNotify( - tenantId, edgeId, - Collections.singletonList(new BasicTsKvEntry(System.currentTimeMillis(), new BooleanDataEntry(key, value))), - new AttributeSaveCallback(tenantId, edgeId, key, value)); + tsSubService.save(TimeseriesSaveRequest.builder() + .tenantId(tenantId) + .entityId(edgeId) + .entries(Collections.singletonList(new BasicTsKvEntry(System.currentTimeMillis(), new BooleanDataEntry(key, value)))) + .saveLatest(true) + .callback(new AttributeSaveCallback(tenantId, edgeId, key, value)) + .build()); } else { tsSubService.saveAttrAndNotify(tenantId, edgeId, AttributeScope.SERVER_SCOPE, key, value, new AttributeSaveCallback(tenantId, edgeId, key, value)); } 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 54ca38ad78..78da4eee7f 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 @@ -21,6 +21,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; +import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.DataConstants; @@ -128,7 +129,7 @@ public class DefaultOtaPackageStateService implements OtaPackageStateService { // Device was updated and new firmware is different from previous firmware. send(device.getTenantId(), device.getId(), newFirmwareId, System.currentTimeMillis(), FIRMWARE); } - } else if (oldFirmwareId != null){ + } else if (oldFirmwareId != null) { // Device was updated and new firmware is not set. remove(device, FIRMWARE); } @@ -155,7 +156,7 @@ public class DefaultOtaPackageStateService implements OtaPackageStateService { // Device was updated and new firmware is different from previous firmware. send(device.getTenantId(), device.getId(), newSoftwareId, System.currentTimeMillis(), SOFTWARE); } - } else if (oldSoftwareId != null){ + } else if (oldSoftwareId != null) { // Device was updated and new firmware is not set. remove(device, SOFTWARE); } @@ -261,17 +262,23 @@ public class DefaultOtaPackageStateService implements OtaPackageStateService { telemetry.add(new BasicTsKvEntry(ts, new LongDataEntry(getTargetTelemetryKey(firmware.getType(), TS), ts))); telemetry.add(new BasicTsKvEntry(ts, new StringDataEntry(getTelemetryKey(firmware.getType(), STATE), OtaPackageUpdateStatus.QUEUED.name()))); - telemetryService.saveAndNotify(tenantId, deviceId, telemetry, new FutureCallback<>() { - @Override - public void onSuccess(@Nullable Void tmp) { - log.trace("[{}] Success save firmware status!", deviceId); - } + telemetryService.save(TimeseriesSaveRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .entries(telemetry) + .saveLatest(true) + .callback(new FutureCallback() { + @Override + public void onSuccess(@Nullable Void tmp) { + log.trace("[{}] Success save firmware status!", deviceId); + } - @Override - public void onFailure(Throwable t) { - log.error("[{}] Failed to save firmware status!", deviceId, t); - } - }); + @Override + public void onFailure(Throwable t) { + log.error("[{}] Failed to save firmware status!", deviceId, t); + } + }) + .build()); } @@ -282,19 +289,25 @@ public class DefaultOtaPackageStateService implements OtaPackageStateService { BasicTsKvEntry status = new BasicTsKvEntry(System.currentTimeMillis(), new StringDataEntry(getTelemetryKey(otaPackageType, STATE), OtaPackageUpdateStatus.INITIATED.name())); - telemetryService.saveAndNotify(tenantId, deviceId, Collections.singletonList(status), new FutureCallback<>() { - @Override - public void onSuccess(@Nullable Void tmp) { - log.trace("[{}] Success save telemetry with target {} for device!", deviceId, otaPackage); - updateAttributes(device, otaPackage, ts, tenantId, deviceId, otaPackageType); - } + telemetryService.save(TimeseriesSaveRequest.builder() + .tenantId(tenantId) + .entityId(deviceId) + .entries(Collections.singletonList(status)) + .saveLatest(true) + .callback(new FutureCallback() { + @Override + public void onSuccess(@Nullable Void tmp) { + log.trace("[{}] Success save telemetry with target {} for device!", deviceId, otaPackage); + updateAttributes(device, otaPackage, ts, tenantId, deviceId, otaPackageType); + } - @Override - public void onFailure(Throwable t) { - log.error("[{}] Failed to save telemetry with target {} for device!", deviceId, otaPackage, t); - updateAttributes(device, otaPackage, ts, tenantId, deviceId, otaPackageType); - } - }); + @Override + public void onFailure(Throwable t) { + log.error("[{}] Failed to save telemetry with target {} for device!", deviceId, otaPackage, t); + updateAttributes(device, otaPackage, ts, tenantId, deviceId, otaPackageType); + } + }) + .build()); } private void updateAttributes(Device device, OtaPackageInfo otaPackage, long ts, TenantId tenantId, DeviceId deviceId, OtaPackageType otaPackageType) { @@ -368,4 +381,5 @@ public class DefaultOtaPackageStateService implements OtaPackageStateService { } }); } + } 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 50d8b9c371..9f7d4c0843 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.TimeseriesSaveRequest; import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.EntityType; @@ -206,20 +207,28 @@ public abstract class AbstractBulkImportService { TenantProfile tenantProfile = tenantProfileCache.get(tenantId); long tenantTtl = TimeUnit.DAYS.toSeconds(((DefaultTenantProfileConfiguration) tenantProfile.getProfileData().getConfiguration()).getDefaultStorageTtlDays()); - tsSubscriptionService.saveAndNotify(tenantId, user.getCustomerId(), entityId, timeseries, tenantTtl, new FutureCallback() { - @Override - public void onSuccess(@Nullable Void tmp) { - entityActionService.logEntityAction(user, (UUIDBased & EntityId) entityId, null, null, - ActionType.TIMESERIES_UPDATED, null, timeseries); - } - - @Override - public void onFailure(Throwable t) { - entityActionService.logEntityAction(user, (UUIDBased & EntityId) entityId, null, null, - ActionType.TIMESERIES_UPDATED, BaseController.toException(t), timeseries); - throw new RuntimeException(t); - } - }); + tsSubscriptionService.save(TimeseriesSaveRequest.builder() + .tenantId(tenantId) + .customerId(user.getCustomerId()) + .entityId(entityId) + .entries(timeseries) + .ttl(tenantTtl) + .saveLatest(true) + .callback(new FutureCallback() { + @Override + public void onSuccess(@Nullable Void tmp) { + entityActionService.logEntityAction(user, (UUIDBased & EntityId) entityId, null, null, + ActionType.TIMESERIES_UPDATED, null, timeseries); + } + + @Override + public void onFailure(Throwable t) { + entityActionService.logEntityAction(user, (UUIDBased & EntityId) entityId, null, null, + ActionType.TIMESERIES_UPDATED, BaseController.toException(t), timeseries); + throw new RuntimeException(t); + } + }) + .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 5513c41929..682bddd436 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,8 @@ 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.RuleEngineTelemetryService; +import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.common.data.ApiUsageRecordKey; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.EntityType; @@ -70,7 +72,7 @@ import java.util.concurrent.Executors; */ @Service @Slf4j -public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionService implements TelemetrySubscriptionService { +public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionService implements TelemetrySubscriptionService, RuleEngineTelemetryService { private final AttributesService attrService; private final TimeseriesService tsService; @@ -115,39 +117,29 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer } @Override - public ListenableFuture saveAndNotify(TenantId tenantId, EntityId entityId, TsKvEntry ts) { + public ListenableFuture saveAndNotify(TimeseriesSaveRequest request) { SettableFuture future = SettableFuture.create(); - saveAndNotify(tenantId, entityId, Collections.singletonList(ts), new VoidFutureCallback(future)); + request.setCallback(new VoidFutureCallback(future)); + save(request); return future; } @Override - public void saveAndNotify(TenantId tenantId, EntityId entityId, List ts, FutureCallback callback) { - saveAndNotify(tenantId, null, entityId, ts, 0L, callback); - } - - @Override - public void saveAndNotify(TenantId tenantId, CustomerId customerId, EntityId entityId, List ts, long ttl, FutureCallback callback) { - doSaveAndNotify(tenantId, customerId, entityId, ts, ttl, callback, true); - } - - @Override - public void saveWithoutLatestAndNotify(TenantId tenantId, CustomerId customerId, EntityId entityId, List ts, long ttl, FutureCallback callback) { - doSaveAndNotify(tenantId, customerId, entityId, ts, ttl, callback, false); - } - - private void doSaveAndNotify(TenantId tenantId, CustomerId customerId, EntityId entityId, List ts, long ttl, FutureCallback callback, boolean saveLatest) { + public void save(TimeseriesSaveRequest request) { + TenantId tenantId = request.getTenantId(); + EntityId entityId = request.getEntityId(); checkInternalEntity(entityId); boolean sysTenant = TenantId.SYS_TENANT_ID.equals(tenantId) || tenantId == null; if (sysTenant || apiUsageStateService.getApiUsageState(tenantId).isDbStorageEnabled()) { - KvUtils.validate(ts, valueNoXssValidation); - if (saveLatest) { - saveAndNotifyInternal(tenantId, entityId, ts, ttl, getCallback(tenantId, customerId, sysTenant, callback)); + KvUtils.validate(request.getEntries(), valueNoXssValidation); + FutureCallback callback = getCallback(tenantId, request.getCustomerId(), sysTenant, request.getCallback()); + if (request.isSaveLatest()) { + saveAndNotifyInternal(tenantId, entityId, request.getEntries(), request.getTtl(), callback); } else { - saveWithoutLatestAndNotifyInternal(tenantId, entityId, ts, ttl, getCallback(tenantId, customerId, sysTenant, callback)); + saveWithoutLatestAndNotifyInternal(tenantId, entityId, request.getEntries(), request.getTtl(), callback); } } else { - callback.onFailure(new RuntimeException("DB storage writes are disabled due to API limits!")); + request.getCallback().onFailure(new RuntimeException("DB storage writes are disabled due to API limits!")); } } @@ -285,24 +277,12 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer addWsCallback(saveFuture, success -> onTimeSeriesUpdate(tenantId, entityId, ts)); } - @Override - public void deleteAndNotify(TenantId tenantId, EntityId entityId, String scope, List keys, FutureCallback callback) { - checkInternalEntity(entityId); - deleteAndNotifyInternal(tenantId, entityId, scope, keys, false, callback); - } - @Override public void deleteAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List keys, FutureCallback callback) { checkInternalEntity(entityId); deleteAndNotifyInternal(tenantId, entityId, scope, keys, false, callback); } - @Override - public void deleteAndNotify(TenantId tenantId, EntityId entityId, String scope, List keys, boolean notifyDevice, FutureCallback callback) { - checkInternalEntity(entityId); - deleteAndNotifyInternal(tenantId, entityId, scope, keys, notifyDevice, callback); - } - @Override public void deleteAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List keys, boolean notifyDevice, FutureCallback callback) { checkInternalEntity(entityId); @@ -358,12 +338,6 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer addWsCallback(deleteFuture, list -> onTimeSeriesDelete(tenantId, entityId, keys, list)); } - @Override - public void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String 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, long value, FutureCallback callback) { @@ -371,49 +345,24 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer , System.currentTimeMillis())), callback); } - @Override - public void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String 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, 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, String 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, 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, String scope, String key, boolean value, FutureCallback callback) { - saveAndNotify(tenantId, entityId, scope, Collections.singletonList(new BaseAttributeKvEntry(new BooleanDataEntry(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, String scope, String key, long value) { - SettableFuture future = SettableFuture.create(); - saveAttrAndNotify(tenantId, entityId, scope, key, value, new VoidFutureCallback(future)); - return future; - } - @Override public ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, long value) { SettableFuture future = SettableFuture.create(); @@ -421,13 +370,6 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer return future; } - @Override - public ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, String value) { - SettableFuture future = SettableFuture.create(); - saveAttrAndNotify(tenantId, entityId, scope, key, value, new VoidFutureCallback(future)); - return future; - } - @Override public ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, String value) { SettableFuture future = SettableFuture.create(); @@ -435,13 +377,6 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer return future; } - @Override - public ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, double value) { - SettableFuture future = SettableFuture.create(); - saveAttrAndNotify(tenantId, entityId, scope, key, value, new VoidFutureCallback(future)); - return future; - } - @Override public ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, double value) { SettableFuture future = SettableFuture.create(); @@ -449,13 +384,6 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer return future; } - @Override - public ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, boolean value) { - SettableFuture future = SettableFuture.create(); - saveAttrAndNotify(tenantId, entityId, scope, key, value, new VoidFutureCallback(future)); - return future; - } - @Override public ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, boolean value) { SettableFuture future = SettableFuture.create(); @@ -560,6 +488,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer public void onFailure(Throwable t) { future.setException(t); } + } } 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 e3695d7efb..e32d6cbe5a 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.TimeseriesSaveRequest; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.alarm.AlarmSeverity; @@ -804,19 +805,27 @@ public class WebsocketApiTest extends AbstractControllerTest { private void sendTelemetry(Device device, List tsData) throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); - tsService.saveAndNotify(device.getTenantId(), null, device.getId(), tsData, 0, new FutureCallback() { - @Override - public void onSuccess(@Nullable Void result) { - log.debug("sendTelemetry callback onSuccess"); - latch.countDown(); - } - - @Override - public void onFailure(Throwable t) { - log.error("Failed to send telemetry", t); - latch.countDown(); - } - }); + tsService.save(TimeseriesSaveRequest.builder() + .tenantId(device.getTenantId()) + .customerId(null) + .entityId(device.getId()) + .entries(tsData) + .ttl(0) + .saveLatest(true) + .callback(new FutureCallback() { + @Override + public void onSuccess(@Nullable Void result) { + log.debug("sendTelemetry callback onSuccess"); + latch.countDown(); + } + + @Override + public void onFailure(Throwable t) { + log.error("Failed to send telemetry", t); + latch.countDown(); + } + }) + .build()); assertThat(latch.await(TIMEOUT, TimeUnit.SECONDS)).as("await sendTelemetry callback"); } @@ -841,4 +850,5 @@ public class WebsocketApiTest extends AbstractControllerTest { }); assertThat(latch.await(TIMEOUT, TimeUnit.SECONDS)).as("await sendAttributes callback").isTrue(); } + } 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 0c2de91b2f..9f9a928839 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 @@ -18,7 +18,6 @@ package org.thingsboard.rule.engine.api; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.ListenableFuture; import org.thingsboard.server.common.data.AttributeScope; -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; @@ -33,13 +32,9 @@ import java.util.List; */ public interface RuleEngineTelemetryService { - ListenableFuture saveAndNotify(TenantId tenantId, EntityId entityId, TsKvEntry ts); + void save(TimeseriesSaveRequest request); - void saveAndNotify(TenantId tenantId, EntityId entityId, List ts, FutureCallback callback); - - void saveAndNotify(TenantId tenantId, CustomerId id, EntityId entityId, List ts, long ttl, FutureCallback callback); - - void saveWithoutLatestAndNotify(TenantId tenantId, CustomerId id, EntityId entityId, List ts, long ttl, FutureCallback callback); + ListenableFuture saveAndNotify(TimeseriesSaveRequest request); @Deprecated(since = "3.7.0") void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List attributes, FutureCallback callback); @@ -53,54 +48,24 @@ public interface RuleEngineTelemetryService { void saveLatestAndNotify(TenantId tenantId, EntityId entityId, List ts, FutureCallback callback); - @Deprecated(since = "3.7.0") - ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, long value); - ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, long value); - @Deprecated(since = "3.7.0") - ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, String value); - ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, String value); - @Deprecated(since = "3.7.0") - ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, double value); - ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, double value); - @Deprecated(since = "3.7.0") - ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, boolean value); - ListenableFuture saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, boolean value); - @Deprecated(since = "3.7.0") - void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, long value, FutureCallback callback); - void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, long value, FutureCallback callback); - @Deprecated(since = "3.7.0") - void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, String value, FutureCallback callback); - void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, String value, FutureCallback callback); - @Deprecated(since = "3.7.0") - void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, double value, FutureCallback callback); - void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, double value, FutureCallback callback); - @Deprecated(since = "3.7.0") - void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, boolean value, FutureCallback callback); - void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, boolean value, FutureCallback callback); - @Deprecated(since = "3.7.0") - void deleteAndNotify(TenantId tenantId, EntityId entityId, String scope, List keys, FutureCallback callback); - void deleteAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List keys, FutureCallback callback); - @Deprecated(since = "3.7.0") - void deleteAndNotify(TenantId tenantId, EntityId entityId, String scope, List keys, boolean notifyDevice, FutureCallback callback); - void deleteAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List keys, boolean notifyDevice, FutureCallback callback); void deleteLatest(TenantId tenantId, EntityId entityId, List keys, FutureCallback callback); @@ -108,4 +73,5 @@ public interface RuleEngineTelemetryService { void deleteAllLatest(TenantId tenantId, EntityId entityId, FutureCallback> callback); void deleteTimeseriesAndNotify(TenantId tenantId, EntityId entityId, List keys, List deleteTsKvQueries, 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 new file mode 100644 index 0000000000..042915c95b --- /dev/null +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java @@ -0,0 +1,102 @@ +/** + * 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.Data; +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.TsKvEntry; + +import java.util.List; + +@Data +@AllArgsConstructor(access = AccessLevel.PRIVATE) +public class TimeseriesSaveRequest { + private final TenantId tenantId; + private final CustomerId customerId; + private final EntityId entityId; + private final List entries; + private final long ttl; + private final boolean saveLatest; + private FutureCallback callback; + + public static Builder builder() { + return new Builder(); + } + + public static class Builder { + + private TenantId tenantId; + private CustomerId customerId; + private EntityId entityId; + private List entries; + private long ttl; + private FutureCallback callback; + private boolean saveLatest; + + Builder() {} + + public Builder tenantId(TenantId tenantId) { + this.tenantId = tenantId; + return this; + } + + public Builder customerId(CustomerId customerId) { + this.customerId = customerId; + return this; + } + + public Builder entityId(EntityId entityId) { + this.entityId = entityId; + return this; + } + + public Builder entries(List entries) { + this.entries = entries; + return this; + } + + public Builder entry(TsKvEntry entry) { + this.entries = List.of(entry); + return this; + } + + public Builder ttl(long ttl) { + this.ttl = ttl; + return this; + } + + public Builder saveLatest(boolean saveLatest) { + this.saveLatest = saveLatest; + return this; + } + + public Builder callback(FutureCallback callback) { + this.callback = callback; + return this; + } + + public TimeseriesSaveRequest build() { + return new TimeseriesSaveRequest(tenantId, customerId, entityId, entries, ttl, saveLatest, callback); + } + + } + +} 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 4e64ad854e..83d82534a7 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 @@ -29,6 +29,7 @@ import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbNode; import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeException; +import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.rule.engine.api.util.TbNodeUtils; import org.thingsboard.rule.engine.util.SemaphoreWithTbMsgQueue; import org.thingsboard.server.common.data.AttributeScope; @@ -42,6 +43,7 @@ import org.thingsboard.server.common.msg.TbMsg; import java.math.BigDecimal; import java.math.RoundingMode; +import java.util.Collections; import java.util.List; import java.util.Optional; import java.util.concurrent.ConcurrentMap; @@ -139,9 +141,13 @@ public class TbMathNode implements TbNode { } private ListenableFuture saveTimeSeries(TbContext ctx, TbMsg msg, double result, TbMathResult mathResultDef) { - - return ctx.getTelemetryService().saveAndNotify(ctx.getTenantId(), msg.getOriginator(), - new BasicTsKvEntry(System.currentTimeMillis(), new DoubleDataEntry(mathResultDef.getKey(), result))); + final BasicTsKvEntry basicTsKvEntry = new BasicTsKvEntry(System.currentTimeMillis(), new DoubleDataEntry(mathResultDef.getKey(), result)); + return ctx.getTelemetryService().saveAndNotify(TimeseriesSaveRequest.builder() + .tenantId(ctx.getTenantId()) + .entityId(msg.getOriginator()) + .entries(Collections.singletonList(basicTsKvEntry)) + .saveLatest(true) + .build()); } private ListenableFuture saveAttribute(TbContext ctx, TbMsg msg, double result, TbMathResult mathResultDef) { diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java index c90e8e2375..b0d5e23b50 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java @@ -22,6 +22,7 @@ import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbNode; import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeException; +import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.rule.engine.api.util.TbNodeUtils; import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.data.StringUtils; @@ -104,11 +105,15 @@ public class TbMsgTimeseriesNode implements TbNode { if (ttl == 0L) { ttl = tenantProfileDefaultStorageTtl; } - if (config.isSkipLatestPersistence()) { - ctx.getTelemetryService().saveWithoutLatestAndNotify(ctx.getTenantId(), msg.getCustomerId(), msg.getOriginator(), tsKvEntryList, ttl, new TelemetryNodeCallback(ctx, msg)); - } else { - ctx.getTelemetryService().saveAndNotify(ctx.getTenantId(), msg.getCustomerId(), msg.getOriginator(), tsKvEntryList, ttl, new TelemetryNodeCallback(ctx, msg)); - } + ctx.getTelemetryService().save(TimeseriesSaveRequest.builder() + .tenantId(ctx.getTenantId()) + .customerId(msg.getCustomerId()) + .entityId(msg.getOriginator()) + .entries(tsKvEntryList) + .ttl(ttl) + .saveLatest(!config.isSkipLatestPersistence()) + .callback(new TelemetryNodeCallback(ctx, msg)) + .build()); } public static long computeTs(TbMsg msg, boolean ignoreMetadataTs) { 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 ba81fbad20..137a51d394 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 @@ -47,7 +47,6 @@ 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.LongDataEntry; -import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; @@ -73,6 +72,7 @@ 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; import static org.mockito.BDDMockito.willAnswer; import static org.mockito.BDDMockito.willReturn; @@ -460,14 +460,16 @@ public class TbMathNodeTest { ); TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, originator, TbMsgMetaData.EMPTY, JacksonUtil.newObjectNode().put("a", 5).toString()); - when(telemetryService.saveAndNotify(any(), any(), any(TsKvEntry.class))) - .thenReturn(Futures.immediateFuture(null)); + when(telemetryService.saveAndNotify(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)).saveAndNotify(any(), any(), any(TsKvEntry.class)); + verify(telemetryService, times(1)).saveAndNotify(assertArg(request -> { + assertThat(request.getEntries()).size().isOne(); + assertThat(request.isSaveLatest()).isTrue(); + })); TbMsg resultMsg = msgCaptor.getValue(); assertNotNull(resultMsg); @@ -485,14 +487,16 @@ public class TbMathNodeTest { ); TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, originator, TbMsgMetaData.EMPTY, JacksonUtil.newObjectNode().put("a", 5).toString()); - when(telemetryService.saveAndNotify(any(), any(), any(TsKvEntry.class))) - .thenReturn(Futures.immediateFuture(null)); + when(telemetryService.saveAndNotify(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)).saveAndNotify(any(), any(), any(TsKvEntry.class)); + verify(telemetryService, times(1)).saveAndNotify(assertArg(request -> { + assertThat(request.getEntries()).size().isOne(); + assertThat(request.isSaveLatest()).isTrue(); + })); TbMsg resultMsg = msgCaptor.getValue(); assertNotNull(resultMsg); 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 1fb4e9eefd..d80a0bd340 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 @@ -31,6 +31,7 @@ import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeException; +import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.data.TenantProfile; import org.thingsboard.server.common.data.id.DeviceId; @@ -54,10 +55,8 @@ import java.util.stream.Stream; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyList; -import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.assertArg; import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoMoreInteractions; @@ -127,19 +126,23 @@ public class TbMsgTimeseriesNodeTest { when(ctxMock.getTelemetryService()).thenReturn(telemetryServiceMock); when(ctxMock.getTenantId()).thenReturn(TENANT_ID); doAnswer(invocation -> { - TelemetryNodeCallback callback = invocation.getArgument(5); - callback.onSuccess(null); + TimeseriesSaveRequest request = invocation.getArgument(0); + request.getCallback().onSuccess(null); return null; - }).when(telemetryServiceMock).saveAndNotify(any(), any(), any(), anyList(), anyLong(), any()); + }).when(telemetryServiceMock).save(any()); node.onMsg(ctxMock, msg); List expectedList = getTsKvEntriesListWithTs(data, System.currentTimeMillis()); - ArgumentCaptor> entryListCaptor = ArgumentCaptor.forClass(List.class); - verify(telemetryServiceMock).saveAndNotify(eq(TENANT_ID), isNull(), eq(DEVICE_ID), entryListCaptor.capture(), - eq(tenantProfileDefaultStorageTtl), any(TelemetryNodeCallback.class)); - assertThat(entryListCaptor.getValue()).usingRecursiveFieldByFieldElementComparatorIgnoringFields("ts") - .containsExactlyElementsOf(expectedList); + verify(telemetryServiceMock).save(assertArg(request -> { + assertThat(request.getTenantId()).isEqualTo(TENANT_ID); + assertThat(request.getCustomerId()).isNull(); + assertThat(request.getEntityId()).isEqualTo(DEVICE_ID); + assertThat(request.getEntries()).usingRecursiveFieldByFieldElementComparatorIgnoringFields("ts").containsExactlyElementsOf(expectedList); + assertThat(request.getTtl()).isEqualTo(tenantProfileDefaultStorageTtl); + assertThat(request.isSaveLatest()).isTrue(); + assertThat(request.getCallback()).isInstanceOf(TelemetryNodeCallback.class); + })); verify(ctxMock).tellSuccess(msg); verifyNoMoreInteractions(ctxMock, telemetryServiceMock); } @@ -164,18 +167,23 @@ public class TbMsgTimeseriesNodeTest { when(ctxMock.getTelemetryService()).thenReturn(telemetryServiceMock); when(ctxMock.getTenantId()).thenReturn(TENANT_ID); doAnswer(invocation -> { - TelemetryNodeCallback callback = invocation.getArgument(5); - callback.onSuccess(null); + TimeseriesSaveRequest request = invocation.getArgument(0); + request.getCallback().onSuccess(null); return null; - }).when(telemetryServiceMock).saveWithoutLatestAndNotify(any(), any(), any(), anyList(), anyLong(), any()); + }).when(telemetryServiceMock).save(any()); node.onMsg(ctxMock, msg); List expectedList = getTsKvEntriesListWithTs(data, ts); - ArgumentCaptor> entryListCaptor = ArgumentCaptor.forClass(List.class); - verify(telemetryServiceMock).saveWithoutLatestAndNotify( - eq(TENANT_ID), isNull(), eq(DEVICE_ID), entryListCaptor.capture(), eq(ttlFromConfig), any(TelemetryNodeCallback.class)); - assertThat(entryListCaptor.getValue()).containsExactlyElementsOf(expectedList); + verify(telemetryServiceMock).save(assertArg(request -> { + assertThat(request.getTenantId()).isEqualTo(TENANT_ID); + assertThat(request.getCustomerId()).isNull(); + assertThat(request.getEntityId()).isEqualTo(DEVICE_ID); + assertThat(request.getEntries()).containsExactlyElementsOf(expectedList); + assertThat(request.getTtl()).isEqualTo(ttlFromConfig); + assertThat(request.isSaveLatest()).isFalse(); + assertThat(request.getCallback()).isInstanceOf(TelemetryNodeCallback.class); + })); verify(ctxMock).tellSuccess(msg); verifyNoMoreInteractions(ctxMock, telemetryServiceMock); } @@ -200,7 +208,14 @@ public class TbMsgTimeseriesNodeTest { TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, metadata, data); node.onMsg(ctxMock, msg); - verify(telemetryServiceMock).saveAndNotify(eq(TENANT_ID), isNull(), eq(DEVICE_ID), anyList(), eq(expectedTtl), any(TelemetryNodeCallback.class)); + verify(telemetryServiceMock).save(assertArg(request -> { + assertThat(request.getTenantId()).isEqualTo(TENANT_ID); + assertThat(request.getCustomerId()).isNull(); + assertThat(request.getEntityId()).isEqualTo(DEVICE_ID); + assertThat(request.getTtl()).isEqualTo(expectedTtl); + assertThat(request.isSaveLatest()).isTrue(); + assertThat(request.getCallback()).isInstanceOf(TelemetryNodeCallback.class); + })); } private static Stream givenTtlFromConfigAndTtlFromMd_whenOnMsg_thenVerifyTtl() {