From afd727b097ae4253c7bd8fe445a0c9197eb2d4ad Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Tue, 4 Mar 2025 11:22:51 +0200 Subject: [PATCH] removed check for changing value when update state and moved timeout to config param --- .../server/actors/ActorSystemContext.java | 4 ++++ ...CalculatedFieldEntityMessageProcessor.java | 21 ++++++++----------- .../ctx/state/SingleValueArgumentEntry.java | 4 ---- .../src/main/resources/thingsboard.yml | 2 ++ 4 files changed, 15 insertions(+), 16 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index 50a777840c..9ab7adfcc6 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -644,6 +644,10 @@ public class ActorSystemContext { @Getter private String deviceStateNodeRateLimitConfig; + @Value("${actors.calculated_fields.calculation_result_timeout:5}") + @Getter + private long cfCalculationResultTimeout; + @Getter @Setter private TbActorSystem actorSystem; 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 f7fc204c0f..279628c566 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 @@ -275,9 +275,9 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM private void processStateIfReady(CalculatedFieldCtx ctx, List cfIdList, CalculatedFieldState state, UUID tbMsgId, TbMsgType tbMsgType, TbCallback callback) throws CalculatedFieldException { CalculatedFieldEntityCtxId ctxId = new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId); boolean stateSizeOk; - if (ctx.isInitialized() && state.isReady()) { - try { - CalculatedFieldResult calculationResult = state.performCalculation(ctx).get(5, TimeUnit.SECONDS); + try { + if (ctx.isInitialized() && state.isReady()) { + CalculatedFieldResult calculationResult = state.performCalculation(ctx).get(systemContext.getCfCalculationResultTimeout(), TimeUnit.SECONDS); state.checkStateSize(ctxId, ctx.getMaxStateSize()); stateSizeOk = state.isSizeOk(); if (stateSizeOk) { @@ -286,21 +286,18 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM systemContext.persistCalculatedFieldDebugEvent(tenantId, ctx.getCfId(), entityId, state.getArguments(), tbMsgId, tbMsgType, JacksonUtil.writeValueAsString(calculationResult.getResult()), null); } } - } catch (Exception e) { - throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).msgId(tbMsgId).msgType(tbMsgType).arguments(state.getArguments()).cause(e).build(); } - } else { + } catch (Exception e) { + throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).msgId(tbMsgId).msgType(tbMsgType).arguments(state.getArguments()).cause(e).build(); + } finally { state.checkStateSize(ctxId, ctx.getMaxStateSize()); stateSizeOk = state.isSizeOk(); if (stateSizeOk) { - callback.onSuccess(); // State was updated but no calculation performed; + cfStateService.persistState(ctxId, state, callback); + } else { + removeStateAndRaiseSizeException(ctxId, CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).errorMessage(ctx.getSizeExceedsLimitMessage()).build(), callback); } } - if (stateSizeOk) { - cfStateService.persistState(ctxId, state, callback); - } else { - removeStateAndRaiseSizeException(ctxId, CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).errorMessage(ctx.getSizeExceedsLimitMessage()).build(), callback); - } } private void removeStateAndRaiseSizeException(CalculatedFieldEntityCtxId ctxId, CalculatedFieldException ex, TbCallback callback) { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java index a064a99935..a237f3d022 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java @@ -100,10 +100,6 @@ public class SingleValueArgumentEntry implements ArgumentEntry { if (newVersion == null || this.version == null || newVersion > this.version) { this.ts = singleValueEntry.getTs(); this.version = newVersion; - BasicKvEntry newValue = singleValueEntry.getKvEntryValue(); - if (this.kvEntryValue != null && this.kvEntryValue.getValue().equals(newValue.getValue())) { - return false; - } this.kvEntryValue = singleValueEntry.getKvEntryValue(); return true; } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index ec9925bf06..d0ccef8ba9 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -512,6 +512,8 @@ actors: enabled: "${ACTORS_CALCULATED_FIELD_DEBUG_MODE_RATE_LIMITS_PER_TENANT_ENABLED:true}" # The value of DEBUG mode rate limit. By default, no more than 50 thousand events per hour configuration: "${ACTORS_CALCULATED_FIELD_DEBUG_MODE_RATE_LIMITS_PER_TENANT_CONFIGURATION:50000:3600}" + # Time in seconds to receive calculation result. + calculation_result_timeout: "${ACTORS_CALCULATION_RESULT_TIMEOUT_SEC:5}" debug: settings: