tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Refactor saveAndNotify for timeseries

pull/12297/head
ViacheslavKlimov 2 years ago
parent
0a11b47a7b
commit
2bb65923dc
12 changed files with 317 additions and 241 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 57
      application/src/main/java/org/thingsboard/server/controller/TelemetryController.java
  2. 23
      application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
  3. 62
      application/src/main/java/org/thingsboard/server/service/ota/DefaultOtaPackageStateService.java
  4. 37
      application/src/main/java/org/thingsboard/server/service/sync/ie/importing/csv/AbstractBulkImportService.java
  5. 103
      application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java
  6. 36
      application/src/test/java/org/thingsboard/server/controller/WebsocketApiTest.java
  7. 40
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java
  8. 102
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java
  9. 12
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java
  10. 15
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java
  11. 18
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/math/TbMathNodeTest.java
  12. 53
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeTest.java

57
application/src/main/java/org/thingsboard/server/controller/TelemetryController.java
View File

@ -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<ResponseEntity> 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<ResponseEntity> 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<ResponseEntity> 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<ResponseEntity> 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<ResponseEntity> 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<Void>() {
tsSubService.deleteAndNotify(tenantId, entityId, scope, keys, new FutureCallback<Void>() {
@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<Void>() {
@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<Void>() {
@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());
});
}

23
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
View File

@ -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));
}

62
application/src/main/java/org/thingsboard/server/service/ota/DefaultOtaPackageStateService.java
View File

@ -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<Void>() {
@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<Void>() {
@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 {
}
});
}
}

37
application/src/main/java/org/thingsboard/server/service/sync/ie/importing/csv/AbstractBulkImportService.java
View File

@ -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<E extends HasId<? extends Entity
accessValidator.validateEntityAndCallback(user, Operation.WRITE_TELEMETRY, entity.getId(), (result, tenantId, entityId) -> {
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<Void>() {
@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<Void>() {
@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());
});
}

103
application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java
View File

@ -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<Void> saveAndNotify(TenantId tenantId, EntityId entityId, TsKvEntry ts) {
public ListenableFuture<Void> saveAndNotify(TimeseriesSaveRequest request) {
SettableFuture<Void> 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<TsKvEntry> ts, FutureCallback<Void> callback) {
saveAndNotify(tenantId, null, entityId, ts, 0L, callback);
}
@Override
public void saveAndNotify(TenantId tenantId, CustomerId customerId, EntityId entityId, List<TsKvEntry> ts, long ttl, FutureCallback<Void> callback) {
doSaveAndNotify(tenantId, customerId, entityId, ts, ttl, callback, true);
}
@Override
public void saveWithoutLatestAndNotify(TenantId tenantId, CustomerId customerId, EntityId entityId, List<TsKvEntry> ts, long ttl, FutureCallback<Void> callback) {
doSaveAndNotify(tenantId, customerId, entityId, ts, ttl, callback, false);
}
private void doSaveAndNotify(TenantId tenantId, CustomerId customerId, EntityId entityId, List<TsKvEntry> ts, long ttl, FutureCallback<Void> 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<Integer> 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<String> keys, FutureCallback<Void> callback) {
checkInternalEntity(entityId);
deleteAndNotifyInternal(tenantId, entityId, scope, keys, false, callback);
}
@Override
public void deleteAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List<String> keys, FutureCallback<Void> callback) {
checkInternalEntity(entityId);
deleteAndNotifyInternal(tenantId, entityId, scope, keys, false, callback);
}
@Override
public void deleteAndNotify(TenantId tenantId, EntityId entityId, String scope, List<String> keys, boolean notifyDevice, FutureCallback<Void> callback) {
checkInternalEntity(entityId);
deleteAndNotifyInternal(tenantId, entityId, scope, keys, notifyDevice, callback);
}
@Override
public void deleteAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List<String> keys, boolean notifyDevice, FutureCallback<Void> 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<Void> 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<Void> 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<Void> 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<Void> 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<Void> 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<Void> 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<Void> 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<Void> callback) {
saveAndNotify(tenantId, entityId, scope, Collections.singletonList(new BaseAttributeKvEntry(new BooleanDataEntry(key, value)
, System.currentTimeMillis())), callback);
}
@Override
public ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, long value) {
SettableFuture<Void> future = SettableFuture.create();
saveAttrAndNotify(tenantId, entityId, scope, key, value, new VoidFutureCallback(future));
return future;
}
@Override
public ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, long value) {
SettableFuture<Void> future = SettableFuture.create();
@ -421,13 +370,6 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
return future;
}
@Override
public ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, String value) {
SettableFuture<Void> future = SettableFuture.create();
saveAttrAndNotify(tenantId, entityId, scope, key, value, new VoidFutureCallback(future));
return future;
}
@Override
public ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, String value) {
SettableFuture<Void> future = SettableFuture.create();
@ -435,13 +377,6 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
return future;
}
@Override
public ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, double value) {
SettableFuture<Void> future = SettableFuture.create();
saveAttrAndNotify(tenantId, entityId, scope, key, value, new VoidFutureCallback(future));
return future;
}
@Override
public ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, double value) {
SettableFuture<Void> future = SettableFuture.create();
@ -449,13 +384,6 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
return future;
}
@Override
public ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, boolean value) {
SettableFuture<Void> future = SettableFuture.create();
saveAttrAndNotify(tenantId, entityId, scope, key, value, new VoidFutureCallback(future));
return future;
}
@Override
public ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, boolean value) {
SettableFuture<Void> future = SettableFuture.create();
@ -560,6 +488,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
public void onFailure(Throwable t) {
future.setException(t);
}
}
}

