tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Renamed actor message types and java fields for clarity

pull/13920/head
dshvaika 1 year ago
parent
ef9ad6751c
commit
06515a5e13
7 changed files with 38 additions and 42 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 4
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldDynamicArgumentsRefreshMsg.java
  2. 4
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java
  3. 4
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
  4. 4
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerActor.java
  5. 56
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
  6. 4
      application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityCalculatedFieldDynamicArgumentsRefreshMsg.java
  7. 4
      common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java

4
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldScheduledInvalidationMsg.java → application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldDynamicArgumentsRefreshMsg.java
View File

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

4
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java
View File

@ -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;

4
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
View File

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

4
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerActor.java
View File

@ -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;

56
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
View File

@ -76,7 +76,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
private final Map<CalculatedFieldId, CalculatedFieldCtx> calculatedFields = new HashMap<>();
private final Map<EntityId, List<CalculatedFieldCtx>> entityIdCalculatedFields = new HashMap<>();
private final Map<EntityId, List<CalculatedFieldLink>> entityIdCalculatedFieldLinks = new HashMap<>();
private final Map<CalculatedFieldId, ScheduledFuture<?>> cfInvalidationScheduledTasks = new ConcurrentHashMap<>();
private final Map<CalculatedFieldId, ScheduledFuture<?>> 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<CalculatedFieldCtx> 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<CalculatedFieldLink> cfls = new PageDataIterable<>(pageLink -> cfDaoService.findAllCalculatedFieldLinksByTenantId(tenantId, pageLink), cfSettings.getInitTenantFetchPackSize());
cfls.forEach(link -> {
onLinkInitMsg(new CalculatedFieldLinkInitMsg(link.getTenantId(), link));

4
application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityCalculatedFieldMarkStateDirtyMsg.java → application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityCalculatedFieldDynamicArgumentsRefreshMsg.java
View File

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

4
common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java
View File

@ -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;

Write Preview
Loading…
Cancel
Save