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

Refactor saveAndNotify and saveAndNotifyInternal for attributes

pull/12297/head
ViacheslavKlimov 2 years ago
parent
aeb7427833
commit
e8cf3179c5
23 changed files with 619 additions and 450 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 33
      application/src/main/java/org/thingsboard/server/controller/TelemetryController.java
  2. 16
      application/src/main/java/org/thingsboard/server/service/device/ClaimDevicesServiceImpl.java
  3. 22
      application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
  4. 43
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java
  5. 66
      application/src/main/java/org/thingsboard/server/service/entitiy/entityview/DefaultTbEntityViewService.java
  6. 27
      application/src/main/java/org/thingsboard/server/service/ota/DefaultOtaPackageStateService.java
  7. 42
      application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java
  8. 39
      application/src/main/java/org/thingsboard/server/service/sync/ie/importing/csv/AbstractBulkImportService.java
  9. 25
      application/src/main/java/org/thingsboard/server/service/sync/ie/importing/impl/BaseEntityImportService.java
  10. 186
      application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java
  11. 7
      application/src/main/java/org/thingsboard/server/service/telemetry/InternalTelemetryService.java
  12. 33
      application/src/test/java/org/thingsboard/server/controller/WebsocketApiTest.java
  13. 242
      application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java
  14. 115
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java
  15. 27
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java
  16. 9
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java
  17. 10
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java
  18. 15
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java
  19. 31
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java
  20. 39
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNodeTest.java
  21. 11
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/math/TbMathNodeTest.java
  22. 20
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeTest.java
  23. 11
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeTest.java

33
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.springframework.web.context.request.async.DeferredResult;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.common.util.ThingsBoardThreadFactory;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.adaptor.JsonConverter;
import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.AttributeScope;
@ -625,19 +626,25 @@ public class TelemetryController extends BaseController {
} }
SecurityUser user = getCurrentUser(); SecurityUser user = getCurrentUser();
return accessValidator.validateEntityAndCallback(getCurrentUser(), Operation.WRITE_ATTRIBUTES, entityIdSrc, (result, tenantId, entityId) -> { return accessValidator.validateEntityAndCallback(getCurrentUser(), Operation.WRITE_ATTRIBUTES, entityIdSrc, (result, tenantId, entityId) -> {
tsSubService.saveAndNotify(tenantId, entityId, scope, attributes, new FutureCallback<Void>() { tsSubService.save(AttributesSaveRequest.builder()
@Override .tenantId(tenantId)
public void onSuccess(@Nullable Void tmp) { .entityId(entityId)
logAttributesUpdated(user, entityId, scope, attributes, null); .scope(scope)
result.setResult(new ResponseEntity(HttpStatus.OK)); .entries(attributes)
} .callback(new FutureCallback<>() {
@Override
@Override public void onSuccess(@Nullable Void tmp) {
public void onFailure(Throwable t) { logAttributesUpdated(user, entityId, scope, attributes, null);
logAttributesUpdated(user, entityId, scope, attributes, t); result.setResult(new ResponseEntity(HttpStatus.OK));
AccessValidator.handleError(t, result, HttpStatus.INTERNAL_SERVER_ERROR); }
}
}); @Override
public void onFailure(Throwable t) {
logAttributesUpdated(user, entityId, scope, attributes, t);
AccessValidator.handleError(t, result, HttpStatus.INTERNAL_SERVER_ERROR);
}
})
.build());
}); });
} else { } else {
return getImmediateDeferredResult("Request is not a JSON object", HttpStatus.BAD_REQUEST); return getImmediateDeferredResult("Request is not a JSON object", HttpStatus.BAD_REQUEST);

16
application/src/main/java/org/thingsboard/server/service/device/ClaimDevicesServiceImpl.java
View File

@ -28,6 +28,7 @@ import org.springframework.cache.Cache;
import org.springframework.cache.CacheManager; import org.springframework.cache.CacheManager;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService;
import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.Customer;
@ -37,7 +38,6 @@ import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
import org.thingsboard.server.common.data.kv.BooleanDataEntry; import org.thingsboard.server.common.data.kv.BooleanDataEntry;
import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.attributes.AttributesService;
import org.thingsboard.server.dao.customer.CustomerService; import org.thingsboard.server.dao.customer.CustomerService;
@ -178,11 +178,12 @@ public class ClaimDevicesServiceImpl implements ClaimDevicesService {
return Futures.immediateFuture(new ReclaimResult(unassignedCustomer)); return Futures.immediateFuture(new ReclaimResult(unassignedCustomer));
} }
SettableFuture<ReclaimResult> result = SettableFuture.create(); SettableFuture<ReclaimResult> result = SettableFuture.create();
telemetryService.saveAndNotify( telemetryService.save(AttributesSaveRequest.builder()
tenantId, savedDevice.getId(), AttributeScope.SERVER_SCOPE, List.of( .tenantId(tenantId)
new BaseAttributeKvEntry(new BooleanDataEntry(CLAIM_ATTRIBUTE_NAME, true), System.currentTimeMillis()) .entityId(savedDevice.getId())
), .scope(AttributeScope.SERVER_SCOPE)
new FutureCallback<>() { .entry(new BooleanDataEntry(CLAIM_ATTRIBUTE_NAME, true))
.callback(new FutureCallback<>() {
@Override @Override
public void onSuccess(@Nullable Void tmp) { public void onSuccess(@Nullable Void tmp) {
result.set(new ReclaimResult(unassignedCustomer)); result.set(new ReclaimResult(unassignedCustomer));
@ -192,7 +193,8 @@ public class ClaimDevicesServiceImpl implements ClaimDevicesService {
public void onFailure(Throwable t) { public void onFailure(Throwable t) {
result.setException(t); result.setException(t);
} }
}); })
.build());
return result; return result;
} }
cacheEviction(device.getId()); cacheEviction(device.getId());

22
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.springframework.stereotype.Service;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.common.util.ThingsBoardExecutors;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
import org.thingsboard.server.cache.TbTransactionalCache; import org.thingsboard.server.cache.TbTransactionalCache;
import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.cluster.TbClusterService;
@ -42,7 +43,6 @@ import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEvent;
import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.EdgeId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.BooleanDataEntry; import org.thingsboard.server.common.data.kv.BooleanDataEntry;
import org.thingsboard.server.common.data.kv.LongDataEntry; import org.thingsboard.server.common.data.kv.LongDataEntry;
import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.data.msg.TbMsgType;
@ -506,11 +506,17 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
tsSubService.save(TimeseriesSaveRequest.builder() tsSubService.save(TimeseriesSaveRequest.builder()
.tenantId(tenantId) .tenantId(tenantId)
.entityId(edgeId) .entityId(edgeId)
.entry(new BasicTsKvEntry(System.currentTimeMillis(), new LongDataEntry(key, value))) .entry(new LongDataEntry(key, value))
.callback(new AttributeSaveCallback(tenantId, edgeId, key, value)) .callback(new AttributeSaveCallback(tenantId, edgeId, key, value))
.build()); .build());
} else { } else {
tsSubService.saveAttrAndNotify(tenantId, edgeId, AttributeScope.SERVER_SCOPE, key, value, new AttributeSaveCallback(tenantId, edgeId, key, value)); tsSubService.save(AttributesSaveRequest.builder()
.tenantId(tenantId)
.entityId(edgeId)
.scope(AttributeScope.SERVER_SCOPE)
.entry(new LongDataEntry(key, value))
.callback(new AttributeSaveCallback(tenantId, edgeId, key, value))
.build());
} }
} }
@ -520,11 +526,17 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
tsSubService.save(TimeseriesSaveRequest.builder() tsSubService.save(TimeseriesSaveRequest.builder()
.tenantId(tenantId) .tenantId(tenantId)
.entityId(edgeId) .entityId(edgeId)
.entry(new BasicTsKvEntry(System.currentTimeMillis(), new BooleanDataEntry(key, value))) .entry(new BooleanDataEntry(key, value))
.callback(new AttributeSaveCallback(tenantId, edgeId, key, value)) .callback(new AttributeSaveCallback(tenantId, edgeId, key, value))
.build()); .build());
} else { } else {
tsSubService.saveAttrAndNotify(tenantId, edgeId, AttributeScope.SERVER_SCOPE, key, value, new AttributeSaveCallback(tenantId, edgeId, key, value)); tsSubService.save(AttributesSaveRequest.builder()
.tenantId(tenantId)
.entityId(edgeId)
.scope(AttributeScope.SERVER_SCOPE)
.entry(new BooleanDataEntry(key, value))
.callback(new AttributeSaveCallback(tenantId, edgeId, key, value))
.build());
} }
} }

43
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java
View File

@ -31,6 +31,7 @@ import org.apache.commons.lang3.tuple.Pair;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy; import org.springframework.context.annotation.Lazy;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.adaptor.JsonConverter;
import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.DataConstants;
@ -277,16 +278,29 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor {
JsonObject json = JsonUtils.getJsonObject(msg.getKvList()); JsonObject json = JsonUtils.getJsonObject(msg.getKvList());
List<AttributeKvEntry> attributes = new ArrayList<>(JsonConverter.convertToAttributes(json)); List<AttributeKvEntry> attributes = new ArrayList<>(JsonConverter.convertToAttributes(json));
String scope = metaData.getValue("scope"); String scope = metaData.getValue("scope");
tsSubService.saveAndNotify(tenantId, entityId, AttributeScope.valueOf(scope), attributes, new FutureCallback<Void>() { tsSubService.save(AttributesSaveRequest.builder()
@Override .tenantId(tenantId)
public void onSuccess(@Nullable Void tmp) { .entityId(entityId)
var defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId); .scope(AttributeScope.valueOf(scope))
TbMsg tbMsg = TbMsg.newMsg(defaultQueueAndRuleChain.getKey(), TbMsgType.ATTRIBUTES_UPDATED, entityId, .entries(attributes)
customerId, metaData, gson.toJson(json), defaultQueueAndRuleChain.getValue(), null); .callback(new FutureCallback<>() {
edgeCtx.getClusterService().pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() {
@Override @Override
public void onSuccess(TbQueueMsgMetadata metadata) { public void onSuccess(@Nullable Void tmp) {
futureToSet.set(null); var defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId);
TbMsg tbMsg = TbMsg.newMsg(defaultQueueAndRuleChain.getKey(), TbMsgType.ATTRIBUTES_UPDATED, entityId,
customerId, metaData, gson.toJson(json), defaultQueueAndRuleChain.getValue(), null);
edgeCtx.getClusterService().pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() {
@Override
public void onSuccess(TbQueueMsgMetadata metadata) {
futureToSet.set(null);
}
@Override
public void onFailure(Throwable t) {
log.error("[{}] Can't process attributes update [{}]", tenantId, msg, t);
futureToSet.setException(t);
}
});
} }
@Override @Override
@ -294,15 +308,8 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor {
log.error("[{}] Can't process attributes update [{}]", tenantId, msg, t); log.error("[{}] Can't process attributes update [{}]", tenantId, msg, t);
futureToSet.setException(t); futureToSet.setException(t);
} }
}); })
} .build());
@Override
public void onFailure(Throwable t) {
log.error("[{}] Can't process attributes update [{}]", tenantId, msg, t);
futureToSet.setException(t);
}
});
return futureToSet; return futureToSet;
} }

