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 2685e7f434..93fd7498e5 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 @@ -159,10 +159,14 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } else { state.setCtx(ctx, actorCtx); } - if (state.isSizeOk()) { - processStateIfReady(state, Collections.emptyMap(), ctx, Collections.singletonList(ctx.getCfId()), null, null, msg.getCallback()); + if (msg.getStateAction() != StateAction.REFRESH_CTX) { + if (state.isSizeOk()) { + processStateIfReady(state, Collections.emptyMap(), ctx, Collections.singletonList(ctx.getCfId()), null, null, msg.getCallback()); + } else { + throw new RuntimeException(ctx.getSizeExceedsLimitMessage()); + } } else { - throw new RuntimeException(ctx.getSizeExceedsLimitMessage()); + msg.getCallback().onSuccess(); } } catch (Exception e) { log.debug("[{}][{}] Failed to initialize CF state", entityId, ctx.getCfId(), e); 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 b19ad1a8b4..97d5d0dcea 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 @@ -455,6 +455,8 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware stateAction = StateAction.REINIT; // refetch arguments, call state.init, then calculate } else if (newCfCtx.hasContextOnlyChanges(oldCfCtx)) { stateAction = StateAction.REPROCESS; // call state.setCtx, then calculate + } else if (newCfCtx.hasRefreshContextOnlyChanges(oldCfCtx)) { + stateAction = StateAction.REFRESH_CTX; } else { callback.onSuccess(); return; diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityInitCalculatedFieldMsg.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityInitCalculatedFieldMsg.java index 1e0025988d..49f2c691d3 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityInitCalculatedFieldMsg.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityInitCalculatedFieldMsg.java @@ -39,6 +39,7 @@ public class EntityInitCalculatedFieldMsg implements ToCalculatedFieldSystemMsg INIT, REINIT, RECREATE, - REPROCESS + REPROCESS, + REFRESH_CTX } } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java index c2c65d3327..489632b9b9 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java @@ -37,12 +37,14 @@ import org.thingsboard.server.common.data.cf.configuration.AlarmCalculatedFieldC import org.thingsboard.server.common.data.cf.configuration.Argument; import org.thingsboard.server.common.data.cf.configuration.ArgumentType; import org.thingsboard.server.common.data.cf.configuration.ArgumentsBasedCalculatedFieldConfiguration; +import org.thingsboard.server.common.data.cf.configuration.AttributesImmediateOutputStrategy; import org.thingsboard.server.common.data.cf.configuration.ExpressionBasedCalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.Output; import org.thingsboard.server.common.data.cf.configuration.PropagationCalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey; import org.thingsboard.server.common.data.cf.configuration.ScheduledUpdateSupportedCalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.SimpleCalculatedFieldConfiguration; +import org.thingsboard.server.common.data.cf.configuration.TimeSeriesImmediateOutputStrategy; import org.thingsboard.server.common.data.cf.configuration.aggregation.AggFunctionInput; import org.thingsboard.server.common.data.cf.configuration.aggregation.RelatedEntitiesAggregationCalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.aggregation.single.EntityAggregationCalculatedFieldConfiguration; @@ -623,16 +625,42 @@ public class CalculatedFieldCtx implements Closeable { return new CalculatedFieldEntityCtxId(tenantId, cfId, entityId); } + public boolean hasRefreshContextOnlyChanges(CalculatedFieldCtx other) { // has changes that do not require state recalculation + var thisConfig = calculatedField.getConfiguration(); + var otherConfig = other.getCalculatedField().getConfiguration(); + + var thisOutputStrategy = thisConfig.getOutput().getStrategy(); + var otherOutputStrategy = otherConfig.getOutput().getStrategy(); + + if (!thisOutputStrategy.getType().equals(otherOutputStrategy.getType())) { + return true; + } + + if (thisOutputStrategy instanceof TimeSeriesImmediateOutputStrategy thisTimeSeriesImmediateOutputStrategy + && otherOutputStrategy instanceof TimeSeriesImmediateOutputStrategy otherTimeSeriesImmediateOutputStrategy) { + return thisTimeSeriesImmediateOutputStrategy.getTtl() != otherTimeSeriesImmediateOutputStrategy.getTtl(); + } + + if (thisOutputStrategy instanceof AttributesImmediateOutputStrategy thisAttributesImmediateOutputStrategy + && otherOutputStrategy instanceof AttributesImmediateOutputStrategy otherAttributesImmediateOutputStrategy) { + boolean updateAttributesOnlyOnValueChangeChanged = thisAttributesImmediateOutputStrategy.isUpdateAttributesOnlyOnValueChange() != otherAttributesImmediateOutputStrategy.isUpdateAttributesOnlyOnValueChange(); + boolean sendAttributesUpdatedNotificationUpdated = thisAttributesImmediateOutputStrategy.isSendAttributesUpdatedNotification() != otherAttributesImmediateOutputStrategy.isSendAttributesUpdatedNotification(); + return updateAttributesOnlyOnValueChangeChanged || sendAttributesUpdatedNotificationUpdated; + } + + return false; + } + public boolean hasContextOnlyChanges(CalculatedFieldCtx other) { // has changes that do not require state reinit and will be picked up by the state on the fly if (calculatedField.getConfiguration() instanceof ExpressionBasedCalculatedFieldConfiguration && !Objects.equals(expression, other.expression)) { return true; } - if (!Objects.equals(output, other.output)) { + if (hasOutputChanges(other.output)) { return true; } if (calculatedField.getConfiguration() instanceof SimpleCalculatedFieldConfiguration thisConfig - && other.calculatedField.getConfiguration() instanceof SimpleCalculatedFieldConfiguration otherConfig - && thisConfig.isUseLatestTs() != otherConfig.isUseLatestTs()) { + && other.calculatedField.getConfiguration() instanceof SimpleCalculatedFieldConfiguration otherConfig + && thisConfig.isUseLatestTs() != otherConfig.isUseLatestTs()) { return true; } if (cfType == CalculatedFieldType.ALARM) { @@ -654,14 +682,14 @@ public class CalculatedFieldCtx implements Closeable { return true; } if (calculatedField.getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration thisConfig - && other.getCalculatedField().getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration otherConfig - && (thisConfig.getDeduplicationIntervalInSec() != otherConfig.getDeduplicationIntervalInSec() + && other.getCalculatedField().getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration otherConfig + && (thisConfig.getDeduplicationIntervalInSec() != otherConfig.getDeduplicationIntervalInSec() || !thisConfig.getMetrics().equals(otherConfig.getMetrics()) || thisConfig.isUseLatestTs() != otherConfig.isUseLatestTs())) { return true; } if (calculatedField.getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration thisConfig - && other.getCalculatedField().getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration otherConfig) { + && other.getCalculatedField().getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration otherConfig) { boolean metricsChanged = !Objects.equals(thisConfig.getMetrics(), otherConfig.getMetrics()); boolean watermarkChanged = !Objects.equals(thisConfig.getWatermark(), otherConfig.getWatermark()); return metricsChanged || watermarkChanged; @@ -695,9 +723,47 @@ public class CalculatedFieldCtx implements Closeable { return false; } + private boolean hasOutputChanges(Output otherOutput) { + if (!output.getType().equals(otherOutput.getType())) { + return true; + } + if (!output.getName().equals(otherOutput.getName())) { + return true; + } + if (output.getScope() != (otherOutput.getScope())) { + return true; + } + if (!Objects.equals(output.getDecimalsByDefault(), otherOutput.getDecimalsByDefault())) { + return true; + } + + var thisOutputStrategy = output.getStrategy(); + var otherOutputStrategy = otherOutput.getStrategy(); + + if (thisOutputStrategy instanceof TimeSeriesImmediateOutputStrategy thisTimeSeriesImmediateOutputStrategy + && otherOutputStrategy instanceof TimeSeriesImmediateOutputStrategy otherTimeSeriesImmediateOutputStrategy) { + boolean saveTimeSeriesUpdated = thisTimeSeriesImmediateOutputStrategy.isSaveTimeSeries() != otherTimeSeriesImmediateOutputStrategy.isSaveTimeSeries(); + boolean saveLatestUpdated = thisTimeSeriesImmediateOutputStrategy.isSaveLatest() != otherTimeSeriesImmediateOutputStrategy.isSaveLatest(); + boolean sendWsUpdateUpdated = thisTimeSeriesImmediateOutputStrategy.isSendWsUpdate() != otherTimeSeriesImmediateOutputStrategy.isSendWsUpdate(); + boolean processCfsUpdated = thisTimeSeriesImmediateOutputStrategy.isProcessCfs() != otherTimeSeriesImmediateOutputStrategy.isProcessCfs(); + return saveTimeSeriesUpdated || saveLatestUpdated || sendWsUpdateUpdated || processCfsUpdated; + } + + if (thisOutputStrategy instanceof AttributesImmediateOutputStrategy thisAttributesImmediateOutputStrategy + && otherOutputStrategy instanceof AttributesImmediateOutputStrategy otherAttributesImmediateOutputStrategy) { + + boolean saveTimeSeriesUpdated = thisAttributesImmediateOutputStrategy.isSaveAttribute() != otherAttributesImmediateOutputStrategy.isSaveAttribute(); + boolean sendWsUpdateUpdated = thisAttributesImmediateOutputStrategy.isSendWsUpdate() != otherAttributesImmediateOutputStrategy.isSendWsUpdate(); + boolean processCfsUpdated = thisAttributesImmediateOutputStrategy.isProcessCfs() != otherAttributesImmediateOutputStrategy.isProcessCfs(); + return saveTimeSeriesUpdated || sendWsUpdateUpdated || processCfsUpdated; + } + + return false; + } + private boolean hasGeofencingZoneGroupConfigurationChanges(CalculatedFieldCtx other) { if (calculatedField.getConfiguration() instanceof GeofencingCalculatedFieldConfiguration thisConfig - && other.calculatedField.getConfiguration() instanceof GeofencingCalculatedFieldConfiguration otherConfig) { + && other.calculatedField.getConfiguration() instanceof GeofencingCalculatedFieldConfiguration otherConfig) { return !thisConfig.getZoneGroups().equals(otherConfig.getZoneGroups()); } return false; @@ -705,7 +771,7 @@ public class CalculatedFieldCtx implements Closeable { private boolean hasRelatedEntitiesAggregationConfigurationChanges(CalculatedFieldCtx other) { if (calculatedField.getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration thisConfig - && other.calculatedField.getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration otherConfig) { + && other.calculatedField.getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration otherConfig) { return !thisConfig.getRelation().equals(otherConfig.getRelation()); } return false; @@ -713,7 +779,7 @@ public class CalculatedFieldCtx implements Closeable { private boolean hasEntityAggregationConfigurationChanges(CalculatedFieldCtx other) { if (calculatedField.getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration thisConfig - && other.calculatedField.getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration otherConfig) { + && other.calculatedField.getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration otherConfig) { return !thisConfig.getInterval().equals(otherConfig.getInterval()); } return false; @@ -738,7 +804,7 @@ public class CalculatedFieldCtx implements Closeable { yield true; } yield geofencingState.getLastDynamicArgumentsRefreshTs() < - System.currentTimeMillis() - scheduledUpdateIntervalMillis; + System.currentTimeMillis() - scheduledUpdateIntervalMillis; } default -> false; }; @@ -782,10 +848,10 @@ public class CalculatedFieldCtx implements Closeable { @Override public String toString() { return "CalculatedFieldCtx{" + - "cfId=" + cfId + - ", cfType=" + cfType + - ", entityId=" + entityId + - '}'; + "cfId=" + cfId + + ", cfType=" + cfType + + ", entityId=" + entityId + + '}'; } }