removed check for changing value when update state and moved timeout to config param

This commit is contained in:
IrynaMatveieva 2025-03-04 11:22:51 +02:00
parent 2363dc8edf
commit afd727b097
4 changed files with 15 additions and 16 deletions

View File

@ -644,6 +644,10 @@ public class ActorSystemContext {
@Getter @Getter
private String deviceStateNodeRateLimitConfig; private String deviceStateNodeRateLimitConfig;
@Value("${actors.calculated_fields.calculation_result_timeout:5}")
@Getter
private long cfCalculationResultTimeout;
@Getter @Getter
@Setter @Setter
private TbActorSystem actorSystem; private TbActorSystem actorSystem;

View File

@ -275,9 +275,9 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
private void processStateIfReady(CalculatedFieldCtx ctx, List<CalculatedFieldId> cfIdList, CalculatedFieldState state, UUID tbMsgId, TbMsgType tbMsgType, TbCallback callback) throws CalculatedFieldException { private void processStateIfReady(CalculatedFieldCtx ctx, List<CalculatedFieldId> cfIdList, CalculatedFieldState state, UUID tbMsgId, TbMsgType tbMsgType, TbCallback callback) throws CalculatedFieldException {
CalculatedFieldEntityCtxId ctxId = new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId); CalculatedFieldEntityCtxId ctxId = new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId);
boolean stateSizeOk; boolean stateSizeOk;
if (ctx.isInitialized() && state.isReady()) {
try { try {
CalculatedFieldResult calculationResult = state.performCalculation(ctx).get(5, TimeUnit.SECONDS); if (ctx.isInitialized() && state.isReady()) {
CalculatedFieldResult calculationResult = state.performCalculation(ctx).get(systemContext.getCfCalculationResultTimeout(), TimeUnit.SECONDS);
state.checkStateSize(ctxId, ctx.getMaxStateSize()); state.checkStateSize(ctxId, ctx.getMaxStateSize());
stateSizeOk = state.isSizeOk(); stateSizeOk = state.isSizeOk();
if (stateSizeOk) { if (stateSizeOk) {
@ -286,22 +286,19 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
systemContext.persistCalculatedFieldDebugEvent(tenantId, ctx.getCfId(), entityId, state.getArguments(), tbMsgId, tbMsgType, JacksonUtil.writeValueAsString(calculationResult.getResult()), null); systemContext.persistCalculatedFieldDebugEvent(tenantId, ctx.getCfId(), entityId, state.getArguments(), tbMsgId, tbMsgType, JacksonUtil.writeValueAsString(calculationResult.getResult()), null);
} }
} }
}
} catch (Exception e) { } catch (Exception e) {
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).msgId(tbMsgId).msgType(tbMsgType).arguments(state.getArguments()).cause(e).build(); throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).msgId(tbMsgId).msgType(tbMsgType).arguments(state.getArguments()).cause(e).build();
} } finally {
} else {
state.checkStateSize(ctxId, ctx.getMaxStateSize()); state.checkStateSize(ctxId, ctx.getMaxStateSize());
stateSizeOk = state.isSizeOk(); stateSizeOk = state.isSizeOk();
if (stateSizeOk) {
callback.onSuccess(); // State was updated but no calculation performed;
}
}
if (stateSizeOk) { if (stateSizeOk) {
cfStateService.persistState(ctxId, state, callback); cfStateService.persistState(ctxId, state, callback);
} else { } else {
removeStateAndRaiseSizeException(ctxId, CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).errorMessage(ctx.getSizeExceedsLimitMessage()).build(), callback); removeStateAndRaiseSizeException(ctxId, CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).errorMessage(ctx.getSizeExceedsLimitMessage()).build(), callback);
} }
} }
}
private void removeStateAndRaiseSizeException(CalculatedFieldEntityCtxId ctxId, CalculatedFieldException ex, TbCallback callback) { private void removeStateAndRaiseSizeException(CalculatedFieldEntityCtxId ctxId, CalculatedFieldException ex, TbCallback callback) {
// We remove the state, but remember that it is over-sized in a local map. // We remove the state, but remember that it is over-sized in a local map.

View File

@ -100,10 +100,6 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
if (newVersion == null || this.version == null || newVersion > this.version) { if (newVersion == null || this.version == null || newVersion > this.version) {
this.ts = singleValueEntry.getTs(); this.ts = singleValueEntry.getTs();
this.version = newVersion; this.version = newVersion;
BasicKvEntry newValue = singleValueEntry.getKvEntryValue();
if (this.kvEntryValue != null && this.kvEntryValue.getValue().equals(newValue.getValue())) {
return false;
}
this.kvEntryValue = singleValueEntry.getKvEntryValue(); this.kvEntryValue = singleValueEntry.getKvEntryValue();
return true; return true;
} }

View File

@ -512,6 +512,8 @@ actors:
enabled: "${ACTORS_CALCULATED_FIELD_DEBUG_MODE_RATE_LIMITS_PER_TENANT_ENABLED:true}" 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 # 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}" 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: debug:
settings: settings: