From 6905cb530b915898cf90201bde255a3507c20293 Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Fri, 31 Jan 2025 17:12:02 +0200 Subject: [PATCH] added new protos for states --- ...efaultCalculatedFieldExecutionService.java | 23 +-- .../server/service/cf/RocksDBService.java | 28 +++- .../cf/ctx/CalculatedFieldStateService.java | 2 - .../cf/ctx/state/RocksDBStateService.java | 149 ++++++++++++++++-- .../ctx/state/ScriptCalculatedFieldState.java | 3 +- .../ctx/state/SimpleCalculatedFieldState.java | 4 +- .../ctx/state/SingleValueArgumentEntry.java | 14 +- .../cf/ctx/state/TsRollingArgumentEntry.java | 28 +++- .../ctx/state/TsRollingArgumentEntryTest.java | 25 +-- .../server/common/util/ProtoUtils.java | 39 ++++- common/proto/src/main/proto/queue.proto | 25 ++- 11 files changed, 263 insertions(+), 77 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java index 8f948aa265..7367eb611f 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java @@ -71,11 +71,13 @@ import org.thingsboard.server.common.util.ProtoUtils; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.cf.CalculatedFieldService; import org.thingsboard.server.dao.timeseries.TimeseriesService; -import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.AttributeScopeProto; import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityUpdateMsgProto; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldIdProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldLinkedTelemetryMsgProto; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldLinkedTelemetryMsgProto.Builder; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.ComponentLifecycleEvent; import org.thingsboard.server.gen.transport.TransportProtos.ComponentLifecycleMsgProto; @@ -84,7 +86,6 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNot import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto; import org.thingsboard.server.queue.TbQueueCallback; import org.thingsboard.server.queue.TbQueueMsgMetadata; -import org.thingsboard.server.queue.discovery.HashPartitionService; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtx; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService; @@ -383,7 +384,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas log.info("Initializing state for all entities in profile: tenantId=[{}], profileId=[{}]", tenantId, entityId); Map commonArguments = calculatedFieldCtx.getArguments().entrySet().stream() .filter(entry -> entry.getValue().getRefEntityId() != null) - .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + .collect(Collectors.toMap(Entry::getKey, Entry::getValue)); fetchArguments(tenantId, entityId, commonArguments, commonArgs -> { calculatedFieldCache.getEntitiesByProfile(tenantId, entityId).forEach(targetEntityId -> { initializeStateForEntity(calculatedFieldCtx, targetEntityId, commonArgs, callback); @@ -551,7 +552,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas private void executeTelemetryUpdate(CalculatedFieldCtx cfCtx, EntityId entityId, List previousCalculatedFieldIds, Map updatedTelemetry) { log.info("Received telemetry update msg: tenantId=[{}], entityId=[{}], calculatedFieldId=[{}]", cfCtx.getTenantId(), entityId, cfCtx.getCfId()); Map argumentValues = updatedTelemetry.entrySet().stream() - .collect(Collectors.toMap(Map.Entry::getKey, entry -> ArgumentEntry.createSingleValueArgument(entry.getValue()))); + .collect(Collectors.toMap(Entry::getKey, entry -> ArgumentEntry.createSingleValueArgument(entry.getValue()))); // updateOrInitializeState(cfCtx, entityId, argumentValues, previousCalculatedFieldIds); } @@ -677,7 +678,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas if (broadcast) { broadcasts.add(link); } else { - TopicPartitionInfo tpi = partitionService.resolve(HashPartitionService.CALCULATED_FIELD_QUEUE_KEY, link.entityId()); + TopicPartitionInfo tpi = partitionService.resolve(CALCULATED_FIELD_QUEUE_KEY, link.entityId()); unicasts.computeIfAbsent(tpi, k -> new ArrayList<>()).add(link); } } @@ -710,7 +711,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas } private CalculatedFieldLinkedTelemetryMsgProto buildLinkedTelemetryMsgProto(CalculatedFieldTelemetryMsgProto telemetryProto, List links) { - TransportProtos.CalculatedFieldLinkedTelemetryMsgProto.Builder builder = TransportProtos.CalculatedFieldLinkedTelemetryMsgProto.newBuilder(); + Builder builder = CalculatedFieldLinkedTelemetryMsgProto.newBuilder(); builder.setMsg(telemetryProto); for (CalculatedFieldEntityCtxId link : links) { builder.addLinks(toProto(link)); @@ -719,8 +720,10 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas } //TODO: IM: move to utils; - private TransportProtos.CalculatedFieldEntityCtxIdProto toProto(CalculatedFieldEntityCtxId ctxId) { - return TransportProtos.CalculatedFieldEntityCtxIdProto.newBuilder() + private CalculatedFieldEntityCtxIdProto toProto(CalculatedFieldEntityCtxId ctxId) { + return CalculatedFieldEntityCtxIdProto.newBuilder() + .setTenantIdMSB(ctxId.tenantId().getId().getMostSignificantBits()) + .setTenantIdLSB(ctxId.tenantId().getId().getLeastSignificantBits()) .setCalculatedFieldIdMSB(ctxId.cfId().getId().getMostSignificantBits()) .setCalculatedFieldIdLSB(ctxId.cfId().getId().getLeastSignificantBits()) .setEntityType(ctxId.entityId().getEntityType().name()) @@ -890,8 +893,8 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas return telemetryMsg; } - private TransportProtos.CalculatedFieldIdProto toProto(CalculatedFieldId cfId) { - return TransportProtos.CalculatedFieldIdProto.newBuilder() + private CalculatedFieldIdProto toProto(CalculatedFieldId cfId) { + return CalculatedFieldIdProto.newBuilder() .setCalculatedFieldIdMSB(cfId.getId().getMostSignificantBits()) .setCalculatedFieldIdLSB(cfId.getId().getLeastSignificantBits()) .build(); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/RocksDBService.java b/application/src/main/java/org/thingsboard/server/service/cf/RocksDBService.java index 3aed65eced..5eaeff120a 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/RocksDBService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/RocksDBService.java @@ -22,6 +22,8 @@ import org.rocksdb.RocksIterator; import org.rocksdb.WriteOptions; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Service; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; import org.thingsboard.server.utils.RocksDBConfig; import java.nio.charset.StandardCharsets; @@ -49,6 +51,14 @@ public class RocksDBService { } } + public void put(CalculatedFieldEntityCtxIdProto key, CalculatedFieldStateProto value) { + try { + db.put(writeOptions, key.toByteArray(), value.toByteArray()); + } catch (RocksDBException e) { + log.error("Failed to store data to RocksDB", e); + } + } + public void delete(String key) { try { db.delete(writeOptions, key.getBytes(StandardCharsets.UTF_8)); @@ -67,18 +77,20 @@ public class RocksDBService { } } - public Map getAll() { - Map map = new HashMap<>(); + public Map getAll() { + Map results = new HashMap<>(); try (RocksIterator iterator = db.newIterator()) { for (iterator.seekToFirst(); iterator.isValid(); iterator.next()) { - String key = new String(iterator.key(), StandardCharsets.UTF_8); - String value = new String(iterator.value(), StandardCharsets.UTF_8); - map.put(key, value); + try { + CalculatedFieldEntityCtxIdProto key = CalculatedFieldEntityCtxIdProto.parseFrom(iterator.key()); + CalculatedFieldStateProto value = CalculatedFieldStateProto.parseFrom(iterator.value()); + results.put(key, value); + } catch (Exception e) { + log.error("Failed to retrieve data from RocksDB", e); + } } - } catch (Exception e) { - log.error("Failed to retrieve data from RocksDB", e); } - return map; + return results; } } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/CalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/CalculatedFieldStateService.java index 2bf6b7e0f2..4f9a998937 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/CalculatedFieldStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/CalculatedFieldStateService.java @@ -24,8 +24,6 @@ public interface CalculatedFieldStateService { Map restoreStates(); - CalculatedFieldState restoreState(CalculatedFieldEntityCtxId ctxId); - void persistState(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback); void removeState(CalculatedFieldEntityCtxId ctxId); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java index aa82f2fc57..4de15f70bf 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java @@ -19,14 +19,28 @@ import lombok.RequiredArgsConstructor; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Service; import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.server.common.data.StringUtils; +import org.thingsboard.server.common.data.cf.CalculatedFieldType; +import org.thingsboard.server.common.data.id.CalculatedFieldId; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.EntityIdFactory; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.kv.BasicKvEntry; +import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.msg.queue.TbCallback; +import org.thingsboard.server.common.util.ProtoUtils; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; +import org.thingsboard.server.gen.transport.TransportProtos.RollingArgumentProto; +import org.thingsboard.server.gen.transport.TransportProtos.SingleValueArgumentProto; +import org.thingsboard.server.gen.transport.TransportProtos.SingleValueProto; import org.thingsboard.server.service.cf.RocksDBService; -import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtx; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService; import java.util.Map; -import java.util.Optional; +import java.util.TreeMap; +import java.util.UUID; import java.util.stream.Collectors; @Service @@ -40,21 +54,14 @@ public class RocksDBStateService implements CalculatedFieldStateService { public Map restoreStates() { return rocksDBService.getAll().entrySet().stream() .collect(Collectors.toMap( - entry -> JacksonUtil.fromString(entry.getKey(), CalculatedFieldEntityCtxId.class), - entry -> JacksonUtil.fromString(entry.getValue(), CalculatedFieldState.class) + entry -> fromProto(entry.getKey()), + entry -> fromProto(entry.getValue()) )); } @Override - public CalculatedFieldState restoreState(CalculatedFieldEntityCtxId ctxId) { - return Optional.ofNullable(rocksDBService.get(JacksonUtil.writeValueAsString(ctxId))) - .map(storedState -> JacksonUtil.fromString(storedState, CalculatedFieldState.class)) - .orElse(null); - } - - @Override - public void persistState(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback){ - rocksDBService.put(JacksonUtil.writeValueAsString(stateId), JacksonUtil.writeValueAsString(state)); + public void persistState(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) { + rocksDBService.put(toProto(stateId), toProto(stateId, state)); callback.onSuccess(); } @@ -63,4 +70,120 @@ public class RocksDBStateService implements CalculatedFieldStateService { rocksDBService.delete(JacksonUtil.writeValueAsString(ctxId)); } + private CalculatedFieldEntityCtxIdProto toProto(CalculatedFieldEntityCtxId ctxId) { + return CalculatedFieldEntityCtxIdProto.newBuilder() + .setTenantIdMSB(ctxId.tenantId().getId().getMostSignificantBits()) + .setTenantIdLSB(ctxId.tenantId().getId().getLeastSignificantBits()) + .setCalculatedFieldIdMSB(ctxId.cfId().getId().getMostSignificantBits()) + .setCalculatedFieldIdLSB(ctxId.cfId().getId().getLeastSignificantBits()) + .setEntityType(ctxId.entityId().getEntityType().name()) + .setEntityIdMSB(ctxId.entityId().getId().getMostSignificantBits()) + .setEntityIdLSB(ctxId.entityId().getId().getLeastSignificantBits()) + .build(); + } + + private CalculatedFieldEntityCtxId fromProto(CalculatedFieldEntityCtxIdProto ctxIdProto) { + TenantId tenantId = TenantId.fromUUID(new UUID(ctxIdProto.getTenantIdMSB(), ctxIdProto.getTenantIdLSB())); + EntityId entityId = EntityIdFactory.getByTypeAndUuid(ctxIdProto.getEntityType(), new UUID(ctxIdProto.getEntityIdMSB(), ctxIdProto.getEntityIdLSB())); + CalculatedFieldId calculatedFieldId = new CalculatedFieldId(new UUID(ctxIdProto.getCalculatedFieldIdMSB(), ctxIdProto.getCalculatedFieldIdLSB())); + return new CalculatedFieldEntityCtxId(tenantId, calculatedFieldId, entityId); + } + + private CalculatedFieldStateProto toProto(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state) { + CalculatedFieldStateProto.Builder builder = CalculatedFieldStateProto.newBuilder() + .setId(toProto(stateId)) + .setType(state.getType().name()) + .addAllRequiredArguments(state.getRequiredArguments()); + + state.getArguments().forEach((argName, argEntry) -> { + if (argEntry instanceof SingleValueArgumentEntry singleValueArgumentEntry) { + builder.addSingleValueArguments(toSingleValueArgumentProto(argName, singleValueArgumentEntry)); + } else if (argEntry instanceof TsRollingArgumentEntry rollingArgumentEntry) { + builder.addRollingValueArguments(toRollingArgumentProto(argName, rollingArgumentEntry)); + } + }); + + return builder.build(); + } + + private SingleValueArgumentProto toSingleValueArgumentProto(String argName, SingleValueArgumentEntry entry) { + SingleValueProto.Builder singleValueProtoBuilder = SingleValueProto.newBuilder() + .setTs(entry.getTs()); + + if (entry.getVersion() != null) { + singleValueProtoBuilder.setVersion(entry.getVersion()); + } + + KvEntry value = entry.getValue(); + if (value != null) { + singleValueProtoBuilder.setHasV(true) + .setValue(ProtoUtils.toKeyValueProto(value)); + } + + return SingleValueArgumentProto.newBuilder() + .setArgName(argName) + .setValue(singleValueProtoBuilder.build()) + .build(); + } + + private RollingArgumentProto toRollingArgumentProto(String argName, TsRollingArgumentEntry entry) { + RollingArgumentProto.Builder rollingArgumentProtoBuilder = RollingArgumentProto.newBuilder() + .setArgName(argName); + + entry.getTsRecords().forEach((ts, value) -> { + SingleValueProto.Builder singleValueProtoBuilder = SingleValueProto.newBuilder() + .setTs(ts); + + if (value != null) { + singleValueProtoBuilder.setHasV(true) + .setValue(ProtoUtils.toKeyValueProto(value)); + } + + rollingArgumentProtoBuilder.addValues(singleValueProtoBuilder.build()); + }); + + return rollingArgumentProtoBuilder.build(); + } + + private CalculatedFieldState fromProto(CalculatedFieldStateProto proto) { + if (StringUtils.isEmpty(proto.getType())) { + return null; + } + + CalculatedFieldType type = CalculatedFieldType.valueOf(proto.getType()); + + CalculatedFieldState state = switch (type) { + case SIMPLE -> new SimpleCalculatedFieldState(proto.getRequiredArgumentsList()); + case SCRIPT -> new ScriptCalculatedFieldState(proto.getRequiredArgumentsList()); + }; + + proto.getSingleValueArgumentsList().forEach(argProto -> + state.getArguments().put(argProto.getArgName(), fromSingleValueArgumentProto(argProto))); + + if (CalculatedFieldType.SCRIPT.equals(type)) { + proto.getRollingValueArgumentsList().forEach(argProto -> + state.getArguments().put(argProto.getArgName(), fromRollingArgumentProto(argProto))); + } + + return state; + } + + private SingleValueArgumentEntry fromSingleValueArgumentProto(SingleValueArgumentProto proto) { + SingleValueProto valueProto = proto.getValue(); + BasicKvEntry value = valueProto.getHasV() ? ProtoUtils.fromProto(valueProto.getValue()) : null; + + return new SingleValueArgumentEntry(valueProto.getTs(), value, valueProto.getVersion()); + } + + private TsRollingArgumentEntry fromRollingArgumentProto(RollingArgumentProto proto) { + TreeMap tsRecords = new TreeMap<>(); + + proto.getValuesList().forEach(singleValueProto -> { + BasicKvEntry value = singleValueProto.getHasV() ? ProtoUtils.fromProto(singleValueProto.getValue()) : null; + tsRecords.put(singleValueProto.getTs(), value); + }); + + return new TsRollingArgumentEntry(tsRecords); + } + } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java index a3c7efb388..4a24d13c93 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java @@ -24,6 +24,7 @@ import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.common.data.cf.CalculatedFieldType; import org.thingsboard.server.common.data.cf.configuration.Argument; import org.thingsboard.server.common.data.cf.configuration.Output; +import org.thingsboard.server.common.data.kv.BasicKvEntry; import org.thingsboard.server.service.cf.CalculatedFieldResult; import java.util.List; @@ -53,7 +54,7 @@ public class ScriptCalculatedFieldState extends BaseCalculatedFieldState { arguments.forEach((key, argumentEntry) -> { if (argumentEntry instanceof TsRollingArgumentEntry tsRollingEntry) { Argument argument = ctx.getArguments().get(key); - TreeMap tsRecords = tsRollingEntry.getTsRecords(); + TreeMap tsRecords = tsRollingEntry.getTsRecords(); if (tsRecords.size() > argument.getLimit()) { tsRecords.pollFirstEntry(); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java index 8b8fe6e8c7..d233b60512 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java @@ -21,6 +21,7 @@ import lombok.Data; import lombok.NoArgsConstructor; import org.thingsboard.server.common.data.cf.CalculatedFieldType; import org.thingsboard.server.common.data.cf.configuration.Output; +import org.thingsboard.server.common.data.kv.BasicKvEntry; import org.thingsboard.server.service.cf.CalculatedFieldResult; import java.util.List; @@ -52,7 +53,8 @@ public class SimpleCalculatedFieldState extends BaseCalculatedFieldState { for (Map.Entry entry : this.arguments.entrySet()) { try { - expr.setVariable(entry.getKey(), Double.parseDouble(entry.getValue().getValue().toString())); + BasicKvEntry kvEntry = ((SingleValueArgumentEntry) entry.getValue()).getValue(); + expr.setVariable(entry.getKey(), Double.parseDouble(kvEntry.getValueAsString())); } catch (NumberFormatException e) { throw new IllegalArgumentException("Argument '" + entry.getKey() + "' is not a number."); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java index 5117fcab0e..8d4d40d39b 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java @@ -19,6 +19,7 @@ import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.BasicKvEntry; import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.util.ProtoUtils; @@ -33,19 +34,19 @@ public class SingleValueArgumentEntry implements ArgumentEntry { public static final ArgumentEntry EMPTY = new SingleValueArgumentEntry(0); private long ts; - private Object value; + private BasicKvEntry value; private Long version; public SingleValueArgumentEntry(TsKvProto entry) { this.ts = entry.getTs(); this.version = entry.getVersion(); - this.value = ProtoUtils.fromProto(entry).getValue(); + this.value = ProtoUtils.fromProto(entry.getKv()); } public SingleValueArgumentEntry(AttributeValueProto entry) { this.ts = entry.getLastUpdateTs(); this.version = entry.getVersion(); - this.value = ProtoUtils.fromProto(entry).getValue(); + this.value = ProtoUtils.basicKvEntryFromProto(entry); } public SingleValueArgumentEntry(KvEntry entry) { @@ -56,7 +57,7 @@ public class SingleValueArgumentEntry implements ArgumentEntry { this.ts = attributeKvEntry.getLastUpdateTs(); this.version = attributeKvEntry.getVersion(); } - this.value = entry.getValue(); + this.value = ProtoUtils.basicKvEntryFromKvEntry(entry); } /** @@ -72,11 +73,6 @@ public class SingleValueArgumentEntry implements ArgumentEntry { return ArgumentEntryType.SINGLE_VALUE; } - @Override - public Object getValue() { - return value; - } - @Override public ArgumentEntry copy() { return new SingleValueArgumentEntry(this.ts, this.value, this.version); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java index b86a51ca03..6c1a772c44 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java @@ -21,11 +21,15 @@ import lombok.Data; import lombok.NoArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.math.NumberUtils; +import org.thingsboard.server.common.data.kv.BasicKvEntry; +import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; +import org.thingsboard.server.common.util.ProtoUtils; import java.util.List; import java.util.Map; import java.util.TreeMap; +import java.util.stream.Collectors; @Data @NoArgsConstructor @@ -37,10 +41,10 @@ public class TsRollingArgumentEntry implements ArgumentEntry { private static final int MAX_ROLLING_ARGUMENT_ENTRY_SIZE = 1000; - private TreeMap tsRecords = new TreeMap<>(); + private TreeMap tsRecords = new TreeMap<>(); public TsRollingArgumentEntry(List kvEntries) { - kvEntries.forEach(tsKvEntry -> addTsRecord(tsKvEntry.getTs(), tsKvEntry.getValue())); + kvEntries.forEach(tsKvEntry -> addTsRecord(tsKvEntry.getTs(), tsKvEntry)); } /** @@ -58,7 +62,15 @@ public class TsRollingArgumentEntry implements ArgumentEntry { @JsonIgnore @Override public Object getValue() { - return tsRecords; + return tsRecords.entrySet() + .stream() + .collect(Collectors.toMap( + Map.Entry::getKey, + entry -> entry.getValue().getValue(), + (oldValue, newValue) -> oldValue, + TreeMap::new + )); + } @Override @@ -79,7 +91,7 @@ public class TsRollingArgumentEntry implements ArgumentEntry { private boolean updateTsRollingEntry(TsRollingArgumentEntry tsRollingEntry) { boolean updated = false; - for (Map.Entry tsRecordEntry : tsRollingEntry.getTsRecords().entrySet()) { + for (Map.Entry tsRecordEntry : tsRollingEntry.getTsRecords().entrySet()) { updated |= addTsRecordIfAbsent(tsRecordEntry.getKey(), tsRecordEntry.getValue()); } return updated; @@ -89,7 +101,7 @@ public class TsRollingArgumentEntry implements ArgumentEntry { return addTsRecordIfAbsent(singleValueEntry.getTs(), singleValueEntry.getValue()); } - private boolean addTsRecordIfAbsent(Long ts, Object value) { + private boolean addTsRecordIfAbsent(Long ts, KvEntry value) { if (!tsRecords.containsKey(ts)) { addTsRecord(ts, value); return true; @@ -97,9 +109,9 @@ public class TsRollingArgumentEntry implements ArgumentEntry { return false; } - private void addTsRecord(Long ts, Object value) { - if (NumberUtils.isParsable(value.toString())) { - tsRecords.put(ts, value); + private void addTsRecord(Long ts, KvEntry value) { + if (NumberUtils.isParsable(value.getValue().toString())) { + tsRecords.put(ts, ProtoUtils.basicKvEntryFromKvEntry(value)); if (tsRecords.size() > MAX_ROLLING_ARGUMENT_ENTRY_SIZE) { tsRecords.pollFirstEntry(); } diff --git a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntryTest.java b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntryTest.java index b08c5f2a58..9ca242092d 100644 --- a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntryTest.java +++ b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntryTest.java @@ -17,6 +17,9 @@ package org.thingsboard.server.service.cf.ctx.state; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.thingsboard.server.common.data.kv.BasicKvEntry; +import org.thingsboard.server.common.data.kv.DoubleDataEntry; +import org.thingsboard.server.common.data.kv.StringDataEntry; import java.util.Map; import java.util.TreeMap; @@ -32,10 +35,10 @@ public class TsRollingArgumentEntryTest { @BeforeEach void setUp() { - TreeMap values = new TreeMap<>(); - values.put(ts - 40, 10); - values.put(ts - 30, 12); - values.put(ts - 20, 17); + TreeMap values = new TreeMap<>(); + values.put(ts - 40, new DoubleDataEntry("key", 10.0)); + values.put(ts - 30, new DoubleDataEntry("key", 12.0)); + values.put(ts - 20, new DoubleDataEntry("key", 17.0)); entry = new TsRollingArgumentEntry(values); } @@ -47,7 +50,7 @@ public class TsRollingArgumentEntryTest { @Test void testUpdateEntryWhenSingleValueEntryPassed() { - SingleValueArgumentEntry newEntry = new SingleValueArgumentEntry(ts - 10, 23, 123L); + SingleValueArgumentEntry newEntry = new SingleValueArgumentEntry(ts - 10, new DoubleDataEntry("key", 23.0), 123L); assertThat(entry.updateEntry(newEntry)).isTrue(); assertThat(entry.getTsRecords()).hasSize(4); @@ -56,7 +59,7 @@ public class TsRollingArgumentEntryTest { @Test void testUpdateEntryWhenSingleValueEntryWithTheSameTsPassed() { - SingleValueArgumentEntry newEntry = new SingleValueArgumentEntry(ts - 20, 23, 123L); + SingleValueArgumentEntry newEntry = new SingleValueArgumentEntry(ts - 20, new DoubleDataEntry("key", 23.0), 123L); assertThat(entry.updateEntry(newEntry)).isFalse(); } @@ -64,10 +67,10 @@ public class TsRollingArgumentEntryTest { @Test void testUpdateEntryWhenRollingEntryPassed() { TsRollingArgumentEntry newEntry = new TsRollingArgumentEntry(); - TreeMap values = new TreeMap<>(); - values.put(ts - 20, 16); - values.put(ts - 10, 7); - values.put(ts - 5, 1); + TreeMap values = new TreeMap<>(); + values.put(ts - 20, new DoubleDataEntry("key", 16.0)); + values.put(ts - 10, new DoubleDataEntry("key", 7.0)); + values.put(ts - 5, new DoubleDataEntry("key", 1.0)); newEntry.setTsRecords(values); assertThat(entry.updateEntry(newEntry)).isTrue(); @@ -83,7 +86,7 @@ public class TsRollingArgumentEntryTest { @Test void testUpdateEntryWhenValueIsNotNumber() { - SingleValueArgumentEntry newEntry = new SingleValueArgumentEntry(ts - 10, "string", 123L); + SingleValueArgumentEntry newEntry = new SingleValueArgumentEntry(ts - 10, new StringDataEntry("key", "string"), 123L); assertThatThrownBy(() -> entry.updateEntry(newEntry)) .isInstanceOf(IllegalArgumentException.class) diff --git a/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java b/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java index 7ba11d8ea9..91e3276210 100644 --- a/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java +++ b/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java @@ -58,6 +58,7 @@ import org.thingsboard.server.common.data.id.TenantProfileId; import org.thingsboard.server.common.data.kv.AttributeKey; import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; +import org.thingsboard.server.common.data.kv.BasicKvEntry; import org.thingsboard.server.common.data.kv.BasicTsKvEntry; import org.thingsboard.server.common.data.kv.BooleanDataEntry; import org.thingsboard.server.common.data.kv.DoubleDataEntry; @@ -90,7 +91,7 @@ import org.thingsboard.server.common.msg.rule.engine.DeviceDeleteMsg; import org.thingsboard.server.common.msg.rule.engine.DeviceEdgeUpdateMsg; import org.thingsboard.server.common.msg.rule.engine.DeviceNameOrTypeUpdateMsg; import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto; +import org.thingsboard.server.gen.transport.TransportProtos.KeyValueProto; import java.util.ArrayList; import java.util.Arrays; @@ -630,6 +631,42 @@ public class ProtoUtils { return new BaseAttributeKvEntry(entry, proto.getLastUpdateTs(), proto.hasVersion() ? proto.getVersion() : null); } + public static BasicKvEntry basicKvEntryFromProto(TransportProtos.AttributeValueProto proto) { + boolean hasValue = proto.getHasV(); + String key = proto.getKey(); + return switch (proto.getType()) { + case BOOLEAN_V -> new BooleanDataEntry(key, hasValue ? proto.getBoolV() : null); + case LONG_V -> new LongDataEntry(key, hasValue ? proto.getLongV() : null); + case DOUBLE_V -> new DoubleDataEntry(key, hasValue ? proto.getDoubleV() : null); + case STRING_V -> new StringDataEntry(key, hasValue ? proto.getStringV() : null); + case JSON_V -> new JsonDataEntry(key, hasValue ? proto.getJsonV() : null); + default -> null; + }; + } + + public static BasicKvEntry fromProto(KeyValueProto proto) { + String key = proto.getKey(); + return switch (proto.getType()) { + case BOOLEAN_V -> new BooleanDataEntry(key, proto.getBoolV()); + case LONG_V -> new LongDataEntry(key, proto.getLongV()); + case DOUBLE_V -> new DoubleDataEntry(key, proto.getDoubleV()); + case STRING_V -> new StringDataEntry(key, proto.getStringV()); + case JSON_V -> new JsonDataEntry(key, proto.getJsonV()); + default -> null; + }; + } + + public static BasicKvEntry basicKvEntryFromKvEntry(KvEntry kvEntry) { + String key = kvEntry.getKey(); + return switch (kvEntry.getDataType()) { + case BOOLEAN -> new BooleanDataEntry(key, kvEntry.getBooleanValue().orElse(null)); + case LONG -> new LongDataEntry(key, kvEntry.getLongValue().orElse(null)); + case DOUBLE -> new DoubleDataEntry(key, kvEntry.getDoubleValue().orElse(null)); + case STRING -> new StringDataEntry(key, kvEntry.getStrValue().orElse(null)); + case JSON -> new JsonDataEntry(key, kvEntry.getJsonValue().orElse(null)); + }; + } + public static TsKvEntry fromProto(TransportProtos.TsKvProto proto) { TransportProtos.KeyValueProto kvProto = proto.getKv(); String key = kvProto.getKey(); diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 53045fc147..2ed535495a 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -810,11 +810,13 @@ message CalculatedFieldLinkedTelemetryMsgProto { } message CalculatedFieldEntityCtxIdProto { - int64 calculatedFieldIdMSB = 1; - int64 calculatedFieldIdLSB = 2; - string entityType = 3; - int64 entityIdMSB = 4; - int64 entityIdLSB = 5; + int64 tenantIdMSB = 1; + int64 tenantIdLSB = 2; + int64 calculatedFieldIdMSB = 3; + int64 calculatedFieldIdLSB = 4; + string entityType = 5; + int64 entityIdMSB = 6; + int64 entityIdLSB = 7; } message CalculatedFieldIdProto { @@ -825,13 +827,8 @@ message CalculatedFieldIdProto { message SingleValueProto { int64 ts = 1; int64 version = 2; - KeyValueType type = 3; bool has_v = 4; - bool bool_v = 5; - int64 long_v = 6; - double double_v = 7; - string string_v = 8; - string json_v = 9; + KeyValueProto value = 5; } message SingleValueArgumentProto { @@ -847,8 +844,10 @@ message RollingArgumentProto { message CalculatedFieldStateProto { CalculatedFieldEntityCtxIdProto id = 1; // int32 version = 2; - repeated SingleValueArgumentProto singleValueArguments = 3; - repeated RollingArgumentProto rollingValueArguments = 4; + string type = 3; + repeated string requiredArguments = 4; + repeated SingleValueArgumentProto singleValueArguments = 5; + repeated RollingArgumentProto rollingValueArguments = 6; } //Used to report session state to tb-Service and persist this state in the cache on the tb-Service level.