66
application/src/main/java/org/thingsboard/server/service/entitiy/entityview/DefaultTbEntityViewService.java
View File

@ -25,6 +25,7 @@ import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.util.ConcurrentReferenceHashMap; import org.springframework.util.ConcurrentReferenceHashMap;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.Customer;
import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityType;
@ -273,36 +274,41 @@ public class DefaultTbEntityViewService extends AbstractTbEntityService implemen
return Futures.transform(getAttrFuture, attributeKvEntries -> { return Futures.transform(getAttrFuture, attributeKvEntries -> {
List<AttributeKvEntry> attributes; List<AttributeKvEntry> attributes;
if (attributeKvEntries != null && !attributeKvEntries.isEmpty()) { if (attributeKvEntries != null && !attributeKvEntries.isEmpty()) {
attributes = attributes = attributeKvEntries.stream()
attributeKvEntries.stream() .filter(attributeKvEntry -> {
.filter(attributeKvEntry -> { long startTime = entityView.getStartTimeMs();
long startTime = entityView.getStartTimeMs(); long endTime = entityView.getEndTimeMs();
long endTime = entityView.getEndTimeMs(); long lastUpdateTs = attributeKvEntry.getLastUpdateTs();
long lastUpdateTs = attributeKvEntry.getLastUpdateTs(); return startTime == 0 && endTime == 0 ||
return startTime == 0 && endTime == 0 || (endTime == 0 && startTime < lastUpdateTs) ||
(endTime == 0 && startTime < lastUpdateTs) || (startTime == 0 && endTime > lastUpdateTs) ||
(startTime == 0 && endTime > lastUpdateTs) || (startTime < lastUpdateTs && endTime > lastUpdateTs);
(startTime < lastUpdateTs && endTime > lastUpdateTs); }).collect(Collectors.toList());
}).collect(Collectors.toList()); tsSubService.save(AttributesSaveRequest.builder()
tsSubService.saveAndNotify(entityView.getTenantId(), entityId, scope, attributes, new FutureCallback<Void>() { .tenantId(entityView.getTenantId())
@Override .entityId(entityId)
public void onSuccess(@Nullable Void tmp) { .scope(scope)
try { .entries(attributes)
logAttributesUpdated(entityView.getTenantId(), user, entityId, scope, attributes, null); .callback(new FutureCallback<>() {
} catch (ThingsboardException e) { @Override
log.error("Failed to log attribute updates", e); public void onSuccess(@Nullable Void tmp) {
} try {
} logAttributesUpdated(entityView.getTenantId(), user, entityId, scope, attributes, null);
} catch (ThingsboardException e) {
@Override log.error("Failed to log attribute updates", e);
public void onFailure(Throwable t) { }
try { }
logAttributesUpdated(entityView.getTenantId(), user, entityId, scope, attributes, t);
} catch (ThingsboardException e) { @Override
log.error("Failed to log attribute updates", e); public void onFailure(Throwable t) {
} try {
} logAttributesUpdated(entityView.getTenantId(), user, entityId, scope, attributes, t);
}); } catch (ThingsboardException e) {
log.error("Failed to log attribute updates", e);
}
}
})
.build());
} }
return null; return null;
}, MoreExecutors.directExecutor()); }, MoreExecutors.directExecutor());

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

@ -20,6 +20,7 @@ import jakarta.annotation.Nullable;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Lazy; import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService;
import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.cluster.TbClusterService;
@ -346,17 +347,23 @@ public class DefaultOtaPackageStateService implements OtaPackageStateService {
remove(device, otaPackageType, attrToRemove); remove(device, otaPackageType, attrToRemove);
telemetryService.saveAndNotify(tenantId, deviceId, AttributeScope.SHARED_SCOPE, attributes, new FutureCallback<>() { telemetryService.save(AttributesSaveRequest.builder()
@Override .tenantId(tenantId)
public void onSuccess(@Nullable Void tmp) { .entityId(deviceId)
log.trace("[{}] Success save attributes with target firmware!", deviceId); .scope(AttributeScope.SHARED_SCOPE)
} .entries(attributes)
.callback(new FutureCallback<>() {
@Override
public void onSuccess(@Nullable Void tmp) {
log.trace("[{}] Success save attributes with target firmware!", deviceId);
}
@Override @Override
public void onFailure(Throwable t) { public void onFailure(Throwable t) {
log.error("[{}] Failed to save attributes with target firmware!", deviceId, t); log.error("[{}] Failed to save attributes with target firmware!", deviceId, t);
} }
}); })
.build());
} }
private void remove(Device device, OtaPackageType otaPackageType) { private void remove(Device device, OtaPackageType otaPackageType) {

42
application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java
View File

@ -38,6 +38,7 @@ import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.common.util.ThingsBoardExecutors;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.ApiUsageRecordKey; import org.thingsboard.server.common.data.ApiUsageRecordKey;
@ -52,6 +53,7 @@ import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.id.UUIDBased; import org.thingsboard.server.common.data.id.UUIDBased;
import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry; import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.BooleanDataEntry; import org.thingsboard.server.common.data.kv.BooleanDataEntry;
import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.KvEntry;
@ -866,30 +868,30 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
} }
private void save(DeviceId deviceId, String key, long value) { private void save(DeviceId deviceId, String key, long value) {
if (persistToTelemetry) { save(deviceId, new LongDataEntry(key, value), getCurrentTimeMillis());
tsSubService.saveInternal(TimeseriesSaveRequest.builder()
.tenantId(TenantId.SYS_TENANT_ID)
.entityId(deviceId)
.entry(new BasicTsKvEntry(getCurrentTimeMillis(), new LongDataEntry(key, value)))
.ttl(telemetryTtl)
.callback(new TelemetrySaveCallback<>(deviceId, key, value))
.build());
} else {
tsSubService.saveAttrAndNotify(TenantId.SYS_TENANT_ID, deviceId, AttributeScope.SERVER_SCOPE, key, value, new TelemetrySaveCallback<>(deviceId, key, value));
}
} }
private void save(DeviceId deviceId, String key, boolean value) { private void save(DeviceId deviceId, String key, boolean value) {
save(deviceId, new BooleanDataEntry(key, value), getCurrentTimeMillis());
}
private void save(DeviceId deviceId, KvEntry kvEntry, long ts) {
if (persistToTelemetry) { if (persistToTelemetry) {
tsSubService.saveInternal(TimeseriesSaveRequest.builder() tsSubService.saveInternal(TimeseriesSaveRequest.builder()
.tenantId(TenantId.SYS_TENANT_ID) .tenantId(TenantId.SYS_TENANT_ID)
.entityId(deviceId) .entityId(deviceId)
.entry(new BasicTsKvEntry(getCurrentTimeMillis(), new BooleanDataEntry(key, value))) .entry(new BasicTsKvEntry(ts, kvEntry))
.ttl(telemetryTtl) .ttl(telemetryTtl)
.callback(new TelemetrySaveCallback<>(deviceId, key, value)) .callback(new TelemetrySaveCallback<>(deviceId, kvEntry))
.build()); .build());
} else { } else {
tsSubService.saveAttrAndNotify(TenantId.SYS_TENANT_ID, deviceId, AttributeScope.SERVER_SCOPE, key, value, new TelemetrySaveCallback<>(deviceId, key, value)); tsSubService.save(AttributesSaveRequest.builder()
.tenantId(TenantId.SYS_TENANT_ID)
.entityId(deviceId)
.scope(AttributeScope.SERVER_SCOPE)
.entry(new BaseAttributeKvEntry(ts, kvEntry))
.callback(new TelemetrySaveCallback<>(deviceId, kvEntry))
.build());
} }
} }
@ -899,23 +901,21 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
private static class TelemetrySaveCallback<T> implements FutureCallback<T> { private static class TelemetrySaveCallback<T> implements FutureCallback<T> {
private final DeviceId deviceId; private final DeviceId deviceId;
private final String key; private final KvEntry kvEntry;
private final Object value;
TelemetrySaveCallback(DeviceId deviceId, String key, Object value) { TelemetrySaveCallback(DeviceId deviceId, KvEntry kvEntry) {
this.deviceId = deviceId; this.deviceId = deviceId;
this.key = key; this.kvEntry = kvEntry;
this.value = value;
} }
@Override @Override
public void onSuccess(@Nullable T result) { public void onSuccess(@Nullable T result) {
log.trace("[{}] Successfully updated attribute [{}] with value [{}]", deviceId, key, value); log.trace("[{}] Successfully updated entry {}", deviceId, kvEntry);
} }
@Override @Override
public void onFailure(Throwable t) { public void onFailure(Throwable t) {
log.warn("[{}] Failed to update attribute [{}] with value [{}]", deviceId, key, value, t); log.warn("[{}] Failed to update entry {}", deviceId, kvEntry, t);
} }
} }
} }

39
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.DonAsynchron;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.common.util.ThingsBoardExecutors;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.adaptor.JsonConverter;
import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.AttributeScope;
@ -237,23 +238,27 @@ public abstract class AbstractBulkImportService<E extends HasId<? extends Entity
List<AttributeKvEntry> attributes = new ArrayList<>(JsonConverter.convertToAttributes(kvsEntry.getValue())); List<AttributeKvEntry> attributes = new ArrayList<>(JsonConverter.convertToAttributes(kvsEntry.getValue()));
accessValidator.validateEntityAndCallback(user, Operation.WRITE_ATTRIBUTES, entity.getId(), (result, tenantId, entityId) -> { accessValidator.validateEntityAndCallback(user, Operation.WRITE_ATTRIBUTES, entity.getId(), (result, tenantId, entityId) -> {
tsSubscriptionService.saveAndNotify(tenantId, entityId, AttributeScope.valueOf(scope), attributes, new FutureCallback<>() { tsSubscriptionService.save(AttributesSaveRequest.builder()
.tenantId(tenantId)
@Override .entityId(entityId)
public void onSuccess(Void unused) { .scope(AttributeScope.valueOf(scope))
entityActionService.logEntityAction(user, (UUIDBased & EntityId) entityId, null, .entries(attributes)
null, ActionType.ATTRIBUTES_UPDATED, null, AttributeScope.valueOf(scope), attributes); .callback(new FutureCallback<>() {
} @Override
public void onSuccess(Void unused) {
@Override entityActionService.logEntityAction(user, (UUIDBased & EntityId) entityId, null,
public void onFailure(Throwable throwable) { null, ActionType.ATTRIBUTES_UPDATED, null, AttributeScope.valueOf(scope), attributes);
entityActionService.logEntityAction(user, (UUIDBased & EntityId) entityId, null, }
null, ActionType.ATTRIBUTES_UPDATED, BaseController.toException(throwable),
AttributeScope.valueOf(scope), attributes); @Override
throw new RuntimeException(throwable); public void onFailure(Throwable throwable) {
} entityActionService.logEntityAction(user, (UUIDBased & EntityId) entityId, null,
null, ActionType.ATTRIBUTES_UPDATED, BaseController.toException(throwable),
}); AttributeScope.valueOf(scope), attributes);
throw new RuntimeException(throwable);
}
})
.build());
}); });
} }

