From 8bbfcca578d9e695d9907d775ec7152204747470 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Thu, 4 Dec 2025 18:19:26 +0200 Subject: [PATCH] Added propagation entities to proto state && added reset of readiness status on relation updates --- .../CalculatedFieldEntityMessageProcessor.java | 6 +++++- .../propagation/PropagationArgumentEntry.java | 4 ++-- .../PropagationCalculatedFieldState.java | 4 ++++ .../server/utils/CalculatedFieldUtils.java | 14 ++++++++++++++ .../server/utils/CalculatedFieldUtilsTest.java | 4 ++-- common/proto/src/main/proto/queue.proto | 1 + 6 files changed, 28 insertions(+), 5 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java index d3fcb8e2d8..9d8231956c 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java @@ -241,6 +241,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM PropagationArgumentEntry propagationArgument = propagationState.getPropagationArgument(); boolean added = propagationArgument.addPropagationEntityId(msg.getRelatedEntityId()); if (added) { + propagationState.resetReadinessStatus(); updatedArgs = Map.of(PROPAGATION_CONFIG_ARGUMENT, new PropagationArgumentEntry(List.of(msg.getRelatedEntityId()))); } } @@ -286,7 +287,10 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } if (state instanceof PropagationCalculatedFieldState propagationState) { PropagationArgumentEntry propagationArgument = propagationState.getPropagationArgument(); - propagationArgument.removePropagationEntityId(msg.getRelatedEntityId()); + boolean removed = propagationArgument.removePropagationEntityId(msg.getRelatedEntityId()); + if (removed) { + propagationState.resetReadinessStatus(); + } } msg.getCallback().onSuccess(); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationArgumentEntry.java index ebb23a0794..964aa5487e 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationArgumentEntry.java @@ -77,8 +77,8 @@ public class PropagationArgumentEntry implements ArgumentEntry { return propagationEntityIds.add(propagationEntityId); } - public void removePropagationEntityId(EntityId relatedEntityId) { - propagationEntityIds.remove(relatedEntityId); + public boolean removePropagationEntityId(EntityId relatedEntityId) { + return propagationEntityIds.remove(relatedEntityId); } } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationCalculatedFieldState.java index 991d3e66dd..a6abeb1b32 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationCalculatedFieldState.java @@ -117,4 +117,8 @@ public class PropagationCalculatedFieldState extends ScriptCalculatedFieldState return (PropagationArgumentEntry) arguments.get(PROPAGATION_CONFIG_ARGUMENT); } + public void resetReadinessStatus() { + readinessStatus = checkReadiness(requiredArguments, arguments); + } + } 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 09ba7917e9..5fd48398e4 100644 --- a/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java +++ b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java @@ -33,6 +33,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ArgumentIntervalProt import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldIdProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; +import org.thingsboard.server.gen.transport.TransportProtos.EntityIdProto; import org.thingsboard.server.gen.transport.TransportProtos.GeofencingArgumentProto; import org.thingsboard.server.gen.transport.TransportProtos.GeofencingZoneProto; import org.thingsboard.server.gen.transport.TransportProtos.SingleValueArgumentProto; @@ -57,9 +58,11 @@ import org.thingsboard.server.service.cf.ctx.state.alarm.AlarmRuleState; import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingZoneState; +import org.thingsboard.server.service.cf.ctx.state.propagation.PropagationArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.propagation.PropagationCalculatedFieldState; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Optional; import java.util.TreeMap; @@ -67,6 +70,8 @@ import java.util.UUID; import java.util.function.Function; import java.util.stream.Collectors; +import static org.thingsboard.server.common.data.cf.configuration.PropagationCalculatedFieldConfiguration.PROPAGATION_CONFIG_ARGUMENT; + public class CalculatedFieldUtils { public static CalculatedFieldIdProto toProto(CalculatedFieldId cfId) { @@ -105,6 +110,7 @@ public class CalculatedFieldUtils { 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 PROPAGATION -> builder.addAllPropagationEntityIds(toPropagationEntityIdsProto((PropagationArgumentEntry) argEntry)); case RELATED_ENTITIES -> { RelatedEntitiesArgumentEntry relatedEntitiesArgumentEntry = (RelatedEntitiesArgumentEntry) argEntry; relatedEntitiesArgumentEntry.getEntityInputs() @@ -133,6 +139,10 @@ public class CalculatedFieldUtils { return builder.build(); } + private static List toPropagationEntityIdsProto(PropagationArgumentEntry argEntry) { + return argEntry.getPropagationEntityIds().stream().map(ProtoUtils::toProto).collect(Collectors.toList()); + } + private static AlarmRuleStateProto toAlarmRuleStateProto(AlarmRuleState ruleState) { return AlarmRuleStateProto.newBuilder() .setSeverity(Optional.ofNullable(ruleState.getSeverity()).map(Enum::name).orElse("")) @@ -269,6 +279,10 @@ public class CalculatedFieldUtils { proto.getGeofencingArgumentsList().forEach(argProto -> state.getArguments().put(argProto.getArgName(), fromGeofencingArgumentProto(argProto))); } + case PROPAGATION -> { + List propagationEntityIds = proto.getPropagationEntityIdsList().stream().map(ProtoUtils::fromProto).toList(); + state.getArguments().put(PROPAGATION_CONFIG_ARGUMENT, new PropagationArgumentEntry(propagationEntityIds)); + } case ALARM -> { AlarmCalculatedFieldState alarmState = (AlarmCalculatedFieldState) state; AlarmStateProto alarmStateProto = proto.getAlarmState(); diff --git a/application/src/test/java/org/thingsboard/server/utils/CalculatedFieldUtilsTest.java b/application/src/test/java/org/thingsboard/server/utils/CalculatedFieldUtilsTest.java index db24e51123..9573c9fa35 100644 --- a/application/src/test/java/org/thingsboard/server/utils/CalculatedFieldUtilsTest.java +++ b/application/src/test/java/org/thingsboard/server/utils/CalculatedFieldUtilsTest.java @@ -119,7 +119,7 @@ class CalculatedFieldUtilsTest { } @Test - void toProtoAndFromProto_shouldCreatePropagationStateWithoutPropagationArgument() { + void toProtoAndFromProto_shouldCreatePropagationStateWithPropagationArgument() { // given CalculatedFieldEntityCtxId stateId = mock(CalculatedFieldEntityCtxId.class); given(stateId.tenantId()).willReturn(TENANT_ID); @@ -158,7 +158,7 @@ class CalculatedFieldUtilsTest { assertThat(propagationState.getEntityId()).isEqualTo(DEVICE_ID); assertThat(propagationState.getArguments()).isNotNull(); - assertThat(propagationState.getArguments().get(PROPAGATION_CONFIG_ARGUMENT)).isNull(); + assertThat(propagationState.getArguments().get(PROPAGATION_CONFIG_ARGUMENT)).isEqualTo(propagationArgumentEntry); assertThat(propagationState.getArguments().get("state")).isNotNull().isEqualTo(singleValueArgumentEntry); assertThat(propagationState.getRequiredArguments()).isNull(); assertThat(propagationState.getReadinessStatus()).isNull(); diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 9fb8528bce..3cbc84ba1a 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -935,6 +935,7 @@ message CalculatedFieldStateProto { int64 lastArgsUpdateTs = 7; int64 lastMetricsEvalTs = 8; repeated ArgumentIntervalProto aggregationArguments = 9; + repeated EntityIdProto propagationEntityIds = 10; } //Used to report session state to tb-Service and persist this state in the cache on the tb-Service level.