From 06515a5e13cfc34859534f14e41dc48565d020a4 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Tue, 12 Aug 2025 16:23:38 +0300 Subject: [PATCH] Renamed actor message types and java fields for clarity --- ...latedFieldDynamicArgumentsRefreshMsg.java} | 4 +- .../CalculatedFieldEntityActor.java | 4 +- ...CalculatedFieldEntityMessageProcessor.java | 4 +- .../CalculatedFieldManagerActor.java | 4 +- ...alculatedFieldManagerMessageProcessor.java | 56 +++++++++---------- ...latedFieldDynamicArgumentsRefreshMsg.java} | 4 +- .../server/common/msg/MsgType.java | 4 +- 7 files changed, 38 insertions(+), 42 deletions(-) rename application/src/main/java/org/thingsboard/server/actors/calculatedField/{CalculatedFieldScheduledInvalidationMsg.java => CalculatedFieldDynamicArgumentsRefreshMsg.java} (87%) rename application/src/main/java/org/thingsboard/server/actors/calculatedField/{EntityCalculatedFieldMarkStateDirtyMsg.java => EntityCalculatedFieldDynamicArgumentsRefreshMsg.java} (87%) diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldScheduledInvalidationMsg.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldDynamicArgumentsRefreshMsg.java similarity index 87% rename from application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldScheduledInvalidationMsg.java rename to application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldDynamicArgumentsRefreshMsg.java index 08a2394119..301fe22dfb 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldScheduledInvalidationMsg.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldDynamicArgumentsRefreshMsg.java @@ -22,14 +22,14 @@ import org.thingsboard.server.common.msg.MsgType; import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg; @Data -public class CalculatedFieldScheduledInvalidationMsg implements ToCalculatedFieldSystemMsg { +public class CalculatedFieldDynamicArgumentsRefreshMsg implements ToCalculatedFieldSystemMsg { private final TenantId tenantId; private final CalculatedFieldId cfId; @Override public MsgType getMsgType() { - return MsgType.CF_SCHEDULED_INVALIDATION_MSG; + return MsgType.CF_DYNAMIC_ARGUMENTS_REFRESH_MSG; } } diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java index 4879fa4566..2a5f3c3cfd 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java @@ -75,8 +75,8 @@ public class CalculatedFieldEntityActor extends AbstractCalculatedFieldActor { case CF_LINKED_TELEMETRY_MSG: processor.process((EntityCalculatedFieldLinkedTelemetryMsg) msg); break; - case CF_ENTITY_MARK_STATE_DIRTY_MSG: - processor.process((EntityCalculatedFieldMarkStateDirtyMsg) msg); + case CF_ENTITY_DYNAMIC_ARGUMENTS_REFRESH_MSG: + processor.process((EntityCalculatedFieldDynamicArgumentsRefreshMsg) msg); break; default: return false; 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 74474e8a6d..4b277eb3a3 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 @@ -226,8 +226,8 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } } - public void process(EntityCalculatedFieldMarkStateDirtyMsg msg) throws CalculatedFieldException { - log.debug("[{}][{}] Processing entity CF invalidation msg.", entityId, msg.getCfId()); + public void process(EntityCalculatedFieldDynamicArgumentsRefreshMsg msg) throws CalculatedFieldException { + log.debug("[{}][{}] Processing CF dynamic arguments refresh msg.", entityId, msg.getCfId()); CalculatedFieldState currentState = states.get(msg.getCfId()); if (currentState == null) { log.debug("[{}][{}] Failed to find CF state for entity.", entityId, msg.getCfId()); diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerActor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerActor.java index 8494fb3847..c752333f69 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerActor.java @@ -91,8 +91,8 @@ public class CalculatedFieldManagerActor extends AbstractCalculatedFieldActor { case CF_LINKED_TELEMETRY_MSG: processor.onLinkedTelemetryMsg((CalculatedFieldLinkedTelemetryMsg) msg); break; - case CF_SCHEDULED_INVALIDATION_MSG: - processor.onScheduledInvalidationMsg((CalculatedFieldScheduledInvalidationMsg) msg); + case CF_DYNAMIC_ARGUMENTS_REFRESH_MSG: + processor.onDynamicArgumentsRefreshMsg((CalculatedFieldDynamicArgumentsRefreshMsg) msg); break; default: return false; 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 76acdc329b..8a8fc43503 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 @@ -76,7 +76,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private final Map calculatedFields = new HashMap<>(); private final Map> entityIdCalculatedFields = new HashMap<>(); private final Map> entityIdCalculatedFieldLinks = new HashMap<>(); - private final Map> cfInvalidationScheduledTasks = new ConcurrentHashMap<>(); + private final Map> cfDynamicArgumentsRefreshTasks = new ConcurrentHashMap<>(); private final CalculatedFieldProcessingService cfExecService; private final CalculatedFieldStateService cfStateService; @@ -115,8 +115,8 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware calculatedFields.clear(); entityIdCalculatedFields.clear(); entityIdCalculatedFieldLinks.clear(); - cfInvalidationScheduledTasks.values().forEach(future -> future.cancel(true)); - cfInvalidationScheduledTasks.clear(); + cfDynamicArgumentsRefreshTasks.values().forEach(future -> future.cancel(true)); + cfDynamicArgumentsRefreshTasks.clear(); ctx.stop(ctx.getSelf()); } @@ -146,7 +146,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware // We use copy on write lists to safely pass the reference to another actor for the iteration. // Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead) entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(cfCtx); - scheduleCalculatedFieldInvalidationMsgIfNeeded(cfCtx); + scheduleDynamicArgumentsRefreshTaskForCfIfNeeded(cfCtx); msg.getCallback().onSuccess(); } @@ -339,7 +339,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware boolean hasSchedulingConfigChanges = newCfCtx.hasSchedulingConfigChanges(oldCfCtx); if (hasSchedulingConfigChanges) { - cancelCfScheduledInvalidationTaskIfExists(cfId, false); + cancelCfDynamicArgumentsRefreshTaskIfExists(cfId, false); } List newCfList = new CopyOnWriteArrayList<>(); @@ -382,7 +382,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware entityIdCalculatedFields.get(cfCtx.getEntityId()).remove(cfCtx); deleteLinks(cfCtx); - cancelCfScheduledInvalidationTaskIfExists(cfId, true); + cancelCfDynamicArgumentsRefreshTaskIfExists(cfId, true); EntityId entityId = cfCtx.getEntityId(); EntityType entityType = cfCtx.getEntityId().getEntityType(); @@ -407,12 +407,12 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } } - private void cancelCfScheduledInvalidationTaskIfExists(CalculatedFieldId cfId, boolean cfDeleted) { - var existingTask = cfInvalidationScheduledTasks.remove(cfId); + private void cancelCfDynamicArgumentsRefreshTaskIfExists(CalculatedFieldId cfId, boolean cfDeleted) { + var existingTask = cfDynamicArgumentsRefreshTasks.remove(cfId); if (existingTask != null) { existingTask.cancel(false); String reason = cfDeleted ? "deletion" : "update"; - log.debug("[{}][{}] Cancelled scheduled invalidation task due to CF " + reason + "!", tenantId, cfId); + log.debug("[{}][{}] Cancelled dynamic arguments refresh task due to CF " + reason + "!", tenantId, cfId); } } @@ -455,7 +455,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware var targetEntityId = link.entityId(); var targetEntityType = targetEntityId.getEntityType(); var cf = calculatedFields.get(link.cfId()); - if (EntityType.DEVICE_PROFILE.equals(targetEntityType) || EntityType.ASSET_PROFILE.equals(targetEntityType)) { + if (isProfileEntity(targetEntityType)) { // iterate over all entities that belong to profile and push the message for corresponding CF var entityIds = entityProfileCache.getEntityIdsByProfileId(targetEntityId); if (!entityIds.isEmpty()) { @@ -518,7 +518,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private void initCf(CalculatedFieldCtx cfCtx, TbCallback callback, boolean forceStateReinit) { EntityId entityId = cfCtx.getEntityId(); EntityType entityType = cfCtx.getEntityId().getEntityType(); - scheduleCalculatedFieldInvalidationMsgIfNeeded(cfCtx); + scheduleDynamicArgumentsRefreshTaskForCfIfNeeded(cfCtx); if (isProfileEntity(entityType)) { var entityIds = entityProfileCache.getEntityIdsByProfileId(entityId); if (!entityIds.isEmpty()) { @@ -536,31 +536,31 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } } - private void scheduleCalculatedFieldInvalidationMsgIfNeeded(CalculatedFieldCtx cfCtx) { + private void scheduleDynamicArgumentsRefreshTaskForCfIfNeeded(CalculatedFieldCtx cfCtx) { CalculatedField cf = cfCtx.getCalculatedField(); CalculatedFieldConfiguration cfConfig = cf.getConfiguration(); if (!cfConfig.isScheduledUpdateEnabled()) { return; } - if (cfInvalidationScheduledTasks.containsKey(cf.getId())) { - log.debug("[{}][{}] Scheduled invalidation task for CF already exists!", tenantId, cf.getId()); + if (cfDynamicArgumentsRefreshTasks.containsKey(cf.getId())) { + log.debug("[{}][{}] Dynamic arguments refresh task for CF already exists!", tenantId, cf.getId()); return; } long refreshDynamicSourceInterval = TimeUnit.SECONDS.toMillis(cfConfig.getScheduledUpdateIntervalSec()); - var scheduledMsg = new CalculatedFieldScheduledInvalidationMsg(tenantId, cfCtx.getCfId()); + var scheduledMsg = new CalculatedFieldDynamicArgumentsRefreshMsg(tenantId, cfCtx.getCfId()); ScheduledFuture scheduledFuture = systemContext .schedulePeriodicMsgWithDelay(ctx, scheduledMsg, refreshDynamicSourceInterval, refreshDynamicSourceInterval); - cfInvalidationScheduledTasks.put(cf.getId(), scheduledFuture); - log.debug("[{}][{}] Scheduled invalidation task for CF!", tenantId, cf.getId()); + cfDynamicArgumentsRefreshTasks.put(cf.getId(), scheduledFuture); + log.debug("[{}][{}] Scheduled dynamic arguments refresh task for CF!", tenantId, cf.getId()); } - public void onScheduledInvalidationMsg(CalculatedFieldScheduledInvalidationMsg msg) { - log.debug("[{}] [{}] Processing CF scheduled invalidation msg.", tenantId, msg.getCfId()); + public void onDynamicArgumentsRefreshMsg(CalculatedFieldDynamicArgumentsRefreshMsg msg) { + log.debug("[{}] [{}] Processing CF dynamic arguments refresh task.", tenantId, msg.getCfId()); CalculatedFieldCtx cfCtx = calculatedFields.get(msg.getCfId()); if (cfCtx == null) { - log.debug("[{}][{}] Failed to find CF context, going to stop scheduled invalidations for CF.", tenantId, msg.getCfId()); - cancelCfScheduledInvalidationTaskIfExists(msg.getCfId(), true); + log.debug("[{}][{}] Failed to find CF context, going to stop dynamic arguments refresh task for CF.", tenantId, msg.getCfId()); + cancelCfDynamicArgumentsRefreshTaskIfExists(msg.getCfId(), true); return; } EntityId entityId = cfCtx.getEntityId(); @@ -571,7 +571,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware var multiCallback = new MultipleTbCallback(entityIds.size(), msg.getCallback()); entityIds.forEach(id -> { if (isMyPartition(id, multiCallback)) { - InitCfInvalidationForEntity(id, msg.getCfId(), multiCallback); + dynamicArgumentsRefreshForEntity(id, msg.getCfId(), multiCallback); } }); } else { @@ -579,14 +579,14 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } } else { if (isMyPartition(entityId, msg.getCallback())) { - InitCfInvalidationForEntity(entityId, msg.getCfId(), msg.getCallback()); + dynamicArgumentsRefreshForEntity(entityId, msg.getCfId(), msg.getCallback()); } } } - private void InitCfInvalidationForEntity(EntityId entityId, CalculatedFieldId cfId, TbCallback callback) { - log.debug("Pushing entity CF invalidation msg to specific actor [{}]", entityId); - getOrCreateActor(entityId).tell(new EntityCalculatedFieldMarkStateDirtyMsg(tenantId, cfId, callback)); + private void dynamicArgumentsRefreshForEntity(EntityId entityId, CalculatedFieldId cfId, TbCallback callback) { + log.debug("Pushing CF dynamic arguments refresh msg to specific actor [{}]", entityId); + getOrCreateActor(entityId).tell(new EntityCalculatedFieldDynamicArgumentsRefreshMsg(tenantId, cfId, callback)); } private void deleteCfForEntity(EntityId entityId, CalculatedFieldId cfId, TbCallback callback) { @@ -652,10 +652,6 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware log.error("Failed to process calculated field record: {}", cf, e); } }); - // TODO: why we need to do this loop if we do this inside the onFieldInitMsg? - calculatedFields.values().forEach(cf -> { - entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(cf); - }); PageDataIterable cfls = new PageDataIterable<>(pageLink -> cfDaoService.findAllCalculatedFieldLinksByTenantId(tenantId, pageLink), cfSettings.getInitTenantFetchPackSize()); cfls.forEach(link -> { onLinkInitMsg(new CalculatedFieldLinkInitMsg(link.getTenantId(), link)); diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityCalculatedFieldMarkStateDirtyMsg.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityCalculatedFieldDynamicArgumentsRefreshMsg.java similarity index 87% rename from application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityCalculatedFieldMarkStateDirtyMsg.java rename to application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityCalculatedFieldDynamicArgumentsRefreshMsg.java index ef9864aa83..fdf864611f 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityCalculatedFieldMarkStateDirtyMsg.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityCalculatedFieldDynamicArgumentsRefreshMsg.java @@ -23,7 +23,7 @@ import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg; import org.thingsboard.server.common.msg.queue.TbCallback; @Data -public class EntityCalculatedFieldMarkStateDirtyMsg implements ToCalculatedFieldSystemMsg { +public class EntityCalculatedFieldDynamicArgumentsRefreshMsg implements ToCalculatedFieldSystemMsg { private final TenantId tenantId; private final CalculatedFieldId cfId; @@ -31,7 +31,7 @@ public class EntityCalculatedFieldMarkStateDirtyMsg implements ToCalculatedField @Override public MsgType getMsgType() { - return MsgType.CF_ENTITY_MARK_STATE_DIRTY_MSG; + return MsgType.CF_ENTITY_DYNAMIC_ARGUMENTS_REFRESH_MSG; } } diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java index fc0e2262bb..a2eb5303be 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java @@ -152,8 +152,8 @@ public enum MsgType { CF_ENTITY_INIT_CF_MSG, CF_ENTITY_DELETE_MSG, - CF_SCHEDULED_INVALIDATION_MSG, - CF_ENTITY_MARK_STATE_DIRTY_MSG; + CF_DYNAMIC_ARGUMENTS_REFRESH_MSG, + CF_ENTITY_DYNAMIC_ARGUMENTS_REFRESH_MSG; @Getter private final boolean ignoreOnStart;