25
application/src/main/java/org/thingsboard/server/service/sync/ie/importing/impl/BaseEntityImportService.java
View File

@ -24,6 +24,7 @@ import org.checkerframework.checker.nullness.qual.Nullable;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy; import org.springframework.context.annotation.Lazy;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.ExportableEntity; import org.thingsboard.server.common.data.ExportableEntity;
@ -257,16 +258,22 @@ public abstract class BaseEntityImportService<I extends EntityId, E extends Expo
}) })
.collect(Collectors.toList()); .collect(Collectors.toList());
// fixme: attributes are saved outside the transaction // fixme: attributes are saved outside the transaction
tsSubService.saveAndNotify(user.getTenantId(), entity.getId(), scope, attributeKvEntries, new FutureCallback<Void>() { tsSubService.save(AttributesSaveRequest.builder()
@Override .tenantId(user.getTenantId())
public void onSuccess(@Nullable Void unused) { .entityId(entity.getId())
} .scope(scope)
.entries(attributeKvEntries)
.callback(new FutureCallback<>() {
@Override
public void onSuccess(@Nullable Void unused) {
}
@Override @Override
public void onFailure(Throwable thr) { public void onFailure(Throwable thr) {
log.error("Failed to import attributes for {} {}", entity.getId().getEntityType(), entity.getId(), thr); log.error("Failed to import attributes for {} {}", entity.getId().getEntityType(), entity.getId(), thr);
} }
}); })
.build());
}); });
}); });
} }

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

