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

refactoring

pull/14141/head
IrynaMatveieva 10 months ago
parent
9ac6219882
commit
4137a4a615
3 changed files with 117 additions and 136 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. 105
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
  2. 139
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
  3. 9
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldRelatedEntityMsg.java

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

@ -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<CalculatedFieldId, CalculatedFieldState> 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<String, ArgumentEntry> 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<String, ArgumentEntry> 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<String, ArgumentEntry> updateAggregationState(EntityId relatedEntityId, CalculatedFieldState state, CalculatedFieldCtx ctx) {
Map<String, ArgumentEntry> fetchedArgs = fetchAggArguments(ctx, relatedEntityId);
Map<String, ArgumentEntry> 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<String, ArgumentEntry> fetchAggArguments(CalculatedFieldCtx ctx, EntityId entityId) {
ListenableFuture<Map<String, ArgumentEntry>> argumentsFuture = cfService.fetchAggEntityArguments(ctx, entityId);

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

@ -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<CalculatedFieldCtx> toIdMatches = new ArrayList<>();
List<CalculatedFieldCtx> cfsByToId = getCalculatedFieldsByEntityId(toId);
List<CalculatedFieldCtx> cfsByToProfileId = getCalculatedFieldsByEntityId(toIdProfile);
List<CalculatedFieldCtx> 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<CalculatedFieldCtx> fromIdMatches = new ArrayList<>();
List<CalculatedFieldCtx> cfsByFromId = getCalculatedFieldsByEntityId(fromId);
List<CalculatedFieldCtx> cfsByFromProfileId = getCalculatedFieldsByEntityId(fromIdProfile);
List<CalculatedFieldCtx> 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<CalculatedFieldCtx> toIdMatches = new ArrayList<>();
List<CalculatedFieldCtx> cfsByToId = getCalculatedFieldsByEntityId(toId);
List<CalculatedFieldCtx> cfsByToProfileId = getCalculatedFieldsByEntityId(toIdProfile);
List<CalculatedFieldCtx> 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<CalculatedFieldCtx> fromIdMatches = new ArrayList<>();
List<CalculatedFieldCtx> cfsByFromId = getCalculatedFieldsByEntityId(fromId);
List<CalculatedFieldCtx> cfsByFromProfileId = getCalculatedFieldsByEntityId(fromIdProfile);
List<CalculatedFieldCtx> 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<EntityId, CalculatedFieldCtx, TbCallback> relationAction) {
List<CalculatedFieldCtx> 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<CalculatedFieldCtx> 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<CalculatedFieldCtx> cfs = new ArrayList<>();
cfs.addAll(getCalculatedFieldsByEntityId(entityId));
cfs.addAll(getCalculatedFieldsByEntityId(getProfileId(tenantId, entityId)));
List<CalculatedFieldCtx> cfs = getCalculatedFieldsByEntityIdAndProfile(entityId);
if (cfs.isEmpty()) {
msgCallback.onSuccess();
return;
@ -695,6 +637,13 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
return result;
}
private List<CalculatedFieldCtx> getCalculatedFieldsByEntityIdAndProfile(EntityId entityId) {
List<CalculatedFieldCtx> cfsByEntityIdAndProfile = new ArrayList<>();
cfsByEntityIdAndProfile.addAll(getCalculatedFieldsByEntityId(entityId));
cfsByEntityIdAndProfile.addAll(getCalculatedFieldsByEntityId(getProfileId(tenantId, entityId)));
return cfsByEntityIdAndProfile;
}
private List<CalculatedFieldLink> 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) {

9
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldRelatedEntityMsg.java
View File

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

Write Preview
Loading…
Cancel
Save