From d97d932cfbe3bc25705f73b6a1f57af15d1b4419 Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Wed, 22 Oct 2025 10:10:08 +0300 Subject: [PATCH] refactoring --- ...alculatedFieldManagerMessageProcessor.java | 33 +++++++------- .../cf/TelemetryCalculatedFieldResult.java | 8 ++-- .../ctx/state/BaseCalculatedFieldState.java | 18 +++++--- .../cf/ctx/state/CalculatedFieldCtx.java | 32 +++++--------- ...titiesAggregationCalculatedFieldState.java | 16 +++---- .../RelatedEntitiesArgumentEntry.java | 4 ++ .../aggregation/function/BaseAggEntry.java | 4 +- .../aggregation/function/MaxAggEntry.java | 2 +- .../queue/DefaultTbClusterService.java | 7 +-- .../server/utils/CalculatedFieldUtils.java | 43 ++++++++++--------- .../data/plugin/ComponentLifecycleEvent.java | 4 +- common/proto/src/main/proto/queue.proto | 9 ++-- .../script/api/tbel/TbelCfArg.java | 2 +- .../TbelCfRelatedEntitiesAggregation.java | 2 +- 14 files changed, 95 insertions(+), 89 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java index 7f1c7b1925..3d61825ebe 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java @@ -187,9 +187,18 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } public void onEntityLifecycleMsg(CalculatedFieldEntityLifecycleMsg msg) throws CalculatedFieldException { + var event = msg.getData().getEvent(); + if (msg.getData().isRelationChanged()) { + log.debug("Processing relation [{}] event: ", msg.getData().getEvent()); + switch (event) { + case RELATION_UPDATED -> onRelationUpdated(msg.getData(), msg.getCallback()); + case RELATION_DELETED -> onRelationDeleted(msg.getData(), msg.getCallback()); + default -> msg.getCallback().onSuccess(); + } + return; + } log.debug("Processing entity lifecycle event: [{}] for entity: [{}]", msg.getData().getEvent(), msg.getData().getEntityId()); var entityType = msg.getData().getEntityId().getEntityType(); - var event = msg.getData().getEvent(); switch (entityType) { case CALCULATED_FIELD -> { switch (event) { @@ -280,26 +289,20 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } } else if (msg.isOwnerChanged()) { onEntityOwnerChanged(msg, callback); - } else if (msg.isRelationChanged()) { - onRelationUpdated(msg, callback); } else { callback.onSuccess(); } } private void onEntityDeleted(ComponentLifecycleMsg msg, TbCallback callback) { - if (msg.isRelationChanged()) { - onRelationDeleted(msg, callback); - } else { - switch (msg.getEntityId().getEntityType()) { - case DEVICE, ASSET -> entityProfileCache.removeEntityId(msg.getEntityId()); - case CUSTOMER -> ownerEntities.remove(msg.getEntityId()); - } - ownerEntities.values().forEach(entities -> entities.remove(msg.getEntityId())); - if (isMyPartition(msg.getEntityId(), callback)) { - log.debug("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId()); - getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback)); - } + switch (msg.getEntityId().getEntityType()) { + case DEVICE, ASSET -> entityProfileCache.removeEntityId(msg.getEntityId()); + case CUSTOMER -> ownerEntities.remove(msg.getEntityId()); + } + ownerEntities.values().forEach(entities -> entities.remove(msg.getEntityId())); + if (isMyPartition(msg.getEntityId(), callback)) { + log.debug("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId()); + getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback)); } } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/TelemetryCalculatedFieldResult.java b/application/src/main/java/org/thingsboard/server/service/cf/TelemetryCalculatedFieldResult.java index 1ad666eac5..d59ec9cca9 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/TelemetryCalculatedFieldResult.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/TelemetryCalculatedFieldResult.java @@ -39,6 +39,8 @@ public final class TelemetryCalculatedFieldResult implements CalculatedFieldResu private final AttributeScope scope; private final JsonNode result; + public static TelemetryCalculatedFieldResult EMPTY = TelemetryCalculatedFieldResult.builder().result(null).build(); + @Override public TbMsg toTbMsg(EntityId entityId, List cfIds) { TbMsgType msgType = switch (type) { @@ -66,9 +68,9 @@ public final class TelemetryCalculatedFieldResult implements CalculatedFieldResu @Override public boolean isEmpty() { return result == null || result.isMissingNode() || result.isNull() || - (result.isObject() && result.isEmpty()) || - (result.isArray() && result.isEmpty()) || - (result.isTextual() && result.asText().isEmpty()); + (result.isObject() && result.isEmpty()) || + (result.isArray() && result.isEmpty()) || + (result.isTextual() && result.asText().isEmpty()); } } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java index d48ed9c268..d8f13e6a20 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java @@ -20,10 +20,10 @@ import lombok.Getter; import lombok.Setter; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.actors.TbActorRef; -import org.thingsboard.server.common.data.cf.CalculatedFieldType; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; +import org.thingsboard.server.service.cf.ctx.state.aggregation.RelatedEntitiesArgumentEntry; import org.thingsboard.server.utils.CalculatedFieldUtils; import java.io.Closeable; @@ -75,9 +75,13 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState, ArgumentEntry existingEntry = arguments.get(key); boolean entryUpdated; - if (existingEntry == null || !ctx.getCfType().equals(CalculatedFieldType.RELATED_ENTITIES_AGGREGATION) && newEntry.isForceResetPrevious()) { + if (existingEntry == null || newEntry.isForceResetPrevious()) { validateNewEntry(key, newEntry); - arguments.put(key, newEntry); + if (existingEntry instanceof RelatedEntitiesArgumentEntry relatedEntitiesArgumentEntry) { + relatedEntitiesArgumentEntry.updateEntry(newEntry); + } else { + arguments.put(key, newEntry); + } entryUpdated = true; } else { entryUpdated = existingEntry.updateEntry(newEntry); @@ -110,7 +114,7 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState, @Override public boolean isReady() { return arguments.keySet().containsAll(requiredArguments) && - arguments.values().stream().noneMatch(ArgumentEntry::isEmpty); + arguments.values().stream().noneMatch(ArgumentEntry::isEmpty); } @Override @@ -122,9 +126,11 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState, } @Override - public void close() {} + public void close() { + } - protected void validateNewEntry(String key, ArgumentEntry newEntry) {} + protected void validateNewEntry(String key, ArgumentEntry newEntry) { + } protected ObjectNode toSimpleResult(boolean useLatestTs, ObjectNode valuesNode) { if (!useLatestTs) { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java index aab858d85d..40e414920a 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java @@ -263,33 +263,23 @@ public class CalculatedFieldCtx { } public ListenableFuture evaluateTbelExpression(String expression, CalculatedFieldState state) { - return evaluateTbelExpression(tbelExpressions.get(expression), state); + return evaluateTbelExpression(tbelExpressions.get(expression), state.getArguments(), state.getLatestTimestamp()); } public ListenableFuture evaluateTbelExpression(CalculatedFieldScriptEngine expression, CalculatedFieldState state) { - Map arguments = new LinkedHashMap<>(); - List args = new ArrayList<>(argNames.size() + 1); - args.add(new Object()); // first element is a ctx, but we will set it later; - for (String argName : argNames) { - var arg = toTbelArgument(argName, state); - arguments.put(argName, arg); - if (arg instanceof TbelCfSingleValueArg svArg) { - args.add(svArg.getValue()); - } else { - args.add(arg); - } - } - args.set(0, new TbelCfCtx(arguments, state.getLatestTimestamp())); - - return expression.executeScriptAsync(args.toArray()); + return evaluateTbelExpression(expression, state.getArguments(), state.getLatestTimestamp()); } public ListenableFuture evaluateTbelExpression(String expression, Map entries, long latestTimestamp) { + return evaluateTbelExpression(tbelExpressions.get(expression), entries, latestTimestamp); + } + + public ListenableFuture evaluateTbelExpression(CalculatedFieldScriptEngine expression, Map entries, long latestTimestamp) { Map arguments = new LinkedHashMap<>(); List args = new ArrayList<>(argNames.size() + 1); args.add(new Object()); // first element is a ctx, but we will set it later; for (String argName : argNames) { - var arg = entries.get(argName).toTbelCfArg(); + var arg = toTbelArgument(argName, entries); arguments.put(argName, arg); if (arg instanceof TbelCfSingleValueArg svArg) { args.add(svArg.getValue()); @@ -299,7 +289,7 @@ public class CalculatedFieldCtx { } args.set(0, new TbelCfCtx(arguments, latestTimestamp)); - return tbelExpressions.get(expression).executeScriptAsync(args.toArray()); + return expression.executeScriptAsync(args.toArray()); } public ScheduledFuture scheduleReevaluation(long delayMs, TbActorRef actorCtx) { @@ -308,8 +298,8 @@ public class CalculatedFieldCtx { return systemContext.scheduleMsgWithDelay(actorCtx, new CalculatedFieldReevaluateMsg(tenantId, this), delayMs); } - private TbelCfArg toTbelArgument(String key, CalculatedFieldState state) { - return state.getArguments().get(key).toTbelCfArg(); + private TbelCfArg toTbelArgument(String key, Map arguments) { + return arguments.get(key).toTbelCfArg(); } private void initTbelExpression(String expression) { @@ -658,7 +648,7 @@ public class CalculatedFieldCtx { yield true; } yield geofencingState.getLastDynamicArgumentsRefreshTs() < - System.currentTimeMillis() - scheduledUpdateIntervalMillis; + System.currentTimeMillis() - scheduledUpdateIntervalMillis; } default -> false; }; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesAggregationCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesAggregationCalculatedFieldState.java index 8e78824c7c..c8731b71f7 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesAggregationCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesAggregationCalculatedFieldState.java @@ -42,6 +42,8 @@ import java.util.HashMap; import java.util.Map; import java.util.Map.Entry; +import static java.util.concurrent.TimeUnit.SECONDS; + @Slf4j @Getter public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculatedFieldState { @@ -50,7 +52,7 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat private long lastArgsRefreshTs = -1; @Setter private long lastMetricsEvalTs = -1; - private long deduplicationInterval = -1; + private long deduplicationIntervalMs = -1; private Map metrics; public RelatedEntitiesAggregationCalculatedFieldState(EntityId entityId) { @@ -62,7 +64,7 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat super.setCtx(ctx, actorCtx); var configuration = (RelatedEntitiesAggregationCalculatedFieldConfiguration) ctx.getCalculatedField().getConfiguration(); metrics = configuration.getMetrics(); - deduplicationInterval = configuration.getDeduplicationIntervalInSec(); + deduplicationIntervalMs = SECONDS.toMillis(configuration.getDeduplicationIntervalInSec()); } @Override @@ -76,7 +78,7 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat @Override public void init() { super.init(); - ctx.scheduleReevaluation(deduplicationInterval, actorCtx); + ctx.scheduleReevaluation(deduplicationIntervalMs, actorCtx); } @Override @@ -97,16 +99,14 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat Output output = ctx.getOutput(); ObjectNode aggResult = aggregateMetrics(output); lastMetricsEvalTs = System.currentTimeMillis(); - ctx.scheduleReevaluation(deduplicationInterval, actorCtx); + ctx.scheduleReevaluation(deduplicationIntervalMs, actorCtx); return Futures.immediateFuture(TelemetryCalculatedFieldResult.builder() .type(output.getType()) .scope(output.getScope()) .result(toSimpleResult(ctx.isUseLatestTs(), aggResult)) .build()); } else { - return Futures.immediateFuture(TelemetryCalculatedFieldResult.builder() - .result(null) - .build()); + return Futures.immediateFuture(TelemetryCalculatedFieldResult.EMPTY); } } @@ -125,7 +125,7 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat } private boolean shouldRecalculate() { - boolean intervalPassed = lastMetricsEvalTs <= System.currentTimeMillis() - deduplicationInterval; + boolean intervalPassed = lastMetricsEvalTs <= System.currentTimeMillis() - deduplicationIntervalMs; boolean argsUpdatedDuringInterval = lastArgsRefreshTs > lastMetricsEvalTs; return intervalPassed && argsUpdatedDuringInterval; } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesArgumentEntry.java index 45b7755af4..5b97b1ed0a 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesArgumentEntry.java @@ -50,6 +50,10 @@ public class RelatedEntitiesArgumentEntry implements ArgumentEntry { aggInputs.putAll(relatedEntitiesArgumentEntry.aggInputs); return true; } else if (entry instanceof SingleValueArgumentEntry singleValueArgumentEntry) { + if (entry.isForceResetPrevious()) { + aggInputs.put(singleValueArgumentEntry.getEntityId(), singleValueArgumentEntry); + return true; + } ArgumentEntry argumentEntry = aggInputs.get(singleValueArgumentEntry.getEntityId()); if (argumentEntry != null) { argumentEntry.updateEntry(singleValueArgumentEntry); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/function/BaseAggEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/function/BaseAggEntry.java index b320435e99..8ca523938d 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/function/BaseAggEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/function/BaseAggEntry.java @@ -43,8 +43,8 @@ public abstract class BaseAggEntry implements AggEntry { protected double extractDoubleValue(Object value) { try { - if (value instanceof Number) { - return ((Number) value).doubleValue(); + if (value instanceof Number number) { + return number.doubleValue(); } return Double.parseDouble(value.toString()); } catch (Exception e) { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/function/MaxAggEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/function/MaxAggEntry.java index ddc47daf33..6d734a5a08 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/function/MaxAggEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/function/MaxAggEntry.java @@ -20,7 +20,7 @@ import org.thingsboard.server.common.data.cf.configuration.aggregation.AggFuncti public class MaxAggEntry extends BaseAggEntry { - private double max = -Double.MAX_VALUE; + private double max = Double.MIN_VALUE; @Override protected void doUpdate(double value) { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java index 22faeaf61b..5b6596c9c6 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java @@ -741,7 +741,7 @@ public class DefaultTbClusterService implements TbClusterService { .tenantId(tenantId) .entityId(entityRelation.getFrom()) .relationChanged(true) - .event(ComponentLifecycleEvent.UPDATED) + .event(ComponentLifecycleEvent.RELATION_UPDATED) .info(JacksonUtil.valueToTree(entityRelation)) .build(); broadcast(msg); @@ -753,7 +753,7 @@ public class DefaultTbClusterService implements TbClusterService { .tenantId(tenantId) .entityId(entityRelation.getFrom()) .relationChanged(true) - .event(ComponentLifecycleEvent.DELETED) + .event(ComponentLifecycleEvent.RELATION_DELETED) .info(JacksonUtil.valueToTree(entityRelation)) .build(); broadcast(msg); @@ -809,7 +809,8 @@ public class DefaultTbClusterService implements TbClusterService { private void pushDeviceUpdateMessage(TenantId tenantId, EdgeId edgeId, EntityId entityId, EdgeEventActionType action) { log.trace("{} Going to send edge update notification for device actor, device id {}, edge id {}", tenantId, entityId, edgeId); switch (action) { - case ASSIGNED_TO_EDGE -> pushMsgToCore(new DeviceEdgeUpdateMsg(tenantId, new DeviceId(entityId.getId()), edgeId), null); + case ASSIGNED_TO_EDGE -> + pushMsgToCore(new DeviceEdgeUpdateMsg(tenantId, new DeviceId(entityId.getId()), edgeId), null); case UNASSIGNED_FROM_EDGE -> { EdgeId relatedEdgeId = findRelatedEdgeIdIfAny(tenantId, entityId); pushMsgToCore(new DeviceEdgeUpdateMsg(tenantId, new DeviceId(entityId.getId()), relatedEdgeId), null); diff --git a/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java index d75ec8a70a..ba3aa3fd53 100644 --- a/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java +++ b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java @@ -34,7 +34,6 @@ import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldIdPro import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; import org.thingsboard.server.gen.transport.TransportProtos.GeofencingArgumentProto; import org.thingsboard.server.gen.transport.TransportProtos.GeofencingZoneProto; -import org.thingsboard.server.gen.transport.TransportProtos.RelatedEntitiesAggregationStateProto; import org.thingsboard.server.gen.transport.TransportProtos.SingleValueArgumentProto; import org.thingsboard.server.gen.transport.TransportProtos.TsDoubleValProto; import org.thingsboard.server.gen.transport.TransportProtos.TsRollingArgumentProto; @@ -96,16 +95,18 @@ public class CalculatedFieldUtils { .setId(toProto(stateId)) .setType(state.getType().name()); - RelatedEntitiesAggregationStateProto.Builder aggBuilder = RelatedEntitiesAggregationStateProto.newBuilder(); state.getArguments().forEach((argName, argEntry) -> { switch (argEntry.getType()) { - case SINGLE_VALUE -> builder.addSingleValueArguments(toSingleValueArgumentProto(argName, (SingleValueArgumentEntry) argEntry)); - case TS_ROLLING -> builder.addRollingValueArguments(toRollingArgumentProto(argName, (TsRollingArgumentEntry) argEntry)); - case GEOFENCING -> builder.addGeofencingArguments(toGeofencingArgumentProto(argName, (GeofencingArgumentEntry) argEntry)); + case SINGLE_VALUE -> + builder.addSingleValueArguments(toSingleValueArgumentProto(argName, (SingleValueArgumentEntry) argEntry)); + case TS_ROLLING -> + builder.addRollingValueArguments(toRollingArgumentProto(argName, (TsRollingArgumentEntry) argEntry)); + case GEOFENCING -> + builder.addGeofencingArguments(toGeofencingArgumentProto(argName, (GeofencingArgumentEntry) argEntry)); case RELATED_ENTITIES -> { RelatedEntitiesArgumentEntry relatedEntitiesArgumentEntry = (RelatedEntitiesArgumentEntry) argEntry; relatedEntitiesArgumentEntry.getAggInputs() - .forEach((entityId, entry) -> aggBuilder.addAggArguments(toSingleValueArgumentProto(argName, (SingleValueArgumentEntry) entry))); + .forEach((entityId, entry) -> builder.addSingleValueArguments(toSingleValueArgumentProto(argName, (SingleValueArgumentEntry) entry))); } } }); @@ -119,8 +120,7 @@ public class CalculatedFieldUtils { } } if (state instanceof RelatedEntitiesAggregationCalculatedFieldState aggState) { - aggBuilder.setLastArgsUpdateTs(aggState.getLastArgsRefreshTs()); - builder.setRelatedEntitiesAggregationState(aggBuilder.build()); + builder.setLastArgsUpdateTs(aggState.getLastArgsRefreshTs()); } return builder.build(); } @@ -208,6 +208,20 @@ public class CalculatedFieldUtils { case RELATED_ENTITIES_AGGREGATION -> new RelatedEntitiesAggregationCalculatedFieldState(id.entityId()); }; + if (state instanceof RelatedEntitiesAggregationCalculatedFieldState relatedEntitiesAggState) { + Map> arguments = new HashMap<>(); + proto.getSingleValueArgumentsList().forEach(argProto -> { + SingleValueArgumentEntry entry = fromSingleValueArgumentProto(argProto); + arguments.computeIfAbsent(argProto.getArgName(), name -> new HashMap<>()).put(entry.getEntityId(), entry); + }); + arguments.forEach((argName, entityInputs) -> { + relatedEntitiesAggState.getArguments().put(argName, new RelatedEntitiesArgumentEntry(entityInputs, false)); + }); + relatedEntitiesAggState.setLastArgsRefreshTs(proto.getLastArgsUpdateTs()); + + return relatedEntitiesAggState; + } + proto.getSingleValueArgumentsList().forEach(argProto -> state.getArguments().put(argProto.getArgName(), fromSingleValueArgumentProto(argProto))); @@ -231,19 +245,6 @@ public class CalculatedFieldUtils { alarmState.setClearRuleState(fromAlarmRuleStateProto(alarmStateProto.getClearRuleState(), alarmState)); } } - case RELATED_ENTITIES_AGGREGATION -> { - RelatedEntitiesAggregationCalculatedFieldState aggState = (RelatedEntitiesAggregationCalculatedFieldState) state; - RelatedEntitiesAggregationStateProto aggregationStateProto = proto.getRelatedEntitiesAggregationState(); - Map> arguments = new HashMap<>(); - aggregationStateProto.getAggArgumentsList().forEach(argProto -> { - SingleValueArgumentEntry entry = fromSingleValueArgumentProto(argProto); - arguments.computeIfAbsent(argProto.getArgName(), name -> new HashMap<>()).put(entry.getEntityId(), entry); - }); - arguments.forEach((argName, entityInputs) -> { - aggState.getArguments().put(argName, new RelatedEntitiesArgumentEntry(entityInputs, false)); - }); - aggState.setLastArgsRefreshTs(aggregationStateProto.getLastArgsUpdateTs()); - } } return state; diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/plugin/ComponentLifecycleEvent.java b/common/data/src/main/java/org/thingsboard/server/common/data/plugin/ComponentLifecycleEvent.java index 5d13db2348..31cab71e0e 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/plugin/ComponentLifecycleEvent.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/plugin/ComponentLifecycleEvent.java @@ -32,7 +32,9 @@ public enum ComponentLifecycleEvent implements Serializable { STOPPED(5), DELETED(6), FAILED(7), - DEACTIVATED(8); + DEACTIVATED(8), + RELATION_UPDATED(9), + RELATION_DELETED(10); @Getter private final int protoNumber; // corresponds to ComponentLifecycleEvent proto diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 2e3a2387e7..bd441dd679 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -916,11 +916,6 @@ message GeofencingArgumentProto { repeated GeofencingZoneProto zones = 2; } -message RelatedEntitiesAggregationStateProto { - int64 lastArgsUpdateTs = 1; - repeated SingleValueArgumentProto aggArguments = 2; -} - message CalculatedFieldStateProto { CalculatedFieldEntityCtxIdProto id = 1; string type = 2; @@ -928,7 +923,7 @@ message CalculatedFieldStateProto { repeated TsRollingArgumentProto rollingValueArguments = 4; repeated GeofencingArgumentProto geofencingArguments = 5; AlarmStateProto alarmState = 6; - RelatedEntitiesAggregationStateProto relatedEntitiesAggregationState = 7; + int64 lastArgsUpdateTs = 7; } //Used to report session state to tb-Service and persist this state in the cache on the tb-Service level. @@ -1281,6 +1276,8 @@ enum ComponentLifecycleEvent { DELETED = 6; FAILED = 7; DEACTIVATED = 8; + RELATION_UPDATED = 9; + RELATION_DELETED = 10; } message ComponentLifecycleMsgProto { diff --git a/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfArg.java b/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfArg.java index 62d6d3d002..2fb12917ff 100644 --- a/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfArg.java +++ b/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfArg.java @@ -29,7 +29,7 @@ import com.fasterxml.jackson.annotation.JsonTypeInfo; @JsonSubTypes.Type(value = TbelCfTsRollingArg.class, name = "TS_ROLLING"), @JsonSubTypes.Type(value = TbelCfGeofencingArg.class, name = "GEOFENCING_CF_ARGUMENT_VALUE"), @JsonSubTypes.Type(value = TbelCfPropagationArg.class, name = "PROPAGATION_CF_ARGUMENT_VALUE"), - @JsonSubTypes.Type(value = TbelCfRelatedEntitiesAggregation.class, name = "LATEST_VALUES_AGGREGATION") + @JsonSubTypes.Type(value = TbelCfRelatedEntitiesAggregation.class, name = "RELATED_ENTITIES_AGGREGATION") }) public interface TbelCfArg extends TbelCfObject { diff --git a/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfRelatedEntitiesAggregation.java b/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfRelatedEntitiesAggregation.java index 75c17f0a0e..3373aa2474 100644 --- a/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfRelatedEntitiesAggregation.java +++ b/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfRelatedEntitiesAggregation.java @@ -34,7 +34,7 @@ public class TbelCfRelatedEntitiesAggregation implements TbelCfArg { @Override public String getType() { - return "LATEST_VALUES_AGGREGATION"; + return "RELATED_ENTITIES_AGGREGATION"; } @Override