@ -28,6 +28,7 @@ import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy; import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.common.util.ThingsBoardThreadFactory;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService;
import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
import org.thingsboard.server.common.data.ApiUsageRecordKey; import org.thingsboard.server.common.data.ApiUsageRecordKey;
@ -38,12 +39,7 @@ import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
import org.thingsboard.server.common.data.kv.BooleanDataEntry;
import org.thingsboard.server.common.data.kv.DeleteTsKvQuery; import org.thingsboard.server.common.data.kv.DeleteTsKvQuery;
import org.thingsboard.server.common.data.kv.DoubleDataEntry;
import org.thingsboard.server.common.data.kv.LongDataEntry;
import org.thingsboard.server.common.data.kv.StringDataEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.data.kv.TsKvLatestRemovingResult; import org.thingsboard.server.common.data.kv.TsKvLatestRemovingResult;
import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.common.msg.queue.TbCallback;
@ -57,7 +53,6 @@ import org.thingsboard.server.service.subscription.TbSubscriptionUtils;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collection; import java.util.Collection;
import java.util.Collections;
import java.util.Comparator; import java.util.Comparator;
import java.util.HashMap; import java.util.HashMap;
import java.util.List; import java.util.List;
@ -159,89 +154,18 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
return saveFuture; return saveFuture;
} }
private void addEntityViewCallback(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts) {
if (EntityType.DEVICE.equals(entityId.getEntityType()) || EntityType.ASSET.equals(entityId.getEntityType())) {
Futures.addCallback(this.tbEntityViewService.findEntityViewsByTenantIdAndEntityIdAsync(tenantId, entityId),
new FutureCallback<>() {
@Override
public void onSuccess(@Nullable List<EntityView> result) {
if (result != null && !result.isEmpty()) {
Map<String, List<TsKvEntry>> tsMap = new HashMap<>();
for (TsKvEntry entry : ts) {
tsMap.computeIfAbsent(entry.getKey(), s -> new ArrayList<>()).add(entry);
}
for (EntityView entityView : result) {
List<String> keys = entityView.getKeys() != null && entityView.getKeys().getTimeseries() != null ?
entityView.getKeys().getTimeseries() : new ArrayList<>(tsMap.keySet());
List<TsKvEntry> entityViewLatest = new ArrayList<>();
long startTs = entityView.getStartTimeMs();
long endTs = entityView.getEndTimeMs() == 0 ? Long.MAX_VALUE : entityView.getEndTimeMs();
for (String key : keys) {
List<TsKvEntry> entries = tsMap.get(key);
if (entries != null) {
Optional<TsKvEntry> tsKvEntry = entries.stream()
.filter(entry -> entry.getTs() > startTs && entry.getTs() <= endTs)
.max(Comparator.comparingLong(TsKvEntry::getTs));
tsKvEntry.ifPresent(entityViewLatest::add);
}
}
if (!entityViewLatest.isEmpty()) {
saveLatestAndNotify(tenantId, entityView.getId(), entityViewLatest, new FutureCallback<>() {
@Override
public void onSuccess(@Nullable Void tmp) {
}
@Override
public void onFailure(Throwable t) {
}
});
}
}
}
}
@Override
public void onFailure(Throwable t) {
log.error("Error while finding entity views by tenantId and entityId", t);
}
}, MoreExecutors.directExecutor());
}
}
@Override
public void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, FutureCallback<Void> callback) {
saveAndNotify(tenantId, entityId, scope, attributes, true, callback);
}
@Override @Override
public void saveAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List<AttributeKvEntry> attributes, FutureCallback<Void> callback) { public void save(AttributesSaveRequest request) {
saveAndNotify(tenantId, entityId, scope, attributes, true, callback); checkInternalEntity(request.getEntityId());
saveInternal(request);
} }
@Override @Override
public void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice, FutureCallback<Void> callback) { public void saveInternal(AttributesSaveRequest request) {
checkInternalEntity(entityId); log.trace("Executing saveInternal [{}]", request);
saveAndNotifyInternal(tenantId, entityId, scope, attributes, notifyDevice, callback); ListenableFuture<List<Long>> saveFuture = attrService.save(request.getTenantId(), request.getEntityId(), request.getScope(), request.getEntries());
} addVoidCallback(saveFuture, request.getCallback());
addWsCallback(saveFuture, success -> onAttributesUpdate(request.getTenantId(), request.getEntityId(), request.getScope().name(), request.getEntries(), request.isNotifyDevice()));
@Override
public void saveAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List<AttributeKvEntry> attributes, boolean notifyDevice, FutureCallback<Void> callback) {
checkInternalEntity(entityId);
saveAndNotifyInternal(tenantId, entityId, scope, attributes, notifyDevice, callback);
}
@Override
public void saveAndNotifyInternal(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice, FutureCallback<Void> callback) {
ListenableFuture<List<Long>> saveFuture = attrService.save(tenantId, entityId, scope, attributes);
addVoidCallback(saveFuture, callback);
addWsCallback(saveFuture, success -> onAttributesUpdate(tenantId, entityId, scope, attributes, notifyDevice));
}
@Override
public void saveAndNotifyInternal(TenantId tenantId, EntityId entityId, AttributeScope scope, List<AttributeKvEntry> attributes, boolean notifyDevice, FutureCallback<Void> callback) {
ListenableFuture<List<Long>> saveFuture = attrService.save(tenantId, entityId, scope, attributes);
addVoidCallback(saveFuture, callback);
addWsCallback(saveFuture, success -> onAttributesUpdate(tenantId, entityId, scope.name(), attributes, notifyDevice));
} }
@Override @Override
@ -318,57 +242,61 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
addWsCallback(deleteFuture, list -> onTimeSeriesDelete(tenantId, entityId, keys, list)); addWsCallback(deleteFuture, list -> onTimeSeriesDelete(tenantId, entityId, keys, list));
} }
@Override
public void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, long value, FutureCallback<Void> callback) {
saveAndNotify(tenantId, entityId, scope, Collections.singletonList(new BaseAttributeKvEntry(new LongDataEntry(key, value)
, System.currentTimeMillis())), callback);
}
@Override @Override
public void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, String value, FutureCallback<Void> callback) { public ListenableFuture<Void> saveAttrAndNotify(AttributesSaveRequest request) {
saveAndNotify(tenantId, entityId, scope, Collections.singletonList(new BaseAttributeKvEntry(new StringDataEntry(key, value)
, System.currentTimeMillis())), callback);
}
@Override
public void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, double value, FutureCallback<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, 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, AttributeScope scope, String key, long value) {
SettableFuture<Void> future = SettableFuture.create(); SettableFuture<Void> future = SettableFuture.create();
saveAttrAndNotify(tenantId, entityId, scope, key, value, new VoidFutureCallback(future)); request.setCallback(new VoidFutureCallback(future));
save(request);
return future; return future;
} }
@Override private void addEntityViewCallback(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts) {
public ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, String value) { if (EntityType.DEVICE.equals(entityId.getEntityType()) || EntityType.ASSET.equals(entityId.getEntityType())) {
SettableFuture<Void> future = SettableFuture.create(); Futures.addCallback(this.tbEntityViewService.findEntityViewsByTenantIdAndEntityIdAsync(tenantId, entityId),
saveAttrAndNotify(tenantId, entityId, scope, key, value, new VoidFutureCallback(future)); new FutureCallback<>() {
return future; @Override
} public void onSuccess(@Nullable List<EntityView> result) {
if (result != null && !result.isEmpty()) {
Map<String, List<TsKvEntry>> tsMap = new HashMap<>();
for (TsKvEntry entry : ts) {
tsMap.computeIfAbsent(entry.getKey(), s -> new ArrayList<>()).add(entry);
}
for (EntityView entityView : result) {
List<String> keys = entityView.getKeys() != null && entityView.getKeys().getTimeseries() != null ?
entityView.getKeys().getTimeseries() : new ArrayList<>(tsMap.keySet());
List<TsKvEntry> entityViewLatest = new ArrayList<>();
long startTs = entityView.getStartTimeMs();
long endTs = entityView.getEndTimeMs() == 0 ? Long.MAX_VALUE : entityView.getEndTimeMs();
for (String key : keys) {
List<TsKvEntry> entries = tsMap.get(key);
if (entries != null) {
Optional<TsKvEntry> tsKvEntry = entries.stream()
.filter(entry -> entry.getTs() > startTs && entry.getTs() <= endTs)
.max(Comparator.comparingLong(TsKvEntry::getTs));
tsKvEntry.ifPresent(entityViewLatest::add);
}
}
if (!entityViewLatest.isEmpty()) {
saveLatestAndNotify(tenantId, entityView.getId(), entityViewLatest, new FutureCallback<>() {
@Override
public void onSuccess(@Nullable Void tmp) {
}
@Override @Override
public ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, double value) { public void onFailure(Throwable t) {
SettableFuture<Void> future = SettableFuture.create(); }
saveAttrAndNotify(tenantId, entityId, scope, key, value, new VoidFutureCallback(future)); });
return future; }
} }
}
}
@Override @Override
public ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, boolean value) { public void onFailure(Throwable t) {
SettableFuture<Void> future = SettableFuture.create(); log.error("Error while finding entity views by tenantId and entityId", t);
saveAttrAndNotify(tenantId, entityId, scope, key, value, new VoidFutureCallback(future)); }
return future; }, MoreExecutors.directExecutor());
}
} }
private void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice) { private void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice) {

7
application/src/main/java/org/thingsboard/server/service/telemetry/InternalTelemetryService.java
View File

@ -17,12 +17,12 @@ package org.thingsboard.server.service.telemetry;
import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListenableFuture;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService;
import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry;
import java.util.List; import java.util.List;
@ -34,10 +34,7 @@ public interface InternalTelemetryService extends RuleEngineTelemetryService {
ListenableFuture<Integer> saveInternal(TimeseriesSaveRequest request); ListenableFuture<Integer> saveInternal(TimeseriesSaveRequest request);
@Deprecated(since = "3.7.0") void saveInternal(AttributesSaveRequest request);
void saveAndNotifyInternal(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice, FutureCallback<Void> callback);
void saveAndNotifyInternal(TenantId tenantId, EntityId entityId, AttributeScope scope, List<AttributeKvEntry> attributes, boolean notifyDevice, FutureCallback<Void> callback);
void saveLatestAndNotifyInternal(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts, FutureCallback<Void> callback); void saveLatestAndNotifyInternal(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts, FutureCallback<Void> callback);

33
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.springframework.test.context.TestPropertySource;
import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils; import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.alarm.Alarm;
@ -832,19 +833,25 @@ public class WebsocketApiTest extends AbstractControllerTest {
private void sendAttributes(TenantId tenantId, EntityId entityId, TbAttributeSubscriptionScope scope, List<AttributeKvEntry> attrData) throws InterruptedException { private void sendAttributes(TenantId tenantId, EntityId entityId, TbAttributeSubscriptionScope scope, List<AttributeKvEntry> attrData) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1); CountDownLatch latch = new CountDownLatch(1);
tsService.saveAndNotify(tenantId, entityId, scope.getAttributeScope(), attrData, new FutureCallback<Void>() { tsService.save(AttributesSaveRequest.builder()
@Override .tenantId(tenantId)
public void onSuccess(@Nullable Void result) { .entityId(entityId)
log.debug("sendAttributes callback onSuccess"); .scope(scope.getAttributeScope())
latch.countDown(); .entries(attrData)
} .callback(new FutureCallback<>() {
@Override
@Override public void onSuccess(@Nullable Void result) {
public void onFailure(Throwable t) { log.debug("sendAttributes callback onSuccess");
log.error("Failed to sendAttributes", t); latch.countDown();
latch.countDown(); }
}
}); @Override
public void onFailure(Throwable t) {
log.error("Failed to sendAttributes", t);
latch.countDown();
}
})
.build());
assertThat(latch.await(TIMEOUT, TimeUnit.SECONDS)).as("await sendAttributes callback").isTrue(); assertThat(latch.await(TIMEOUT, TimeUnit.SECONDS)).as("await sendAttributes callback").isTrue();
} }

242
application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java
View File

@ -24,9 +24,11 @@ import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource; import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor; import org.mockito.ArgumentCaptor;
import org.mockito.ArgumentMatcher;
import org.mockito.Mock; import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension; import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.test.util.ReflectionTestUtils; import org.springframework.test.util.ReflectionTestUtils;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.Device;
@ -72,7 +74,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await; import static org.awaitility.Awaitility.await;
import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.then; import static org.mockito.BDDMockito.then;
@ -82,6 +84,7 @@ import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.never; import static org.mockito.Mockito.never;
import static org.mockito.Mockito.reset; import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.spy; import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when; import static org.mockito.Mockito.when;
import static org.thingsboard.server.service.state.DefaultDeviceStateService.ACTIVITY_STATE; import static org.thingsboard.server.service.state.DefaultDeviceStateService.ACTIVITY_STATE;
@ -208,9 +211,12 @@ public class DefaultDeviceStateServiceTest {
service.onDeviceConnect(tenantId, deviceId, lastConnectTime); service.onDeviceConnect(tenantId, deviceId, lastConnectTime);
// THEN // THEN
then(telemetrySubscriptionService).should().saveAttrAndNotify( then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), eq(LAST_CONNECT_TIME), eq(lastConnectTime), any() request.getTenantId().equals(TenantId.SYS_TENANT_ID) && request.getEntityId().equals(deviceId) &&
); request.getScope().equals(AttributeScope.SERVER_SCOPE) &&
request.getEntries().get(0).getKey().equals(LAST_CONNECT_TIME) &&
request.getEntries().get(0).getValue().equals(lastConnectTime)
));
var msgCaptor = ArgumentCaptor.forClass(TbMsg.class); var msgCaptor = ArgumentCaptor.forClass(TbMsg.class);
then(clusterService).should().pushMsgToRuleEngine(eq(tenantId), eq(deviceId), msgCaptor.capture(), any()); then(clusterService).should().pushMsgToRuleEngine(eq(tenantId), eq(deviceId), msgCaptor.capture(), any());
@ -292,10 +298,12 @@ public class DefaultDeviceStateServiceTest {
service.onDeviceDisconnect(tenantId, deviceId, lastDisconnectTime); service.onDeviceDisconnect(tenantId, deviceId, lastDisconnectTime);
// THEN // THEN
then(telemetrySubscriptionService).should().saveAttrAndNotify( then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), request.getTenantId().equals(TenantId.SYS_TENANT_ID) && request.getEntityId().equals(deviceId) &&
eq(LAST_DISCONNECT_TIME), eq(lastDisconnectTime), any() request.getScope().equals(AttributeScope.SERVER_SCOPE) &&
); request.getEntries().get(0).getKey().equals(LAST_DISCONNECT_TIME) &&
request.getEntries().get(0).getValue().equals(lastDisconnectTime)
));
var msgCaptor = ArgumentCaptor.forClass(TbMsg.class); var msgCaptor = ArgumentCaptor.forClass(TbMsg.class);
then(clusterService).should().pushMsgToRuleEngine(eq(tenantId), eq(deviceId), msgCaptor.capture(), any()); then(clusterService).should().pushMsgToRuleEngine(eq(tenantId), eq(deviceId), msgCaptor.capture(), any());
@ -413,14 +421,18 @@ public class DefaultDeviceStateServiceTest {
service.onDeviceInactivity(tenantId, deviceId, lastInactivityTime); service.onDeviceInactivity(tenantId, deviceId, lastInactivityTime);
// THEN // THEN
then(telemetrySubscriptionService).should().saveAttrAndNotify( then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), request.getTenantId().equals(TenantId.SYS_TENANT_ID) && request.getEntityId().equals(deviceId) &&
eq(INACTIVITY_ALARM_TIME), eq(lastInactivityTime), any() request.getScope().equals(AttributeScope.SERVER_SCOPE) &&
); request.getEntries().get(0).getKey().equals(INACTIVITY_ALARM_TIME) &&
then(telemetrySubscriptionService).should().saveAttrAndNotify( request.getEntries().get(0).getValue().equals(lastInactivityTime)
eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), ));
eq(ACTIVITY_STATE), eq(false), any() then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
); request.getTenantId().equals(TenantId.SYS_TENANT_ID) && request.getEntityId().equals(deviceId) &&
request.getScope().equals(AttributeScope.SERVER_SCOPE) &&
request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) &&
request.getEntries().get(0).getValue().equals(false)
));
var msgCaptor = ArgumentCaptor.forClass(TbMsg.class); var msgCaptor = ArgumentCaptor.forClass(TbMsg.class);
then(clusterService).should() then(clusterService).should()
@ -453,14 +465,17 @@ public class DefaultDeviceStateServiceTest {
service.updateInactivityStateIfExpired(System.currentTimeMillis(), deviceId, deviceStateData); service.updateInactivityStateIfExpired(System.currentTimeMillis(), deviceId, deviceStateData);
// THEN // THEN
then(telemetrySubscriptionService).should().saveAttrAndNotify( then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), request.getTenantId().equals(TenantId.SYS_TENANT_ID) && request.getEntityId().equals(deviceId) &&
eq(INACTIVITY_ALARM_TIME), anyLong(), any() request.getScope().equals(AttributeScope.SERVER_SCOPE) &&
); request.getEntries().get(0).getKey().equals(INACTIVITY_ALARM_TIME)
then(telemetrySubscriptionService).should().saveAttrAndNotify( ));
eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
eq(ACTIVITY_STATE), eq(false), any() request.getTenantId().equals(TenantId.SYS_TENANT_ID) && request.getEntityId().equals(deviceId) &&
); request.getScope().equals(AttributeScope.SERVER_SCOPE) &&
request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) &&
request.getEntries().get(0).getValue().equals(false)
));
var msgCaptor = ArgumentCaptor.forClass(TbMsg.class); var msgCaptor = ArgumentCaptor.forClass(TbMsg.class);
then(clusterService).should() then(clusterService).should()
@ -612,7 +627,9 @@ public class DefaultDeviceStateServiceTest {
long newTimeout = System.currentTimeMillis() - deviceState.getLastActivityTime() + increase; long newTimeout = System.currentTimeMillis() - deviceState.getLastActivityTime() + increase;
service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, newTimeout); service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, newTimeout);
verify(telemetrySubscriptionService, never()).saveAttrAndNotify(any(), eq(deviceId), any(AttributeScope.class), eq(ACTIVITY_STATE), any(), any()); verify(telemetrySubscriptionService, never()).save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE)
));
Thread.sleep(defaultTimeout + increase); Thread.sleep(defaultTimeout + increase);
service.checkStates(); service.checkStates();
activityVerify(false); activityVerify(false);
@ -651,7 +668,9 @@ public class DefaultDeviceStateServiceTest {
long newTimeout = 1; long newTimeout = 1;
Thread.sleep(newTimeout); Thread.sleep(newTimeout);
verify(telemetrySubscriptionService, never()).saveAttrAndNotify(any(), eq(deviceId), any(AttributeScope.class), eq(ACTIVITY_STATE), any(), any()); verify(telemetrySubscriptionService, never()).save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE)
));
} }
@Test @Test
@ -672,8 +691,6 @@ public class DefaultDeviceStateServiceTest {
service.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis()); service.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis());
activityVerify(true); activityVerify(true);
verify(telemetrySubscriptionService, never()).saveAttrAndNotify(any(), eq(deviceId), any(AttributeScope.class), eq(ACTIVITY_STATE), any(), any());
long newTimeout = 1; long newTimeout = 1;
Thread.sleep(newTimeout); Thread.sleep(newTimeout);
@ -713,11 +730,17 @@ public class DefaultDeviceStateServiceTest {
long newTimeout = 1; long newTimeout = 1;
service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, newTimeout); service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, newTimeout);
verify(telemetrySubscriptionService, never()).saveAttrAndNotify(any(), eq(deviceId), any(AttributeScope.class), eq(ACTIVITY_STATE), any(), any()); verify(telemetrySubscriptionService, never()).save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE)
));
} }
private void activityVerify(boolean isActive) { private void activityVerify(boolean isActive) {
verify(telemetrySubscriptionService).saveAttrAndNotify(any(), eq(deviceId), any(AttributeScope.class), eq(ACTIVITY_STATE), eq(isActive), any()); verify(telemetrySubscriptionService).save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
request.getEntityId().equals(deviceId) &&
request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) &&
request.getEntries().get(0).getValue().equals(isActive)
));
} }
@Test @Test
@ -763,21 +786,27 @@ public class DefaultDeviceStateServiceTest {
// THEN // THEN
assertThat(deviceState.isActive()).isEqualTo(true); assertThat(deviceState.isActive()).isEqualTo(true);
assertThat(deviceState.getLastActivityTime()).isEqualTo(lastReportedActivity); assertThat(deviceState.getLastActivityTime()).isEqualTo(lastReportedActivity);
then(telemetrySubscriptionService).should().saveAttrAndNotify( then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
any(), eq(deviceId), any(AttributeScope.class), eq(LAST_ACTIVITY_TIME), eq(lastReportedActivity), any() request.getEntityId().equals(deviceId) &&
); request.getEntries().get(0).getKey().equals(LAST_ACTIVITY_TIME) &&
request.getEntries().get(0).getValue().equals(lastReportedActivity)
));
assertThat(deviceState.getLastInactivityAlarmTime()).isEqualTo(expectedInactivityAlarmTime); assertThat(deviceState.getLastInactivityAlarmTime()).isEqualTo(expectedInactivityAlarmTime);
if (shouldSetInactivityAlarmTimeToZero) { if (shouldSetInactivityAlarmTimeToZero) {
then(telemetrySubscriptionService).should().saveAttrAndNotify( then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
any(), eq(deviceId), any(AttributeScope.class), eq(INACTIVITY_ALARM_TIME), eq(0L), any() request.getEntityId().equals(deviceId) &&
); request.getEntries().get(0).getKey().equals(INACTIVITY_ALARM_TIME) &&
request.getEntries().get(0).getValue().equals(0L)
));
} }
if (shouldUpdateActivityStateToActive) { if (shouldUpdateActivityStateToActive) {
then(telemetrySubscriptionService).should().saveAttrAndNotify( then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), eq(ACTIVITY_STATE), eq(true), any() request.getEntityId().equals(deviceId) &&
); request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) &&
request.getEntries().get(0).getValue().equals(true)
));
var msgCaptor = ArgumentCaptor.forClass(TbMsg.class); var msgCaptor = ArgumentCaptor.forClass(TbMsg.class);
then(clusterService).should().pushMsgToRuleEngine(eq(tenantId), eq(deviceId), msgCaptor.capture(), any()); then(clusterService).should().pushMsgToRuleEngine(eq(tenantId), eq(deviceId), msgCaptor.capture(), any());
@ -796,28 +825,28 @@ public class DefaultDeviceStateServiceTest {
private static Stream<Arguments> provideParametersForUpdateActivityState() { private static Stream<Arguments> provideParametersForUpdateActivityState() {
return Stream.of( return Stream.of(
Arguments.of(true, 100, 120, 80, 80, false, false), Arguments.of(true, 100, 120, 80, 80, false, false),
Arguments.of(true, 100, 120, 100, 100, false, false), Arguments.of(true, 100, 120, 100, 100, false, false),
Arguments.of(false, 100, 120, 110, 110, false, true), Arguments.of(false, 100, 120, 110, 110, false, true),
Arguments.of(true, 100, 100, 80, 80, false, false), Arguments.of(true, 100, 100, 80, 80, false, false),
Arguments.of(true, 100, 100, 100, 100, false, false), Arguments.of(true, 100, 100, 100, 100, false, false),
Arguments.of(false, 100, 100, 110, 0, true, true), Arguments.of(false, 100, 100, 110, 0, true, true),
Arguments.of(false, 100, 110, 110, 0, true, true), Arguments.of(false, 100, 110, 110, 0, true, true),
Arguments.of(false, 100, 110, 120, 0, true, true), Arguments.of(false, 100, 110, 120, 0, true, true),
Arguments.of(true, 0, 0, 0, 0, false, false), Arguments.of(true, 0, 0, 0, 0, false, false),
Arguments.of(false, 0, 0, 0, 0, true, true) Arguments.of(false, 0, 0, 0, 0, true, true)
); );
} }
@ -857,9 +886,10 @@ public class DefaultDeviceStateServiceTest {
assertThat(deviceState.getInactivityTimeout()).isEqualTo(newInactivityTimeout); assertThat(deviceState.getInactivityTimeout()).isEqualTo(newInactivityTimeout);
assertThat(deviceState.isActive()).isEqualTo(expectedActivityState); assertThat(deviceState.isActive()).isEqualTo(expectedActivityState);
if (activityState && !expectedActivityState) { if (activityState && !expectedActivityState) {
then(telemetrySubscriptionService).should().saveAttrAndNotify( then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
any(), eq(deviceId), any(AttributeScope.class), eq(ACTIVITY_STATE), eq(false), any() request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) &&
); request.getEntries().get(0).getValue().equals(false)
));
} }
} }
@ -954,9 +984,10 @@ public class DefaultDeviceStateServiceTest {
assertThat(state.getLastInactivityAlarmTime()).isEqualTo(expectedLastInactivityAlarmTime); assertThat(state.getLastInactivityAlarmTime()).isEqualTo(expectedLastInactivityAlarmTime);
if (shouldUpdateActivityStateToInactive) { if (shouldUpdateActivityStateToInactive) {
then(telemetrySubscriptionService).should().saveAttrAndNotify( then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), eq(ACTIVITY_STATE), eq(false), any() request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) &&
); request.getEntries().get(0).getValue().equals(false)
));
var msgCaptor = ArgumentCaptor.forClass(TbMsg.class); var msgCaptor = ArgumentCaptor.forClass(TbMsg.class);
then(clusterService).should().pushMsgToRuleEngine(eq(tenantId), eq(deviceId), msgCaptor.capture(), any()); then(clusterService).should().pushMsgToRuleEngine(eq(tenantId), eq(deviceId), msgCaptor.capture(), any());
@ -971,72 +1002,74 @@ public class DefaultDeviceStateServiceTest {
assertThat(actualNotification.getDeviceId()).isEqualTo(deviceId); assertThat(actualNotification.getDeviceId()).isEqualTo(deviceId);
assertThat(actualNotification.isActive()).isFalse(); assertThat(actualNotification.isActive()).isFalse();
then(telemetrySubscriptionService).should().saveAttrAndNotify( then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), request.getTenantId().equals(TenantId.SYS_TENANT_ID) && request.getEntityId().equals(deviceId) &&
eq(INACTIVITY_ALARM_TIME), eq(expectedLastInactivityAlarmTime), any() request.getScope().equals(AttributeScope.SERVER_SCOPE) &&
); request.getEntries().get(0).getKey().equals(INACTIVITY_ALARM_TIME) &&
request.getEntries().get(0).getValue().equals(expectedLastInactivityAlarmTime)
));
} }
} }
private static Stream<Arguments> provideParametersForUpdateInactivityStateIfExpired() { private static Stream<Arguments> provideParametersForUpdateInactivityStateIfExpired() {
return Stream.of( return Stream.of(
Arguments.of(false, 100, 70, 90, 70, 60, false, 90, false), Arguments.of(false, 100, 70, 90, 70, 60, false, 90, false),
Arguments.of(false, 100, 40, 50, 70, 10, false, 50, false), Arguments.of(false, 100, 40, 50, 70, 10, false, 50, false),
Arguments.of(false, 100, 25, 60, 75, 25, false, 60, false), Arguments.of(false, 100, 25, 60, 75, 25, false, 60, false),
Arguments.of(false, 100, 60, 70, 10, 50, false, 70, false), Arguments.of(false, 100, 60, 70, 10, 50, false, 70, false),
Arguments.of(false, 100, 10, 15, 90, 10, false, 15, false), Arguments.of(false, 100, 10, 15, 90, 10, false, 15, false),
Arguments.of(false, 100, 0, 40, 75, 0, false, 40, false), Arguments.of(false, 100, 0, 40, 75, 0, false, 40, false),
Arguments.of(true, 100, 90, 80, 80, 50, true, 80, false), Arguments.of(true, 100, 90, 80, 80, 50, true, 80, false),
Arguments.of(true, 100, 95, 90, 10, 50, true, 90, false), Arguments.of(true, 100, 95, 90, 10, 50, true, 90, false),
Arguments.of(true, 100, 10, 10, 90, 10, false, 100, true), Arguments.of(true, 100, 10, 10, 90, 10, false, 100, true),
Arguments.of(true, 100, 10, 10, 90, 11, true, 10, false), Arguments.of(true, 100, 10, 10, 90, 11, true, 10, false),
Arguments.of(true, 100, 15, 10, 85, 5, false, 100, true), Arguments.of(true, 100, 15, 10, 85, 5, false, 100, true),
Arguments.of(true, 100, 15, 10, 75, 5, false, 100, true), Arguments.of(true, 100, 15, 10, 75, 5, false, 100, true),
Arguments.of(true, 100, 95, 90, 5, 50, false, 100, true), Arguments.of(true, 100, 95, 90, 5, 50, false, 100, true),
Arguments.of(true, 100, 0, 0, 101, 0, true, 0, false), Arguments.of(true, 100, 0, 0, 101, 0, true, 0, false),
Arguments.of(true, 100, 0, 0, 100, 0, false, 100, true), Arguments.of(true, 100, 0, 0, 100, 0, false, 100, true),
Arguments.of(true, 100, 0, 0, 99, 0, false, 100, true), Arguments.of(true, 100, 0, 0, 99, 0, false, 100, true),
Arguments.of(true, 100, 0, 0, 120, 10, true, 0, false), Arguments.of(true, 100, 0, 0, 120, 10, true, 0, false),
Arguments.of(true, 100, 50, 0, 100, 0, true, 0, false), Arguments.of(true, 100, 50, 0, 100, 0, true, 0, false),
Arguments.of(true, 100, 10, 0, 91, 0, true, 0, false), Arguments.of(true, 100, 10, 0, 91, 0, true, 0, false),
Arguments.of(true, 100, 90, 0, 10, 0, false, 100, true), Arguments.of(true, 100, 90, 0, 10, 0, false, 100, true),
Arguments.of(true, 100, 100, 100, 1, 0, true, 100, false), Arguments.of(true, 100, 100, 100, 1, 0, true, 100, false),
Arguments.of(true, 100, 100, 100, 100, 100, true, 100, false), Arguments.of(true, 100, 100, 100, 100, 100, true, 100, false),
Arguments.of(false, 100, 59, 60, 30, 10, false, 60, false), Arguments.of(false, 100, 59, 60, 30, 10, false, 60, false),
Arguments.of(true, 100, 60, 60, 30, 10, false, 100, true), Arguments.of(true, 100, 60, 60, 30, 10, false, 100, true),
Arguments.of(true, 100, 61, 60, 30, 10, false, 100, true), Arguments.of(true, 100, 61, 60, 30, 10, false, 100, true),
Arguments.of(true, 0, 0, 0, 1, 0, true, 0, false), Arguments.of(true, 0, 0, 0, 1, 0, true, 0, false),
Arguments.of(true, 0, 0, 0, 0, 0, false, 0, true), Arguments.of(true, 0, 0, 0, 0, 0, false, 0, true),
Arguments.of(true, 100, 90, 80, 20, 70, true, 80, false), Arguments.of(true, 100, 90, 80, 20, 70, true, 80, false),
Arguments.of(true, 100, 80, 90, 30, 70, true, 90, false) Arguments.of(true, 100, 80, 90, 30, 70, true, 90, false)
); );
} }
@ -1100,7 +1133,10 @@ public class DefaultDeviceStateServiceTest {
// THEN // THEN
await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> {
assertThat(service.deviceStates.get(deviceId).getState().isActive()).isEqualTo(false); assertThat(service.deviceStates.get(deviceId).getState().isActive()).isEqualTo(false);
then(telemetrySubscriptionService).should().saveAttrAndNotify(eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), eq(ACTIVITY_STATE), eq(false), any()); then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) &&
request.getEntries().get(0).getValue().equals(false)
));
}); });
} }
@ -1127,10 +1163,31 @@ public class DefaultDeviceStateServiceTest {
service.onDeviceActivity(tenantId, deviceId, currentTime); service.onDeviceActivity(tenantId, deviceId, currentTime);
// THEN // THEN
ArgumentCaptor<AttributesSaveRequest> attributeRequestCaptor = ArgumentCaptor.forClass(AttributesSaveRequest.class);
then(telemetrySubscriptionService).should(times(2)).save(attributeRequestCaptor.capture());
await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> {
assertThat(service.deviceStates.get(deviceId).getState().isActive()).isEqualTo(true); assertThat(service.deviceStates.get(deviceId).getState().isActive()).isEqualTo(true);
then(telemetrySubscriptionService).should().saveAttrAndNotify(eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), eq(LAST_ACTIVITY_TIME), eq(currentTime), any());
then(telemetrySubscriptionService).should().saveAttrAndNotify(eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), eq(ACTIVITY_STATE), eq(true), any()); assertThat(attributeRequestCaptor.getAllValues()).hasSize(2)
.anySatisfy(request -> {
assertThat(request.getTenantId()).isEqualTo(TenantId.SYS_TENANT_ID);
assertThat(request.getEntityId()).isEqualTo(deviceId);
assertThat(request.getScope()).isEqualTo(AttributeScope.SERVER_SCOPE);
assertThat(request.getEntries()).singleElement().satisfies(attributeKvEntry -> {
assertThat(attributeKvEntry.getKey()).isEqualTo(LAST_ACTIVITY_TIME);
assertThat(attributeKvEntry.getLongValue()).hasValue(currentTime);
});
})
.anySatisfy(request -> {
assertThat(request.getTenantId()).isEqualTo(TenantId.SYS_TENANT_ID);
assertThat(request.getEntityId()).isEqualTo(deviceId);
assertThat(request.getScope()).isEqualTo(AttributeScope.SERVER_SCOPE);
assertThat(request.getEntries()).singleElement().satisfies(attributeKvEntry -> {
assertThat(attributeKvEntry.getKey()).isEqualTo(ACTIVITY_STATE);
assertThat(attributeKvEntry.getBooleanValue()).hasValue(true);
});
});
}); });
} }
@ -1174,7 +1231,10 @@ public class DefaultDeviceStateServiceTest {
// THEN // THEN
await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> {
assertThat(service.deviceStates.get(deviceId).getState().isActive()).isEqualTo(true); assertThat(service.deviceStates.get(deviceId).getState().isActive()).isEqualTo(true);
then(telemetrySubscriptionService).should().saveAttrAndNotify(eq(TenantId.SYS_TENANT_ID), eq(deviceId), eq(AttributeScope.SERVER_SCOPE), eq(ACTIVITY_STATE), eq(true), any()); then(telemetrySubscriptionService).should().save(argThat((ArgumentMatcher<AttributesSaveRequest>) request ->
request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) &&
request.getEntries().get(0).getValue().equals(true)
));
}); });
} }