36
application/src/test/java/org/thingsboard/server/controller/WebsocketApiTest.java
View File

@ -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<TsKvEntry> tsData) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
tsService.saveAndNotify(device.getTenantId(), null, device.getId(), tsData, 0, new FutureCallback<Void>() {
@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<Void>() {
@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();
}
}

40
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java
View File

@ -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<Void> saveAndNotify(TenantId tenantId, EntityId entityId, TsKvEntry ts);
void save(TimeseriesSaveRequest request);
void saveAndNotify(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts, FutureCallback<Void> callback);
void saveAndNotify(TenantId tenantId, CustomerId id, EntityId entityId, List<TsKvEntry> ts, long ttl, FutureCallback<Void> callback);
void saveWithoutLatestAndNotify(TenantId tenantId, CustomerId id, EntityId entityId, List<TsKvEntry> ts, long ttl, FutureCallback<Void> callback);
ListenableFuture<Void> saveAndNotify(TimeseriesSaveRequest request);
@Deprecated(since = "3.7.0")
void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, FutureCallback<Void> callback);
@ -53,54 +48,24 @@ public interface RuleEngineTelemetryService {
void saveLatestAndNotify(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts, FutureCallback<Void> callback);
@Deprecated(since = "3.7.0")
ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, long value);
ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, long value);
@Deprecated(since = "3.7.0")
ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, String value);
ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, String value);
@Deprecated(since = "3.7.0")
ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, double value);
ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, double value);
@Deprecated(since = "3.7.0")
ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, boolean value);
ListenableFuture<Void> 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<Void> callback);
void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, long value, FutureCallback<Void> callback);
@Deprecated(since = "3.7.0")
void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, String value, FutureCallback<Void> callback);
void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, String value, FutureCallback<Void> callback);
@Deprecated(since = "3.7.0")
void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, double value, FutureCallback<Void> callback);
void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, double value, FutureCallback<Void> callback);
@Deprecated(since = "3.7.0")
void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, boolean value, FutureCallback<Void> callback);
void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, boolean value, FutureCallback<Void> callback);
@Deprecated(since = "3.7.0")
void deleteAndNotify(TenantId tenantId, EntityId entityId, String scope, List<String> keys, FutureCallback<Void> callback);
void deleteAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List<String> keys, FutureCallback<Void> callback);
@Deprecated(since = "3.7.0")
void deleteAndNotify(TenantId tenantId, EntityId entityId, String scope, List<String> keys, boolean notifyDevice, FutureCallback<Void> callback);
void deleteAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List<String> keys, boolean notifyDevice, FutureCallback<Void> callback);
void deleteLatest(TenantId tenantId, EntityId entityId, List<String> keys, FutureCallback<Void> callback);
@ -108,4 +73,5 @@ public interface RuleEngineTelemetryService {
void deleteAllLatest(TenantId tenantId, EntityId entityId, FutureCallback<Collection<String>> callback);
void deleteTimeseriesAndNotify(TenantId tenantId, EntityId entityId, List<String> keys, List<DeleteTsKvQuery> deleteTsKvQueries, FutureCallback<Void> callback);
}

102
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java
View File

@ -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<TsKvEntry> entries;
private final long ttl;
private final boolean saveLatest;
private FutureCallback<Void> callback;
public static Builder builder() {
return new Builder();
}
public static class Builder {
private TenantId tenantId;
private CustomerId customerId;
private EntityId entityId;
private List<TsKvEntry> entries;
private long ttl;
private FutureCallback<Void> 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<TsKvEntry> 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<Void> callback) {
this.callback = callback;
return this;
}
public TimeseriesSaveRequest build() {
return new TimeseriesSaveRequest(tenantId, customerId, entityId, entries, ttl, saveLatest, callback);
}
}
}

12
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java
View File

@ -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<Void> 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<Void> saveAttribute(TbContext ctx, TbMsg msg, double result, TbMathResult mathResultDef) {

15
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java
View File

@ -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) {

18
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/math/TbMathNodeTest.java
View File

@ -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<TbMsg> 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<TbMsg> 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);

53
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeTest.java
View File

@ -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<TsKvEntry> expectedList = getTsKvEntriesListWithTs(data, System.currentTimeMillis());
ArgumentCaptor<List<TsKvEntry>> 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<TsKvEntry> expectedList = getTsKvEntriesListWithTs(data, ts);
ArgumentCaptor<List<TsKvEntry>> 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<Arguments> givenTtlFromConfigAndTtlFromMd_whenOnMsg_thenVerifyTtl() {

Write Preview
Loading…
Cancel
Save