From 4137a4a615ef4f8cfc84bb82878661feba9db4dd Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Wed, 15 Oct 2025 10:05:48 +0300 Subject: [PATCH] refactoring --- ...CalculatedFieldEntityMessageProcessor.java | 105 ++++++++----- ...alculatedFieldManagerMessageProcessor.java | 139 ++++++------------ .../CalculatedFieldRelatedEntityMsg.java | 9 +- 3 files changed, 117 insertions(+), 136 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 e253633727..63aaf2bd68 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 @@ -171,7 +171,8 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM throw cfe; } throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build(); - } } + } + } public void process(CalculatedFieldArgumentResetMsg msg) throws CalculatedFieldException { log.debug("[{}] Processing CF argument reset msg.", entityId); @@ -202,63 +203,91 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM actorCtx.stop(actorCtx.getSelf()); } } else { - EntityId msgEntityId = msg.getEntityId(); - if (msgEntityId instanceof CalculatedFieldId cfId) { - var state = removeState(cfId); - if (state != null) { - cfStateService.deleteState(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), msg.getCallback()); - } else { - msg.getCallback().onSuccess(); - } + var cfId = new CalculatedFieldId(msg.getEntityId().getId()); + var state = removeState(cfId); + if (state != null) { + cfStateService.deleteState(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), msg.getCallback()); } else { - if (states.isEmpty()) { - msg.getCallback().onSuccess(); - } - for (Map.Entry entry : states.entrySet()) { - LatestValuesAggregationCalculatedFieldState state = (LatestValuesAggregationCalculatedFieldState) entry.getValue(); - state.getArguments().forEach((argName, argEntry) -> { - AggArgumentEntry aggArgEntry = (AggArgumentEntry) argEntry; - aggArgEntry.getAggInputs().remove(msgEntityId); - }); - state.getInputs().remove(msgEntityId); - state.setLastMetricsEvalTs(-1); - processStateIfReady(state, Collections.emptyMap(), state.getCtx(), Collections.emptyList(), null, null, msg.getCallback()); - } + msg.getCallback().onSuccess(); } } } public void process(CalculatedFieldRelatedEntityMsg msg) throws CalculatedFieldException { - log.debug("[{}] Processing CF related entity msg.", msg.getEntityId()); - CalculatedFieldCtx cfCtx = msg.getCalculatedField(); - var state = states.get(cfCtx.getCfId()); - Map fetchedArguments = fetchAggArguments(msg.getCalculatedField(), msg.getEntityId()); + log.debug("[{}] Processing CF {} related entity msg.", msg.getRelatedEntityId(), msg.getAction()); + switch (msg.getAction()) { + case UPDATED -> handleRelationUpdate(msg); + case DELETED -> handleRelationDelete(msg); + default -> msg.getCallback().onSuccess(); + } + } + + private void handleRelationUpdate(CalculatedFieldRelatedEntityMsg msg) throws CalculatedFieldException { + CalculatedFieldCtx ctx = msg.getCalculatedField(); + var callback = new MultipleTbCallback(CALLBACKS_PER_CF, msg.getCallback()); + var state = states.get(ctx.getCfId()); try { + boolean justRestored = false; if (state == null) { - state = createState(cfCtx); - } else { - state.setCtx(cfCtx, actorCtx); + state = createState(ctx); + justRestored = true; } if (state.isSizeOk()) { - if (state instanceof LatestValuesAggregationCalculatedFieldState latestValuesState) { - latestValuesState.setLastMetricsEvalTs(-1); + Map updatedArgs = new HashMap<>(); + if (!justRestored) { + updatedArgs = updateAggregationState(msg.getRelatedEntityId(), state, ctx); } - state.update(fetchedArguments, cfCtx); - state.checkStateSize(new CalculatedFieldEntityCtxId(tenantId, cfCtx.getCfId(), entityId), cfCtx.getMaxStateSize()); - states.put(cfCtx.getCfId(), state); - processStateIfReady(state, fetchedArguments, cfCtx, Collections.singletonList(cfCtx.getCfId()), null, null, msg.getCallback()); + processStateIfReady(state, updatedArgs, ctx, new ArrayList<>(), null, null, callback); } else { - throw new RuntimeException(cfCtx.getSizeExceedsLimitMessage()); + throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).errorMessage(ctx.getSizeExceedsLimitMessage()).build(); } } catch (Exception e) { - log.debug("[{}][{}] Failed to initialize CF state", entityId, cfCtx.getCfId(), e); + log.debug("[{}][{}] Failed to initialize CF state", entityId, ctx.getCfId(), e); if (e instanceof CalculatedFieldException cfe) { throw cfe; } - throw CalculatedFieldException.builder().ctx(cfCtx).eventEntity(entityId).cause(e).build(); + throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build(); + } + } + + private Map updateAggregationState(EntityId relatedEntityId, CalculatedFieldState state, CalculatedFieldCtx ctx) { + Map fetchedArgs = fetchAggArguments(ctx, relatedEntityId); + Map updatedArgs = state.update(fetchedArgs, ctx); + + if (state instanceof LatestValuesAggregationCalculatedFieldState latestValuesState) { + latestValuesState.setLastMetricsEvalTs(-1); + } + + state.checkStateSize(new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId), ctx.getMaxStateSize()); + + return updatedArgs; + } + + private void handleRelationDelete(CalculatedFieldRelatedEntityMsg msg) throws CalculatedFieldException { + CalculatedFieldCtx ctx = msg.getCalculatedField(); + CalculatedFieldId cfId = ctx.getCfId(); + CalculatedFieldState state = states.get(cfId); + if (state == null) { + msg.getCallback().onSuccess(); + return; + } + if (state instanceof LatestValuesAggregationCalculatedFieldState aggState) { + cleanupAggregationState(msg.getRelatedEntityId(), aggState); + processStateIfReady(state, Collections.emptyMap(), state.getCtx(), Collections.emptyList(), null, null, msg.getCallback()); + } else { + msg.getCallback().onSuccess(); } } + private void cleanupAggregationState(EntityId relatedEntityId, LatestValuesAggregationCalculatedFieldState state) { + state.getArguments().values().forEach(argEntry -> { + AggArgumentEntry aggEntry = (AggArgumentEntry) argEntry; + aggEntry.getAggInputs().remove(relatedEntityId); + }); + state.getInputs().remove(relatedEntityId); + state.setLastMetricsEvalTs(-1); + } + @SneakyThrows private Map fetchAggArguments(CalculatedFieldCtx ctx, EntityId entityId) { ListenableFuture> argumentsFuture = cfService.fetchAggEntityArguments(ctx, entityId); 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 620eee41ee..37ec4cc3d1 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 @@ -16,6 +16,7 @@ package org.thingsboard.server.actors.calculatedField; import lombok.extern.slf4j.Slf4j; +import org.apache.logging.log4j.util.TriConsumer; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.actors.TbActorCtx; @@ -29,6 +30,7 @@ import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.alarm.Alarm; +import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.cf.CalculatedField; import org.thingsboard.server.common.data.cf.CalculatedFieldLink; import org.thingsboard.server.common.data.cf.configuration.aggregation.LatestValuesAggregationCalculatedFieldConfiguration; @@ -303,56 +305,14 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private void onRelationUpdated(ComponentLifecycleMsg msg, TbCallback callback) { try { - MultipleTbCallback callbackForToAndFrom = new MultipleTbCallback(2, callback); EntityRelation entityRelation = JacksonUtil.treeToValue(msg.getInfo(), EntityRelation.class); - EntityId toId = entityRelation.getTo(); EntityId fromId = entityRelation.getFrom(); String relationType = entityRelation.getType(); - EntityId toIdProfile = getProfileId(tenantId, toId); - EntityId fromIdProfile = getProfileId(tenantId, fromId); - - List toIdMatches = new ArrayList<>(); - List cfsByToId = getCalculatedFieldsByEntityId(toId); - List cfsByToProfileId = getCalculatedFieldsByEntityId(toIdProfile); - List cfsByToIdOrItsProfileId = new ArrayList<>(); - cfsByToIdOrItsProfileId.addAll(cfsByToId); - cfsByToIdOrItsProfileId.addAll(cfsByToProfileId); - - cfsByToIdOrItsProfileId.forEach(cf -> { - var configuration = (LatestValuesAggregationCalculatedFieldConfiguration) cf.getCalculatedField().getConfiguration(); - RelationPathLevel relation = configuration.getRelation(); - if (EntitySearchDirection.TO.equals(relation.direction()) && relationType.equals(relation.relationType())) { - toIdMatches.add(cf); - } - }); - - MultipleTbCallback toCfsCallback = new MultipleTbCallback(toIdMatches.size(), callbackForToAndFrom); - toIdMatches.forEach(ctx -> { - applyToTargetCfEntityActors(ctx, toCfsCallback, (entityId, cb) -> initRelatedEntity(entityId, fromId, ctx, cb)); - }); - - List fromIdMatches = new ArrayList<>(); - List cfsByFromId = getCalculatedFieldsByEntityId(fromId); - List cfsByFromProfileId = getCalculatedFieldsByEntityId(fromIdProfile); - List cfsByFromIdOrItsProfileId = new ArrayList<>(); - cfsByFromIdOrItsProfileId.addAll(cfsByFromId); - cfsByFromIdOrItsProfileId.addAll(cfsByFromProfileId); - - cfsByFromIdOrItsProfileId.forEach(cf -> { - var configuration = (LatestValuesAggregationCalculatedFieldConfiguration) cf.getCalculatedField().getConfiguration(); - RelationPathLevel relation = configuration.getRelation(); - if (EntitySearchDirection.FROM.equals(relation.direction()) && relationType.equals(relation.relationType())) { - fromIdMatches.add(cf); - } - }); - - MultipleTbCallback fromCfsCallback = new MultipleTbCallback(fromIdMatches.size(), callbackForToAndFrom); - fromIdMatches.forEach(ctx -> { - applyToTargetCfEntityActors(ctx, fromCfsCallback, (entityId, cb) -> initRelatedEntity(entityId, toId, ctx, cb)); - }); - + MultipleTbCallback callbackForToAndFrom = new MultipleTbCallback(2, callback); + processRelationByDirection(EntitySearchDirection.TO, relationType, toId, callbackForToAndFrom, (entityId, ctx, cb) -> initRelatedEntity(entityId, fromId, ctx, cb)); + processRelationByDirection(EntitySearchDirection.FROM, relationType, fromId, callbackForToAndFrom, (entityId, ctx, cb) -> initRelatedEntity(entityId, toId, ctx, cb)); } catch (Exception e) { callback.onSuccess(); } @@ -360,59 +320,43 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private void onRelationDeleted(ComponentLifecycleMsg msg, TbCallback callback) { try { - MultipleTbCallback callbackForToAndFrom = new MultipleTbCallback(2, callback); EntityRelation entityRelation = JacksonUtil.treeToValue(msg.getInfo(), EntityRelation.class); - EntityId toId = entityRelation.getTo(); EntityId fromId = entityRelation.getFrom(); String relationType = entityRelation.getType(); - EntityId toIdProfile = getProfileId(tenantId, toId); - EntityId fromIdProfile = getProfileId(tenantId, fromId); - - List toIdMatches = new ArrayList<>(); - List cfsByToId = getCalculatedFieldsByEntityId(toId); - List cfsByToProfileId = getCalculatedFieldsByEntityId(toIdProfile); - List cfsByToIdOrItsProfileId = new ArrayList<>(); - cfsByToIdOrItsProfileId.addAll(cfsByToId); - cfsByToIdOrItsProfileId.addAll(cfsByToProfileId); - - cfsByToIdOrItsProfileId.forEach(cf -> { - var configuration = (LatestValuesAggregationCalculatedFieldConfiguration) cf.getCalculatedField().getConfiguration(); - RelationPathLevel relation = configuration.getRelation(); - if (EntitySearchDirection.TO.equals(relation.direction()) && relationType.equals(relation.relationType())) { - toIdMatches.add(cf); - } - }); - MultipleTbCallback toCfsCallback = new MultipleTbCallback(toIdMatches.size(), callbackForToAndFrom); - toIdMatches.forEach(ctx -> { - applyToTargetCfEntityActors(ctx, toCfsCallback, (entityId, cb) -> deleteRelatedEntity(entityId, fromId, cb)); - }); + MultipleTbCallback callbackForToAndFrom = new MultipleTbCallback(2, callback); + processRelationByDirection(EntitySearchDirection.TO, relationType, toId, callbackForToAndFrom, (entityId, ctx, cb) -> deleteRelatedEntity(entityId, fromId, ctx, cb)); + processRelationByDirection(EntitySearchDirection.FROM, relationType, fromId, callbackForToAndFrom, (entityId, ctx, cb) -> deleteRelatedEntity(entityId, toId, ctx, cb)); + } catch (Exception e) { + callback.onSuccess(); + } + } - List fromIdMatches = new ArrayList<>(); - List cfsByFromId = getCalculatedFieldsByEntityId(fromId); - List cfsByFromProfileId = getCalculatedFieldsByEntityId(fromIdProfile); - List cfsByFromIdOrItsProfileId = new ArrayList<>(); - cfsByFromIdOrItsProfileId.addAll(cfsByFromId); - cfsByFromIdOrItsProfileId.addAll(cfsByFromProfileId); - - cfsByFromIdOrItsProfileId.forEach(cf -> { - var configuration = (LatestValuesAggregationCalculatedFieldConfiguration) cf.getCalculatedField().getConfiguration(); - RelationPathLevel relation = configuration.getRelation(); - if (EntitySearchDirection.FROM.equals(relation.direction()) && relationType.equals(relation.relationType())) { - fromIdMatches.add(cf); - } - }); + private void processRelationByDirection(EntitySearchDirection direction, + String relationType, + EntityId mainId, + MultipleTbCallback parentCallback, + TriConsumer relationAction) { + List cfsByEntityIdAndProfile = getCalculatedFieldsByEntityIdAndProfile(mainId); + if (cfsByEntityIdAndProfile.isEmpty()) { + parentCallback.onSuccess(); + return; + } - MultipleTbCallback fromCfsCallback = new MultipleTbCallback(fromIdMatches.size(), callbackForToAndFrom); - fromIdMatches.forEach(ctx -> { - applyToTargetCfEntityActors(ctx, fromCfsCallback, (entityId, cb) -> deleteRelatedEntity(entityId, toId, cb)); - }); + List matchingCfs = cfsByEntityIdAndProfile.stream() + .filter(cf -> { + var config = (LatestValuesAggregationCalculatedFieldConfiguration) cf.getCalculatedField().getConfiguration(); + RelationPathLevel relation = config.getRelation(); + return direction.equals(relation.direction()) && relationType.equals(relation.relationType()); + }) + .toList(); + MultipleTbCallback directionCallback = new MultipleTbCallback(matchingCfs.size(), parentCallback); - } catch (Exception e) { - callback.onSuccess(); - } + matchingCfs.forEach(ctx -> + applyToTargetCfEntityActors(ctx, directionCallback, (entityId, cb) -> relationAction.accept(entityId, ctx, cb)) + ); } private void onCfCreated(ComponentLifecycleMsg msg, TbCallback callback) throws CalculatedFieldException { @@ -629,9 +573,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware EntityId entityId = msg.getEntityId(); log.debug("Received changed owner msg from entity [{}]", entityId); updateEntityOwner(entityId); - List cfs = new ArrayList<>(); - cfs.addAll(getCalculatedFieldsByEntityId(entityId)); - cfs.addAll(getCalculatedFieldsByEntityId(getProfileId(tenantId, entityId))); + List cfs = getCalculatedFieldsByEntityIdAndProfile(entityId); if (cfs.isEmpty()) { msgCallback.onSuccess(); return; @@ -695,6 +637,13 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware return result; } + private List getCalculatedFieldsByEntityIdAndProfile(EntityId entityId) { + List cfsByEntityIdAndProfile = new ArrayList<>(); + cfsByEntityIdAndProfile.addAll(getCalculatedFieldsByEntityId(entityId)); + cfsByEntityIdAndProfile.addAll(getCalculatedFieldsByEntityId(getProfileId(tenantId, entityId))); + return cfsByEntityIdAndProfile; + } + private List getCalculatedFieldLinksByEntityId(EntityId entityId) { if (entityId == null) { return Collections.emptyList(); @@ -722,14 +671,14 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware getOrCreateActor(entityId).tell(msg); } - private void deleteRelatedEntity(EntityId entityId, EntityId relatedEntityId, TbCallback callback) { + private void deleteRelatedEntity(EntityId entityId, EntityId relatedEntityId, CalculatedFieldCtx cf, TbCallback callback) { log.debug("Pushing delete related entity msg to specific actor [{}]", relatedEntityId); - getOrCreateActor(entityId).tell(new CalculatedFieldEntityDeleteMsg(tenantId, relatedEntityId, callback)); + getOrCreateActor(entityId).tell(new CalculatedFieldRelatedEntityMsg(tenantId, relatedEntityId, ActionType.DELETED, cf, callback)); } private void initRelatedEntity(EntityId entityId, EntityId relatedEntityId, CalculatedFieldCtx cf, TbCallback callback) { log.debug("Pushing init related entity msg to specific actor [{}]", relatedEntityId); - getOrCreateActor(entityId).tell(new CalculatedFieldRelatedEntityMsg(tenantId, relatedEntityId, cf, callback)); + getOrCreateActor(entityId).tell(new CalculatedFieldRelatedEntityMsg(tenantId, relatedEntityId, ActionType.UPDATED, cf, callback)); } private void deleteCfForEntity(EntityId entityId, CalculatedFieldId cfId, TbCallback callback) { diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldRelatedEntityMsg.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldRelatedEntityMsg.java index d3bdd9cc5c..bf73a1b1b0 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldRelatedEntityMsg.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldRelatedEntityMsg.java @@ -16,6 +16,7 @@ package org.thingsboard.server.actors.calculatedField; import lombok.Data; +import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.MsgType; @@ -27,16 +28,18 @@ import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; public class CalculatedFieldRelatedEntityMsg implements ToCalculatedFieldSystemMsg { private final TenantId tenantId; - private final EntityId entityId; + private final EntityId relatedEntityId; + private final ActionType action; private final CalculatedFieldCtx calculatedField; private final TbCallback callback; public CalculatedFieldRelatedEntityMsg(TenantId tenantId, - EntityId entityId, + EntityId relatedEntityId, ActionType action, CalculatedFieldCtx calculatedField, TbCallback callback) { this.tenantId = tenantId; - this.entityId = entityId; + this.relatedEntityId = relatedEntityId; + this.action = action; this.calculatedField = calculatedField; this.callback = callback; }