115
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java
View File

@ -0,0 +1,115 @@
/**
* Copyright © 2016-2024 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.rule.engine.api;
import com.google.common.util.concurrent.FutureCallback;
import lombok.AccessLevel;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
import org.thingsboard.server.common.data.kv.KvEntry;
import java.util.List;
@Getter
@ToString
@AllArgsConstructor(access = AccessLevel.PRIVATE)
public class AttributesSaveRequest {
private final TenantId tenantId;
private final EntityId entityId;
private final AttributeScope scope;
private final List<AttributeKvEntry> entries; // todo: rename to attributes? same with timeseries
private final boolean notifyDevice;
@Setter
private FutureCallback<Void> callback;
public static Builder builder() {
return new Builder();
}
public static class Builder {
private TenantId tenantId;
private EntityId entityId;
private AttributeScope scope;
private List<AttributeKvEntry> entries;
private boolean notifyDevice = true;
private FutureCallback<Void> callback;
Builder() {}
public Builder tenantId(TenantId tenantId) {
this.tenantId = tenantId;
return this;
}
public Builder entityId(EntityId entityId) {
this.entityId = entityId;
return this;
}
public Builder scope(AttributeScope scope) {
this.scope = scope;
return this;
}
@Deprecated
public Builder scope(String scope) {
try {
this.scope = AttributeScope.valueOf(scope);
} catch (IllegalArgumentException e) {
throw new IllegalArgumentException("Invalid attribute scope '" + scope + "'");
}
return this;
}
public Builder entries(List<AttributeKvEntry> entries) {
this.entries = entries;
return this;
}
public Builder entry(AttributeKvEntry entry) {
return entries(List.of(entry));
}
public Builder entry(KvEntry kvEntry) {
return entry(new BaseAttributeKvEntry(kvEntry, System.currentTimeMillis()));
}
public Builder notifyDevice(boolean notifyDevice) {
this.notifyDevice = notifyDevice;
return this;
}
public Builder callback(FutureCallback<Void> callback) {
this.callback = callback;
return this;
}
public AttributesSaveRequest build() {
return new AttributesSaveRequest(tenantId, entityId, scope, entries, notifyDevice, callback);
}
}
}

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

@ -20,7 +20,6 @@ import com.google.common.util.concurrent.ListenableFuture;
import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.DeleteTsKvQuery; import org.thingsboard.server.common.data.kv.DeleteTsKvQuery;
import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry;
@ -36,33 +35,11 @@ public interface RuleEngineTelemetryService {
ListenableFuture<Void> saveAndNotify(TimeseriesSaveRequest request); ListenableFuture<Void> saveAndNotify(TimeseriesSaveRequest request);
@Deprecated(since = "3.7.0") void save(AttributesSaveRequest request);
void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, FutureCallback<Void> callback);
void saveAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List<AttributeKvEntry> attributes, FutureCallback<Void> callback);
@Deprecated(since = "3.7.0")
void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice, FutureCallback<Void> callback);
void saveAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List<AttributeKvEntry> attributes, boolean notifyDevice, FutureCallback<Void> callback);
void saveLatestAndNotify(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts, FutureCallback<Void> callback); void saveLatestAndNotify(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts, FutureCallback<Void> callback);
ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, long value); ListenableFuture<Void> saveAttrAndNotify(AttributesSaveRequest request);
ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, String value);
ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, double value);
ListenableFuture<Void> saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, boolean value);
void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, long value, FutureCallback<Void> callback);
void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, String value, FutureCallback<Void> callback);
void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, double value, FutureCallback<Void> callback);
void saveAttrAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, String key, boolean value, FutureCallback<Void> callback);
void deleteAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List<String> keys, FutureCallback<Void> callback); void deleteAndNotify(TenantId tenantId, EntityId entityId, AttributeScope scope, List<String> keys, FutureCallback<Void> callback);

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

@ -23,6 +23,8 @@ import lombok.Setter;
import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry;
import java.util.List; import java.util.List;
@ -77,8 +79,11 @@ public class TimeseriesSaveRequest {
} }
public Builder entry(TsKvEntry entry) { public Builder entry(TsKvEntry entry) {
this.entries = List.of(entry); return entries(List.of(entry));
return this; }
public Builder entry(KvEntry kvEntry) {
return entry(new BasicTsKvEntry(System.currentTimeMillis(), kvEntry));
} }
public Builder ttl(long ttl) { public Builder ttl(long ttl) {

10
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java
View File

@ -23,6 +23,7 @@ import com.google.gson.JsonPrimitive;
import jakarta.annotation.Nullable; import jakarta.annotation.Nullable;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.thingsboard.common.util.DonAsynchron; import org.thingsboard.common.util.DonAsynchron;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.rule.engine.api.EmptyNodeConfiguration; import org.thingsboard.rule.engine.api.EmptyNodeConfiguration;
import org.thingsboard.rule.engine.api.RuleNode; import org.thingsboard.rule.engine.api.RuleNode;
import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbContext;
@ -112,8 +113,13 @@ public class TbCopyAttributesToEntityViewNode implements TbNode {
Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(JsonParser.parseString(msg.getData())); Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(JsonParser.parseString(msg.getData()));
List<AttributeKvEntry> filteredAttributes = List<AttributeKvEntry> filteredAttributes =
attributes.stream().filter(attr -> attributeContainsInEntityView(scope, attr.getKey(), entityView)).collect(Collectors.toList()); attributes.stream().filter(attr -> attributeContainsInEntityView(scope, attr.getKey(), entityView)).collect(Collectors.toList());
ctx.getTelemetryService().saveAndNotify(ctx.getTenantId(), entityView.getId(), scope, filteredAttributes, ctx.getTelemetryService().save(AttributesSaveRequest.builder()
getFutureCallback(ctx, msg, entityView)); .tenantId(ctx.getTenantId())
.entityId(entityView.getId())
.scope(scope)
.entries(filteredAttributes)
.callback(getFutureCallback(ctx, msg, entityView))
.build());
} }
} }
} }

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

@ -24,6 +24,7 @@ import net.objecthunter.exp4j.Expression;
import net.objecthunter.exp4j.ExpressionBuilder; import net.objecthunter.exp4j.ExpressionBuilder;
import org.springframework.util.ConcurrentReferenceHashMap; import org.springframework.util.ConcurrentReferenceHashMap;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.rule.engine.api.RuleNode; import org.thingsboard.rule.engine.api.RuleNode;
import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbContext;
import org.thingsboard.rule.engine.api.TbNode; import org.thingsboard.rule.engine.api.TbNode;
@ -38,6 +39,7 @@ import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry; import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.DoubleDataEntry; import org.thingsboard.server.common.data.kv.DoubleDataEntry;
import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.LongDataEntry;
import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsg;
@ -150,15 +152,20 @@ public class TbMathNode implements TbNode {
private ListenableFuture<Void> saveAttribute(TbContext ctx, TbMsg msg, double result, TbMathResult mathResultDef) { private ListenableFuture<Void> saveAttribute(TbContext ctx, TbMsg msg, double result, TbMathResult mathResultDef) {
AttributeScope attributeScope = getAttributeScope(mathResultDef.getAttributeScope()); AttributeScope attributeScope = getAttributeScope(mathResultDef.getAttributeScope());
KvEntry kvEntry;
if (isIntegerResult(mathResultDef, config.getOperation())) { if (isIntegerResult(mathResultDef, config.getOperation())) {
var value = toIntValue(result); var value = toIntValue(result);
return ctx.getTelemetryService().saveAttrAndNotify( kvEntry = new LongDataEntry(mathResultDef.getKey(), value);
ctx.getTenantId(), msg.getOriginator(), attributeScope, mathResultDef.getKey(), value);
} else { } else {
var value = toDoubleValue(mathResultDef, result); var value = toDoubleValue(mathResultDef, result);
return ctx.getTelemetryService().saveAttrAndNotify( kvEntry = new DoubleDataEntry(mathResultDef.getKey(), value);
ctx.getTenantId(), msg.getOriginator(), attributeScope, mathResultDef.getKey(), value);
} }
return ctx.getTelemetryService().saveAttrAndNotify(AttributesSaveRequest.builder()
.tenantId(ctx.getTenantId())
.entityId(msg.getOriginator())
.scope(attributeScope)
.entry(kvEntry)
.build());
} }
private boolean isIntegerResult(TbMathResult mathResultDef, TbRuleNodeMathFunctionType function) { private boolean isIntegerResult(TbMathResult mathResultDef, TbRuleNodeMathFunctionType function) {

31
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java
View File

@ -17,11 +17,13 @@ package org.thingsboard.rule.engine.telemetry;
import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode; import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors; import com.google.common.util.concurrent.MoreExecutors;
import com.google.gson.JsonParser; import com.google.gson.JsonParser;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.thingsboard.common.util.DonAsynchron; import org.thingsboard.common.util.DonAsynchron;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.rule.engine.api.RuleNode; import org.thingsboard.rule.engine.api.RuleNode;
import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbContext;
import org.thingsboard.rule.engine.api.TbNode; import org.thingsboard.rule.engine.api.TbNode;
@ -56,10 +58,10 @@ import static org.thingsboard.server.common.data.msg.TbMsgType.POST_ATTRIBUTES_R
version = 2, version = 2,
nodeDescription = "Saves attributes data", nodeDescription = "Saves attributes data",
nodeDetails = "Saves entity attributes based on configurable scope parameter. Expects messages with 'POST_ATTRIBUTES_REQUEST' message type. " + nodeDetails = "Saves entity attributes based on configurable scope parameter. Expects messages with 'POST_ATTRIBUTES_REQUEST' message type. " +
"If upsert(update/insert) operation is completed successfully rule node will send the incoming message via <b>Success</b> chain, otherwise, <b>Failure</b> chain is used. " + "If upsert(update/insert) operation is completed successfully rule node will send the incoming message via <b>Success</b> chain, otherwise, <b>Failure</b> chain is used. " +
"Additionally if checkbox <b>Send attributes updated notification</b> is set to true, rule node will put the \"Attributes Updated\" " + "Additionally if checkbox <b>Send attributes updated notification</b> is set to true, rule node will put the \"Attributes Updated\" " +
"event for <b>SHARED_SCOPE</b> and <b>SERVER_SCOPE</b> attributes updates to the corresponding rule engine queue." + "event for <b>SHARED_SCOPE</b> and <b>SERVER_SCOPE</b> attributes updates to the corresponding rule engine queue." +
"Performance checkbox 'Save attributes only if the value changes' will skip attributes overwrites for values with no changes (avoid concurrent writes because this check is not transactional; will not update 'Last updated time' for skipped attributes).", "Performance checkbox 'Save attributes only if the value changes' will skip attributes overwrites for values with no changes (avoid concurrent writes because this check is not transactional; will not update 'Last updated time' for skipped attributes).",
uiResources = {"static/rulenode/rulenode-core-config.js"}, uiResources = {"static/rulenode/rulenode-core-config.js"},
configDirective = "tbActionNodeAttributesConfig", configDirective = "tbActionNodeAttributesConfig",
icon = "file_upload" icon = "file_upload"
@ -114,16 +116,17 @@ public class TbMsgAttributesNode implements TbNode {
ctx.tellSuccess(msg); ctx.tellSuccess(msg);
return; return;
} }
ctx.getTelemetryService().saveAndNotify( FutureCallback<Void> callback = sendAttributesUpdateNotification ?
ctx.getTenantId(), new AttributesUpdateNodeCallback(ctx, msg, scope.name(), attributes) :
msg.getOriginator(), new TelemetryNodeCallback(ctx, msg);
scope, ctx.getTelemetryService().save(AttributesSaveRequest.builder()
attributes, .tenantId(ctx.getTenantId())
config.isNotifyDevice() || checkNotifyDeviceMdValue(msg.getMetaData().getValue(NOTIFY_DEVICE_METADATA_KEY)), .entityId(msg.getOriginator())
sendAttributesUpdateNotification ? .scope(scope)
new AttributesUpdateNodeCallback(ctx, msg, scope.name(), attributes) : .entries(attributes)
new TelemetryNodeCallback(ctx, msg) .notifyDevice(config.isNotifyDevice() || checkNotifyDeviceMdValue(msg.getMetaData().getValue(NOTIFY_DEVICE_METADATA_KEY)))
); .callback(callback)
.build());
} }
List<AttributeKvEntry> filterChangedAttr(List<AttributeKvEntry> currentAttributes, List<AttributeKvEntry> newAttributes) { List<AttributeKvEntry> filterChangedAttr(List<AttributeKvEntry> currentAttributes, List<AttributeKvEntry> newAttributes) {

39
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNodeTest.java
View File

@ -24,8 +24,10 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource; import org.junit.jupiter.params.provider.EnumSource;
import org.mockito.ArgumentCaptor; import org.mockito.ArgumentCaptor;
import org.mockito.Mock; import org.mockito.Mock;
import org.mockito.ThrowingConsumer;
import org.mockito.junit.jupiter.MockitoExtension; import org.mockito.junit.jupiter.MockitoExtension;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.rule.engine.api.EmptyNodeConfiguration; import org.thingsboard.rule.engine.api.EmptyNodeConfiguration;
import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService;
import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbContext;
@ -37,7 +39,6 @@ import org.thingsboard.server.common.data.EntityView;
import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.EntityViewId; import org.thingsboard.server.common.data.id.EntityViewId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.data.msg.TbMsgType;
import org.thingsboard.server.common.data.msg.TbNodeConnectionType; import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
import org.thingsboard.server.common.data.objects.AttributesEntityView; import org.thingsboard.server.common.data.objects.AttributesEntityView;
@ -56,6 +57,7 @@ import java.util.UUID;
import static org.assertj.core.api.AssertionsForClassTypes.assertThat; import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.assertArg;
import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verify;
@ -113,10 +115,10 @@ public class TbCopyAttributesToEntityViewNodeTest {
mockEntityViewLookup(entityView); mockEntityViewLookup(entityView);
when(ctxMock.getTelemetryService()).thenReturn(telemetryServiceMock); when(ctxMock.getTelemetryService()).thenReturn(telemetryServiceMock);
doAnswer(invocation -> { doAnswer(invocation -> {
FutureCallback<Void> callback = invocation.getArgument(4); AttributesSaveRequest request = invocation.getArgument(0);
callback.onSuccess(null); request.getCallback().onSuccess(null);
return null; return null;
}).when(telemetryServiceMock).saveAndNotify(any(), any(), any(AttributeScope.class), anyList(), any(FutureCallback.class)); }).when(telemetryServiceMock).save(any(AttributesSaveRequest.class));
TbMsg newMsg = TbMsg.newMsg(msg, msg.getQueueName(), msg.getRuleChainId(), msg.getRuleNodeId()); TbMsg newMsg = TbMsg.newMsg(msg, msg.getQueueName(), msg.getRuleChainId(), msg.getRuleNodeId());
// TODO: use newMsg() with any(TbMsgType.class), replace in other tests as well. // TODO: use newMsg() with any(TbMsgType.class), replace in other tests as well.
doAnswer(invocation -> newMsg).when(ctxMock).newMsg(any(), any(String.class), any(), any(), any(), any()); doAnswer(invocation -> newMsg).when(ctxMock).newMsg(any(), any(String.class), any(), any(), any(), any());
@ -124,13 +126,15 @@ public class TbCopyAttributesToEntityViewNodeTest {
node.onMsg(ctxMock, msg); node.onMsg(ctxMock, msg);
verify(entityViewServiceMock).findEntityViewsByTenantIdAndEntityIdAsync(eq(TENANT_ID), eq(DEVICE_ID)); verify(entityViewServiceMock).findEntityViewsByTenantIdAndEntityIdAsync(eq(TENANT_ID), eq(DEVICE_ID));
ArgumentCaptor<List<AttributeKvEntry>> filteredAttributesCaptor = ArgumentCaptor.forClass(List.class); verify(telemetryServiceMock).save(assertArg((ThrowingConsumer<AttributesSaveRequest>) request -> {
verify(telemetryServiceMock).saveAndNotify(eq(TENANT_ID), eq(ENTITY_VIEW_ID), eq(AttributeScope.CLIENT_SCOPE), assertThat(request.getTenantId()).isEqualTo(TENANT_ID);
filteredAttributesCaptor.capture(), any(FutureCallback.class)); assertThat(request.getEntityId()).isEqualTo(ENTITY_VIEW_ID);
List<AttributeKvEntry> filteredAttributesCaptorValue = filteredAttributesCaptor.getValue(); assertThat(request.getScope()).isEqualTo(AttributeScope.CLIENT_SCOPE);
assertThat(filteredAttributesCaptorValue.size()).isEqualTo(1);
assertThat(filteredAttributesCaptorValue.get(0).getKey()).isEqualTo("clientAttribute1"); assertThat(request.getEntries().size()).isEqualTo(1);
assertThat(filteredAttributesCaptorValue.get(0).getValue()).isEqualTo(100L); assertThat(request.getEntries().get(0).getKey()).isEqualTo("clientAttribute1");
assertThat(request.getEntries().get(0).getValue()).isEqualTo(100L);
}));
verify(ctxMock).ack(eq(msg)); verify(ctxMock).ack(eq(msg));
verify(ctxMock).enqueueForTellNext(eq(newMsg), eq(TbNodeConnectionType.SUCCESS)); verify(ctxMock).enqueueForTellNext(eq(newMsg), eq(TbNodeConnectionType.SUCCESS));
verifyNoMoreInteractions(ctxMock, entityViewServiceMock, telemetryServiceMock); verifyNoMoreInteractions(ctxMock, entityViewServiceMock, telemetryServiceMock);
@ -195,17 +199,22 @@ public class TbCopyAttributesToEntityViewNodeTest {
mockEntityViewLookup(entityView); mockEntityViewLookup(entityView);
when(ctxMock.getTelemetryService()).thenReturn(telemetryServiceMock); when(ctxMock.getTelemetryService()).thenReturn(telemetryServiceMock);
doAnswer(invocation -> { doAnswer(invocation -> {
FutureCallback<Void> callback = invocation.getArgument(4); AttributesSaveRequest request = invocation.getArgument(0);
callback.onSuccess(null); request.getCallback().onSuccess(null);
return null; return null;
}).when(telemetryServiceMock).saveAndNotify(any(), any(), any(AttributeScope.class), anyList(), any(FutureCallback.class)); }).when(telemetryServiceMock).save(any(AttributesSaveRequest.class));
TbMsg newMsg = TbMsg.newMsg(msg, msg.getQueueName(), msg.getRuleChainId(), msg.getRuleNodeId()); TbMsg newMsg = TbMsg.newMsg(msg, msg.getQueueName(), msg.getRuleChainId(), msg.getRuleNodeId());
doAnswer(invocation -> newMsg).when(ctxMock).newMsg(any(), any(String.class), any(), any(), any(), any()); doAnswer(invocation -> newMsg).when(ctxMock).newMsg(any(), any(String.class), any(), any(), any(), any());
node.onMsg(ctxMock, msg); node.onMsg(ctxMock, msg);
verify(entityViewServiceMock).findEntityViewsByTenantIdAndEntityIdAsync(eq(TENANT_ID), eq(DEVICE_ID)); verify(entityViewServiceMock).findEntityViewsByTenantIdAndEntityIdAsync(eq(TENANT_ID), eq(DEVICE_ID));
verify(telemetryServiceMock).saveAndNotify(eq(TENANT_ID), eq(ENTITY_VIEW_ID), eq(AttributeScope.CLIENT_SCOPE), eq(Collections.emptyList()), any(FutureCallback.class)); verify(telemetryServiceMock).save(assertArg((ThrowingConsumer<AttributesSaveRequest>) request -> {
assertThat(request.getTenantId()).isEqualTo(TENANT_ID);
assertThat(request.getEntityId()).isEqualTo(ENTITY_VIEW_ID);
assertThat(request.getScope()).isEqualTo(AttributeScope.CLIENT_SCOPE);
assertThat(request.getEntries().isEmpty()).isTrue();
}));
verify(ctxMock).ack(eq(msg)); verify(ctxMock).ack(eq(msg));
verify(ctxMock).enqueueForTellNext(eq(newMsg), eq(TbNodeConnectionType.SUCCESS)); verify(ctxMock).enqueueForTellNext(eq(newMsg), eq(TbNodeConnectionType.SUCCESS));
verifyNoMoreInteractions(ctxMock, entityViewServiceMock, telemetryServiceMock); verifyNoMoreInteractions(ctxMock, entityViewServiceMock, telemetryServiceMock);

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

@ -46,6 +46,7 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry; import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.DoubleDataEntry; import org.thingsboard.server.common.data.kv.DoubleDataEntry;
import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.LongDataEntry; import org.thingsboard.server.common.data.kv.LongDataEntry;
import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.data.msg.TbMsgType;
import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsg;
@ -69,8 +70,6 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any; 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.argThat;
import static org.mockito.ArgumentMatchers.assertArg; import static org.mockito.ArgumentMatchers.assertArg;
import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.eq;
@ -435,14 +434,16 @@ public class TbMathNodeTest {
TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, originator, TbMsgMetaData.EMPTY, JacksonUtil.newObjectNode().put("a", 5).toString()); TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, originator, TbMsgMetaData.EMPTY, JacksonUtil.newObjectNode().put("a", 5).toString());
when(telemetryService.saveAttrAndNotify(any(), any(), any(AttributeScope.class), anyString(), anyDouble())) when(telemetryService.saveAttrAndNotify(any()))
.thenReturn(Futures.immediateFuture(null)); .thenReturn(Futures.immediateFuture(null));
node.onMsg(ctx, msg); node.onMsg(ctx, msg);
ArgumentCaptor<TbMsg> msgCaptor = ArgumentCaptor.forClass(TbMsg.class); ArgumentCaptor<TbMsg> msgCaptor = ArgumentCaptor.forClass(TbMsg.class);
verify(ctx, timeout(TIMEOUT)).tellSuccess(msgCaptor.capture()); verify(ctx, timeout(TIMEOUT)).tellSuccess(msgCaptor.capture());
verify(telemetryService, times(1)).saveAttrAndNotify(any(), any(), any(AttributeScope.class), anyString(), anyDouble()); verify(telemetryService, times(1)).saveAttrAndNotify(assertArg(request -> {
assertThat(request.getEntries()).singleElement().extracting(KvEntry::getValue).isInstanceOf(Double.class);
}));
TbMsg resultMsg = msgCaptor.getValue(); TbMsg resultMsg = msgCaptor.getValue();
assertNotNull(resultMsg); assertNotNull(resultMsg);
@ -554,7 +555,7 @@ public class TbMathNodeTest {
new TbMathArgument(TbMathArgumentType.MESSAGE_BODY, "a") new TbMathArgument(TbMathArgumentType.MESSAGE_BODY, "a")
); );
TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, originator, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_ARRAY); TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, originator, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_ARRAY);
node.onMsg(ctx, msg); node.onMsg(ctx, msg);
ArgumentCaptor<Throwable> tCaptor = ArgumentCaptor.forClass(Throwable.class); ArgumentCaptor<Throwable> tCaptor = ArgumentCaptor.forClass(Throwable.class);

20
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeTest.java
View File

@ -22,9 +22,10 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.ArgumentCaptor; import org.mockito.ThrowingConsumer;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.rule.engine.AbstractRuleNodeUpgradeTest; import org.thingsboard.rule.engine.AbstractRuleNodeUpgradeTest;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService;
import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbContext;
import org.thingsboard.rule.engine.api.TbNode; import org.thingsboard.rule.engine.api.TbNode;
@ -53,6 +54,7 @@ import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.assertArg;
import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.BDDMockito.willCallRealMethod; import static org.mockito.BDDMockito.willCallRealMethod;
import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mock;
@ -169,17 +171,15 @@ class TbMsgAttributesNodeTest extends AbstractRuleNodeUpgradeTest {
node.saveAttr(testAttrList, ctxMock, testTbMsg, AttributeScope.SHARED_SCOPE, false); node.saveAttr(testAttrList, ctxMock, testTbMsg, AttributeScope.SHARED_SCOPE, false);
ArgumentCaptor<Boolean> notifyDeviceCaptor = ArgumentCaptor.forClass(Boolean.class); verify(telemetryServiceMock, times(1)).save(assertArg((ThrowingConsumer<AttributesSaveRequest>) request -> {
assertThat(request.getTenantId()).isEqualTo(tenantId);
verify(telemetryServiceMock, times(1)).saveAndNotify( assertThat(request.getEntityId()).isEqualTo(deviceId);
eq(tenantId), eq(deviceId), eq(AttributeScope.SHARED_SCOPE), assertThat(request.getScope()).isEqualTo(AttributeScope.SHARED_SCOPE);
eq(testAttrList), notifyDeviceCaptor.capture(), any() assertThat(request.getEntries()).isEqualTo(testAttrList);
); assertThat(request.isNotifyDevice()).isEqualTo(expectedArgumentValue);
boolean notifyDevice = notifyDeviceCaptor.getValue(); }));
assertThat(notifyDevice).isEqualTo(expectedArgumentValue);
} }
// Rule nodes upgrade // Rule nodes upgrade
private static Stream<Arguments> givenFromVersionAndConfig_whenUpgrade_thenVerifyHasChangesAndConfig() { private static Stream<Arguments> givenFromVersionAndConfig_whenUpgrade_thenVerifyHasChangesAndConfig() {
return Stream.of( return Stream.of(

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

@ -25,6 +25,7 @@ import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.ArgumentCaptor; import org.mockito.ArgumentCaptor;
import org.mockito.Mock; import org.mockito.Mock;
import org.mockito.ThrowingConsumer;
import org.mockito.junit.jupiter.MockitoExtension; import org.mockito.junit.jupiter.MockitoExtension;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService;
@ -129,12 +130,12 @@ public class TbMsgTimeseriesNodeTest {
TimeseriesSaveRequest request = invocation.getArgument(0); TimeseriesSaveRequest request = invocation.getArgument(0);
request.getCallback().onSuccess(null); request.getCallback().onSuccess(null);
return null; return null;
}).when(telemetryServiceMock).save(any()); }).when(telemetryServiceMock).save(any(TimeseriesSaveRequest.class));
node.onMsg(ctxMock, msg); node.onMsg(ctxMock, msg);
List<TsKvEntry> expectedList = getTsKvEntriesListWithTs(data, System.currentTimeMillis()); List<TsKvEntry> expectedList = getTsKvEntriesListWithTs(data, System.currentTimeMillis());
verify(telemetryServiceMock).save(assertArg(request -> { verify(telemetryServiceMock).save(assertArg((ThrowingConsumer<TimeseriesSaveRequest>) request -> {
assertThat(request.getTenantId()).isEqualTo(TENANT_ID); assertThat(request.getTenantId()).isEqualTo(TENANT_ID);
assertThat(request.getCustomerId()).isNull(); assertThat(request.getCustomerId()).isNull();
assertThat(request.getEntityId()).isEqualTo(DEVICE_ID); assertThat(request.getEntityId()).isEqualTo(DEVICE_ID);
@ -170,12 +171,12 @@ public class TbMsgTimeseriesNodeTest {
TimeseriesSaveRequest request = invocation.getArgument(0); TimeseriesSaveRequest request = invocation.getArgument(0);
request.getCallback().onSuccess(null); request.getCallback().onSuccess(null);
return null; return null;
}).when(telemetryServiceMock).save(any()); }).when(telemetryServiceMock).save(any(TimeseriesSaveRequest.class));
node.onMsg(ctxMock, msg); node.onMsg(ctxMock, msg);
List<TsKvEntry> expectedList = getTsKvEntriesListWithTs(data, ts); List<TsKvEntry> expectedList = getTsKvEntriesListWithTs(data, ts);
verify(telemetryServiceMock).save(assertArg(request -> { verify(telemetryServiceMock).save(assertArg((ThrowingConsumer<TimeseriesSaveRequest>) request -> {
assertThat(request.getTenantId()).isEqualTo(TENANT_ID); assertThat(request.getTenantId()).isEqualTo(TENANT_ID);
assertThat(request.getCustomerId()).isNull(); assertThat(request.getCustomerId()).isNull();
assertThat(request.getEntityId()).isEqualTo(DEVICE_ID); assertThat(request.getEntityId()).isEqualTo(DEVICE_ID);
@ -208,7 +209,7 @@ public class TbMsgTimeseriesNodeTest {
TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, metadata, data); TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, metadata, data);
node.onMsg(ctxMock, msg); node.onMsg(ctxMock, msg);
verify(telemetryServiceMock).save(assertArg(request -> { verify(telemetryServiceMock).save(assertArg((ThrowingConsumer<TimeseriesSaveRequest>) request -> {
assertThat(request.getTenantId()).isEqualTo(TENANT_ID); assertThat(request.getTenantId()).isEqualTo(TENANT_ID);
assertThat(request.getCustomerId()).isNull(); assertThat(request.getCustomerId()).isNull();
assertThat(request.getEntityId()).isEqualTo(DEVICE_ID); assertThat(request.getEntityId()).isEqualTo(DEVICE_ID);

Write Preview
Loading…
Cancel
Save