Added propagation entities to proto state && added reset of readiness status on relation updates

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

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

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

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<EntityIdProto> 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<EntityId> 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();

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();

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.