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 bf84163a8b..35cf9cb467 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -860,8 +860,9 @@ public class ActorSystemContext { if (errorMessage != null) { eventBuilder.error(errorMessage); } - - ListenableFuture future = eventService.saveAsync(eventBuilder.build()); + CalculatedFieldDebugEvent event = eventBuilder.build(); + log.debug("Persisting calculated field debug event: {}", event); + ListenableFuture future = eventService.saveAsync(event); Futures.addCallback(future, CALCULATED_FIELD_DEBUG_EVENT_ERROR_CALLBACK, MoreExecutors.directExecutor()); } catch (IllegalArgumentException ex) { log.warn("Failed to persist calculated field debug message", ex); 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 ca97dae51b..9cee93611e 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 @@ -175,7 +175,6 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM public void process(CalculatedFieldArgumentResetMsg msg) throws CalculatedFieldException { log.debug("[{}] Processing CF argument reset msg.", entityId); var ctx = msg.getCtx(); - var callback = new MultipleTbCallback(CALLBACKS_PER_CF, msg.getCallback()); try { Map dynamicSourceArgs = ctx.getArguments().entrySet().stream() .filter(entry -> entry.getValue().hasOwnerSource()) @@ -184,7 +183,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM Map fetchedArgs = cfService.fetchArgsFromDb(tenantId, entityId, dynamicSourceArgs); fetchedArgs.values().forEach(arg -> arg.setForceResetPrevious(true)); - processArgumentValuesUpdate(ctx, Collections.singletonList(ctx.getCfId()), callback, fetchedArgs, null, null); + processArgumentValuesUpdate(ctx, Collections.singletonList(ctx.getCfId()), msg.getCallback(), fetchedArgs, null, null); } catch (Exception e) { throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build(); } @@ -277,7 +276,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM public void process(EntityCalculatedFieldTelemetryMsg msg) throws CalculatedFieldException { log.trace("[{}] Processing CF telemetry msg: {}", msg.getEntityId(), msg); var proto = msg.getProto(); - var numberOfCallbacks = CALLBACKS_PER_CF * (msg.getEntityIdFields().size() + msg.getProfileIdFields().size()); + var numberOfCallbacks = msg.getEntityIdFields().size() + msg.getProfileIdFields().size(); MultipleTbCallback callback = new MultipleTbCallback(numberOfCallbacks, msg.getCallback()); List cfIdList = getCalculatedFieldIds(proto); Set cfIdSet = new HashSet<>(cfIdList); @@ -293,11 +292,11 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM log.trace("[{}] Processing CF link telemetry msg: {}", msg.getEntityId(), msg); var proto = msg.getProto(); var ctx = msg.getCtx(); - var callback = new MultipleTbCallback(CALLBACKS_PER_CF, msg.getCallback()); + var callback = msg.getCallback(); try { List cfIds = getCalculatedFieldIds(proto); if (cfIds.contains(ctx.getCfId())) { - callback.onSuccess(CALLBACKS_PER_CF); + callback.onSuccess(); } else { if (proto.getTsDataCount() > 0) { processArgumentValuesUpdate(ctx, cfIds, callback, mapToArguments(ctx, msg.getEntityId(), proto.getTsDataList()), toTbMsgId(proto), toTbMsgType(proto)); @@ -308,7 +307,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } else if (proto.getRemovedAttrKeysCount() > 0) { processArgumentValuesUpdate(ctx, cfIds, callback, mapToArgumentsWithDefaultValue(ctx, msg.getEntityId(), proto.getScope(), proto.getRemovedAttrKeysList()), toTbMsgId(proto), toTbMsgType(proto)); } else { - callback.onSuccess(CALLBACKS_PER_CF); + callback.onSuccess(); } } } catch (Exception e) { @@ -317,10 +316,10 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } } - private void process(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, Collection cfIds, List cfIdList, MultipleTbCallback callback) throws CalculatedFieldException { + private void process(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, Collection cfIds, List cfIdList, TbCallback callback) throws CalculatedFieldException { try { if (cfIds.contains(ctx.getCfId())) { - callback.onSuccess(CALLBACKS_PER_CF); + callback.onSuccess(); } else { if (proto.getTsDataCount() > 0) { processTelemetry(ctx, proto, cfIdList, callback); @@ -331,7 +330,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } else if (proto.getRemovedAttrKeysCount() > 0) { processRemovedAttributes(ctx, proto, cfIdList, callback); } else { - callback.onSuccess(CALLBACKS_PER_CF); + callback.onSuccess(); } } } catch (Exception e) { @@ -371,27 +370,27 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM msg.getCallback().onSuccess(); } - private void processTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, MultipleTbCallback callback) throws CalculatedFieldException { + private void processTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, TbCallback callback) throws CalculatedFieldException { processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArguments(ctx, proto.getTsDataList()), toTbMsgId(proto), toTbMsgType(proto)); } - private void processAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, MultipleTbCallback callback) throws CalculatedFieldException { + private void processAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, TbCallback callback) throws CalculatedFieldException { processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArguments(ctx, proto.getScope(), proto.getAttrDataList()), toTbMsgId(proto), toTbMsgType(proto)); } - private void processRemovedTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, MultipleTbCallback callback) throws CalculatedFieldException { + private void processRemovedTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, TbCallback callback) throws CalculatedFieldException { processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArgumentsWithFetchedValue(ctx, entityId, proto.getRemovedTsKeysList()), toTbMsgId(proto), toTbMsgType(proto)); } - private void processRemovedAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, MultipleTbCallback callback) throws CalculatedFieldException { + private void processRemovedAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, TbCallback callback) throws CalculatedFieldException { processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArgumentsWithDefaultValue(ctx, proto.getScope(), proto.getRemovedAttrKeysList()), toTbMsgId(proto), toTbMsgType(proto)); } - private void processArgumentValuesUpdate(CalculatedFieldCtx ctx, List cfIdList, MultipleTbCallback callback, + private void processArgumentValuesUpdate(CalculatedFieldCtx ctx, List cfIdList, TbCallback callback, Map newArgValues, UUID tbMsgId, TbMsgType tbMsgType) throws CalculatedFieldException { if (newArgValues.isEmpty()) { log.debug("[{}] No new argument values to process for CF.", ctx.getCfId()); - callback.onSuccess(CALLBACKS_PER_CF); + callback.onSuccess(); } CalculatedFieldState state = states.get(ctx.getCfId()); boolean justRestored = false; @@ -418,7 +417,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM cfIdList.add(ctx.getCfId()); processStateIfReady(state, updatedArgs, ctx, cfIdList, tbMsgId, tbMsgType, callback); } else { - callback.onSuccess(CALLBACKS_PER_CF); + callback.onSuccess(); } } else { throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).errorMessage(ctx.getSizeExceedsLimitMessage()).build(); @@ -459,6 +458,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM private void processStateIfReady(CalculatedFieldState state, Map updatedArgs, CalculatedFieldCtx ctx, List cfIdList, UUID tbMsgId, TbMsgType tbMsgType, TbCallback callback) throws CalculatedFieldException { + callback = new MultipleTbCallback(CALLBACKS_PER_CF, callback); log.trace("[{}][{}] Processing state if ready. Current args: {}, updated args: {}", entityId, ctx.getCfId(), state.getArguments(), updatedArgs); CalculatedFieldEntityCtxId ctxId = new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId); boolean stateSizeChecked = false; 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 96f7e9aa80..40707e32f3 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 @@ -134,7 +134,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware public void stop() { log.info("[{}] Stopping CF manager actor.", tenantId); - calculatedFields.values().forEach(CalculatedFieldCtx::stop); + calculatedFields.values().forEach(CalculatedFieldCtx::close); calculatedFields.clear(); entityIdCalculatedFields.clear(); entityIdCalculatedFieldLinks.clear(); @@ -398,7 +398,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware log.debug("[{}] Failed to lookup CF by id [{}]", tenantId, cfId); callback.onSuccess(); } else { - var newCfCtx = getCfCtx(newCf); // fixme wtf? why isn't oldCfCtx closed properly? when to close it? + var newCfCtx = getCfCtx(newCf); try { newCfCtx.init(); } catch (Exception e) { @@ -441,14 +441,26 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware return; } - applyToTargetCfEntityActors(newCfCtx, callback, (id, cb) -> initCfForEntity(id, newCfCtx, stateAction, cb)); + applyToTargetCfEntityActors(newCfCtx, new TbCallback() { + @Override + public void onSuccess() { + oldCfCtx.close(); + callback.onSuccess(); + } + + @Override + public void onFailure(Throwable t) { + oldCfCtx.close(); + callback.onFailure(t); + } + }, (id, cb) -> initCfForEntity(id, newCfCtx, stateAction, cb)); } } } private void onCfDeleted(ComponentLifecycleMsg msg, TbCallback callback) { var cfId = new CalculatedFieldId(msg.getEntityId().getId()); - var cfCtx = calculatedFields.remove(cfId); // fixme wtf? why isn't ctx closed properly? + var cfCtx = calculatedFields.remove(cfId); aggCalculatedFields.remove(cfId); if (cfCtx == null) { log.debug("[{}] CF was already deleted [{}]", tenantId, cfId); @@ -457,7 +469,19 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } entityIdCalculatedFields.get(cfCtx.getEntityId()).remove(cfCtx); deleteLinks(cfCtx); - applyToTargetCfEntityActors(cfCtx, callback, (id, cb) -> deleteCfForEntity(id, cfId, cb)); + applyToTargetCfEntityActors(cfCtx, new TbCallback() { + @Override + public void onSuccess() { + cfCtx.close(); + callback.onSuccess(); + } + + @Override + public void onFailure(Throwable t) { + cfCtx.close(); + callback.onFailure(t); + } + }, (id, cb) -> deleteCfForEntity(id, cfId, cb)); } public void onTelemetryMsg(CalculatedFieldTelemetryMsg msg) { 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 40e414920a..218d03ac38 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 @@ -60,6 +60,7 @@ import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingCalculatedFieldState; import org.thingsboard.server.service.telemetry.AlarmSubscriptionService; +import java.io.Closeable; import java.util.ArrayList; import java.util.HashMap; import java.util.LinkedHashMap; @@ -68,11 +69,10 @@ import java.util.Map; import java.util.Objects; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; -import java.util.stream.Stream; @Data @Slf4j -public class CalculatedFieldCtx { +public class CalculatedFieldCtx implements Closeable { private CalculatedField calculatedField; @@ -211,15 +211,12 @@ public class CalculatedFieldCtx { } case ALARM -> { AlarmCalculatedFieldConfiguration configuration = (AlarmCalculatedFieldConfiguration) calculatedField.getConfiguration(); - Stream rules = configuration.getCreateRules().values().stream(); - if (configuration.getClearRule() != null) { - rules = Stream.concat(rules, Stream.of(configuration.getClearRule())); - } - rules.map(rule -> rule.getCondition().getExpression()).forEach(expression -> { - if (expression instanceof TbelAlarmConditionExpression tbelExpression) { - initTbelExpression(tbelExpression.getExpression()); - } - }); + configuration.getAllRules().map(rule -> rule.getValue().getCondition().getExpression()) + .forEach(expression -> { + if (expression instanceof TbelAlarmConditionExpression tbelExpression) { + initTbelExpression(tbelExpression.getExpression()); + } + }); initialized = true; } case PROPAGATION -> { @@ -294,7 +291,6 @@ public class CalculatedFieldCtx { public ScheduledFuture scheduleReevaluation(long delayMs, TbActorRef actorCtx) { log.debug("[{}] Scheduling CF reevaluation in {} ms", cfId, delayMs); - // TODO: use single lazy-loaded instance of CalculatedFieldReevaluateMsg return systemContext.scheduleMsgWithDelay(actorCtx, new CalculatedFieldReevaluateMsg(tenantId, this), delayMs); } @@ -578,8 +574,17 @@ public class CalculatedFieldCtx { if (!Objects.equals(output, other.output)) { return true; } - if (cfType == CalculatedFieldType.ALARM && !calculatedField.getName().equals(other.getCalculatedField().getName())) { - return true; + if (cfType == CalculatedFieldType.ALARM) { + if (!calculatedField.getName().equals(other.getCalculatedField().getName())) { + return true; + } + + var thisConfig = (AlarmCalculatedFieldConfiguration) calculatedField.getConfiguration(); + var otherConfig = (AlarmCalculatedFieldConfiguration) other.getCalculatedField().getConfiguration(); + if (!thisConfig.rulesEqual(otherConfig, AlarmRule::equals)) { + // if the rules have any changes not tracked by hasStateChanges + return true; + } } if (scheduledUpdateIntervalMillis != other.scheduledUpdateIntervalMillis) { return true; @@ -599,8 +604,10 @@ public class CalculatedFieldCtx { if (cfType == CalculatedFieldType.ALARM) { var thisConfig = (AlarmCalculatedFieldConfiguration) calculatedField.getConfiguration(); var otherConfig = (AlarmCalculatedFieldConfiguration) other.getCalculatedField().getConfiguration(); - if (!thisConfig.getCreateRules().equals(otherConfig.getCreateRules()) || - !Objects.equals(thisConfig.getClearRule(), otherConfig.getClearRule())) { + if (!thisConfig.rulesEqual(otherConfig, (thisRule, otherRule) -> { + return thisRule.getCondition().getType() == otherRule.getCondition().getType(); + })) { + // reinitializing only if the rule list changed, or if a condition type changed for any rule return true; } } @@ -654,12 +661,17 @@ public class CalculatedFieldCtx { }; } - public void stop() { - if (tbelExpressions != null) { - tbelExpressions.values().forEach(CalculatedFieldScriptEngine::destroy); - } - if (simpleExpressions != null) { - simpleExpressions.values().forEach(ThreadLocal::remove); + @Override + public void close() { + try { + if (tbelExpressions != null) { + tbelExpressions.values().forEach(CalculatedFieldScriptEngine::destroy); + } + if (simpleExpressions != null) { + simpleExpressions.values().forEach(ThreadLocal::remove); + } + } catch (Exception e) { + log.warn("Failed to stop {}", this, e); } } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmCalculatedFieldState.java index 02f1725cf2..518121a2d0 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmCalculatedFieldState.java @@ -103,6 +103,18 @@ public class AlarmCalculatedFieldState extends BaseCalculatedFieldState { this.configuration = getConfiguration(ctx); this.alarmType = ctx.getCalculatedField().getName(); + Map createRules = configuration.getCreateRules(); + createRules.forEach((severity, rule) -> { + AlarmRuleState ruleState = createRuleStates.get(severity); + if (ruleState != null) { + ruleState.setAlarmRule(rule); + } + }); + AlarmRule clearRule = configuration.getClearRule(); + if (clearRule != null && clearRuleState != null) { + clearRuleState.setAlarmRule(clearRule); + } + if (currentAlarm != null && !currentAlarm.getType().equals(alarmType)) { currentAlarm = null; initialFetchDone = false; @@ -265,7 +277,7 @@ public class AlarmCalculatedFieldState extends BaseCalculatedFieldState { clearState(state); } AlarmApiCallResult clearResult = ctx.getAlarmService().clearAlarm( - ctx.getTenantId(), currentAlarm.getId(), System.currentTimeMillis(), createDetails(clearRuleState), true + ctx.getTenantId(), currentAlarm.getId(), System.currentTimeMillis(), createDetails(clearRuleState), false ); if (clearResult.isCleared()) { result = TbAlarmResult.builder() diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmRuleState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmRuleState.java index 9c1b966878..8612607dfb 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmRuleState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmRuleState.java @@ -237,7 +237,15 @@ public class AlarmRuleState { } public void clear() { + clearRepeatingConditionState(); + clearDurationConditionState(); + } + + private void clearRepeatingConditionState() { eventCount = 0L; + } + + private void clearDurationConditionState() { firstEventTs = 0L; lastEventTs = 0L; duration = 0L; @@ -289,6 +297,20 @@ public class AlarmRuleState { public void setAlarmRule(AlarmRule alarmRule) { this.alarmRule = alarmRule; this.condition = alarmRule.getCondition(); + + // clearing state for other condition types (possibly left from a previous condition type) + switch (condition.getType()) { + case SIMPLE -> { + clearRepeatingConditionState(); + clearDurationConditionState(); + } + case REPEATING -> { + clearDurationConditionState(); + } + case DURATION -> { + clearRepeatingConditionState(); + } + } } public StateInfo getStateInfo() { diff --git a/application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java b/application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java index f762cf5be5..48a318f981 100644 --- a/application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java +++ b/application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java @@ -257,7 +257,6 @@ public class EntityStateSourcingListener { if (calculatedFieldCache.hasCalculatedFields(tenantId, alarm.getOriginator(), ctx -> ctx.getCfType() == CalculatedFieldType.ALARM)) { ToCalculatedFieldMsg msg = ToCalculatedFieldMsg.newBuilder() .setEventMsg(toProto(event)) - .addCfTypes(CalculatedFieldType.ALARM.name()) .build(); tbClusterService.pushMsgToCalculatedFields(tenantId, alarm.getOriginator(), msg, new TbQueueCallback() { @Override diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java index 952ca845f0..0449d116c8 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java @@ -15,7 +15,6 @@ */ package org.thingsboard.server.service.queue; -import com.google.protobuf.ProtocolStringList; import jakarta.annotation.PreDestroy; import org.apache.commons.collections4.CollectionUtils; import org.springframework.beans.factory.annotation.Value; @@ -28,7 +27,6 @@ import org.thingsboard.server.actors.calculatedField.CalculatedFieldLinkedTeleme import org.thingsboard.server.actors.calculatedField.CalculatedFieldTelemetryMsg; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.EntityType; -import org.thingsboard.server.common.data.cf.CalculatedFieldType; import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; @@ -61,7 +59,6 @@ import org.thingsboard.server.service.queue.processing.AbstractPartitionBasedCon import org.thingsboard.server.service.queue.processing.IdMsgPair; import org.thingsboard.server.service.security.auth.jwt.settings.JwtSettingsService; -import java.util.EnumSet; import java.util.List; import java.util.Set; import java.util.UUID; @@ -183,8 +180,7 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractPartitionBa } private void processMsg(ToCalculatedFieldMsg toCfMsg, UUID id, TbCallback callback) { - Set cfTypes = getCfTypes(toCfMsg.getCfTypesList()); - if (toCfMsg.hasTelemetryMsg()) { // TODO: add CF type filter to the message. or just rename the CF strategy to "Process alarms and calculated fields + if (toCfMsg.hasTelemetryMsg()) { log.trace("[{}] Forwarding regular telemetry message for processing {}", id, toCfMsg.getTelemetryMsg()); forwardToActorSystem(toCfMsg.getTelemetryMsg(), callback); } else if (toCfMsg.hasLinkedTelemetryMsg()) { @@ -264,18 +260,6 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractPartitionBa return TenantId.fromUUID(new UUID(tenantIdMSB, tenantIdLSB)); } - private Set getCfTypes(ProtocolStringList cfTypesList) { - Set cfTypes; - if (cfTypesList.isEmpty()) { - cfTypes = EnumSet.allOf(CalculatedFieldType.class); - } else { - cfTypes = cfTypesList.stream() - .map(CalculatedFieldType::valueOf) - .collect(Collectors.toSet()); - } - return cfTypes; - } - @Override protected void stopConsumers() { super.stopConsumers(); diff --git a/application/src/test/java/org/thingsboard/server/cf/AlarmRulesTest.java b/application/src/test/java/org/thingsboard/server/cf/AlarmRulesTest.java index 2b43373ee5..f71bdd02a8 100644 --- a/application/src/test/java/org/thingsboard/server/cf/AlarmRulesTest.java +++ b/application/src/test/java/org/thingsboard/server/cf/AlarmRulesTest.java @@ -458,7 +458,14 @@ public class AlarmRulesTest extends AbstractControllerTest { schedule = schedule.replace("\"enabled\":false", "\"enabled\":true"); postAttributes(deviceId, AttributeScope.SERVER_SCOPE, "{\"schedule\":" + schedule + "}"); - checkAlarmResult(calculatedField, alarmResult -> { + + await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { + // checking multiple debug events due to scheduled reevaluation (which also produces debug events) + CalculatedFieldDebugEvent debugEvent = getDebugEvents(calculatedField.getId(), 5).stream() + .filter(event -> event.getResult() != null) + .findFirst().orElse(null); + assertThat(debugEvent).isNotNull(); + TbAlarmResult alarmResult = JacksonUtil.fromString(debugEvent.getResult(), TbAlarmResult.class); assertThat(alarmResult.isCreated()).isTrue(); assertThat(alarmResult.getAlarm().getSeverity()).isEqualTo(AlarmSeverity.CRITICAL); assertThat(alarmResult.getAlarm().getStatus()).isEqualTo(AlarmStatus.ACTIVE_UNACK); @@ -638,11 +645,11 @@ public class AlarmRulesTest extends AbstractControllerTest { AlarmSeverity.CRITICAL, new Condition("return temperature >= 50 && humidity >= 50;", null, null) ); CalculatedField calculatedField = createAlarmCf(deviceId, "High Temperature and Humidity Alarm", - arguments, createRules, null); - AlarmCalculatedFieldConfiguration configuration = (AlarmCalculatedFieldConfiguration) calculatedField.getConfiguration(); - configuration.getCreateRules().get(AlarmSeverity.CRITICAL).setAlarmDetails(""" - temperature is ${temperature}, humidity is ${humidity}"""); - calculatedField = saveCalculatedField(calculatedField); + arguments, createRules, null, configuration -> { + configuration.getCreateRules().get(AlarmSeverity.CRITICAL).setAlarmDetails( + "temperature is ${temperature}, humidity is ${humidity}" + ); + }); postTelemetry(deviceId, "{\"temperature\":50}"); postAttributes(deviceId, AttributeScope.SERVER_SCOPE, "{\"humidity\":50}"); @@ -653,6 +660,18 @@ public class AlarmRulesTest extends AbstractControllerTest { assertThat(alarmResult.getAlarm().getDetails().get("data").asText()) .isEqualTo("temperature is 50, humidity is 50"); }); + + ((AlarmCalculatedFieldConfiguration) calculatedField.getConfiguration()).getCreateRules().get(AlarmSeverity.CRITICAL).setAlarmDetails( + "UPDATED temperature is ${temperature}, humidity is ${humidity}" + ); + calculatedField = saveCalculatedField(calculatedField); + checkAlarmResult(calculatedField, alarmResult -> { + assertThat(alarmResult.isCreated()).isFalse(); + assertThat(alarmResult.isUpdated()).isTrue(); + assertThat(alarmResult.getAlarm().getSeverity()).isEqualTo(AlarmSeverity.CRITICAL); + assertThat(alarmResult.getAlarm().getDetails().get("data").asText()) + .isEqualTo("UPDATED temperature is 50, humidity is 50"); + }); } @Test @@ -683,7 +702,12 @@ public class AlarmRulesTest extends AbstractControllerTest { Thread.sleep(10000); assertThat(getLatestAlarmResult(calculatedField.getId())).isNull(); - checkAlarmResult(calculatedField, alarmResult -> { + await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { + CalculatedFieldDebugEvent debugEvent = getDebugEvents(calculatedField.getId(), 5).stream() + .filter(event -> event.getResult() != null) + .findFirst().orElse(null); + assertThat(debugEvent).isNotNull(); + TbAlarmResult alarmResult = JacksonUtil.fromString(debugEvent.getResult(), TbAlarmResult.class); assertThat(alarmResult.isCreated()).isTrue(); assertThat(alarmResult.getAlarm().getSeverity()).isEqualTo(AlarmSeverity.CRITICAL); assertThat(alarmResult.getAlarm().getStatus()).isEqualTo(AlarmStatus.ACTIVE_UNACK); @@ -764,24 +788,14 @@ public class AlarmRulesTest extends AbstractControllerTest { String alarmType, Map arguments, Map createConditions, - Condition clearCondition) { + Condition clearCondition, + Consumer... modifier) { Map createRules = new HashMap<>(); createConditions.forEach((severity, condition) -> { createRules.put(severity, toAlarmRule(condition)); }); AlarmRule clearRule = clearCondition != null ? toAlarmRule(clearCondition) : null; - CalculatedField calculatedField = createAlarmCf(entityId, alarmType, arguments, createRules, clearRule); - - CalculatedFieldDebugEvent debugEvent = await().atMost(TIMEOUT, TimeUnit.SECONDS).until(() -> getDebugEvents(calculatedField.getId(), 1), events -> !events.isEmpty()).get(0); - latestEventId = debugEvent.getId(); - return calculatedField; - } - private CalculatedField createAlarmCf(EntityId entityId, - String alarmType, - Map arguments, - Map createRules, - AlarmRule clearRule) { CalculatedField calculatedField = new CalculatedField(); calculatedField.setEntityId(entityId); calculatedField.setName(alarmType); @@ -792,7 +806,16 @@ public class AlarmRulesTest extends AbstractControllerTest { configuration.setClearRule(clearRule); calculatedField.setConfiguration(configuration); calculatedField.setDebugSettings(DebugSettings.all()); - return saveCalculatedField(calculatedField); + if (modifier.length > 0) { + modifier[0].accept(configuration); + } + CalculatedField savedCalculatedField = saveCalculatedField(calculatedField); + + CalculatedFieldDebugEvent debugEvent = await().atMost(TIMEOUT, TimeUnit.SECONDS) + .until(() -> getDebugEvents(savedCalculatedField.getId(), 1), + events -> !events.isEmpty()).get(0); + latestEventId = debugEvent.getId(); + return savedCalculatedField; } private AlarmRule toAlarmRule(Condition condition) { diff --git a/application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java b/application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java index 068c5f851d..a7b67baf3f 100644 --- a/application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java +++ b/application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java @@ -17,10 +17,10 @@ package org.thingsboard.server.cf; import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.ObjectNode; -import lombok.extern.slf4j.Slf4j; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.springframework.test.annotation.DirtiesContext; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; @@ -64,8 +64,8 @@ import static org.awaitility.Awaitility.await; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; import static org.thingsboard.server.cf.CalculatedFieldIntegrationTest.POLL_INTERVAL; -@Slf4j @DaoSqlTest +@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD) public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractControllerTest { private Tenant savedTenant; @@ -79,10 +79,10 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr private AssetProfile assetProfile; private Asset asset; - private long deduplicationInterval = 10; + private final long deduplicationInterval = 5; @Before - public void beforeTest() throws Exception { + public void beforeEach() throws Exception { loginSysAdmin(); updateDefaultTenantProfileConfig(tenantProfileConfig -> { @@ -135,7 +135,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr createOccupancyCF(assetProfile.getId()); - await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("create CF and perform initial aggregation").atMost(TIMEOUT, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset.getId(), Map.of( @@ -153,7 +153,9 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr postTelemetry(device3.getId(), "{\"occupied\":true}"); - await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("update telemetry and perform aggregation") + .atLeast(deduplicationInterval / 2, TimeUnit.SECONDS) + .atMost(TIMEOUT, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset2.getId(), Map.of( @@ -200,7 +202,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr postTelemetry(device3.getId(), "{\"occupied\":false}"); - await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) + await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset2.getId(), Map.of( @@ -284,7 +286,9 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr postTelemetry(device1.getId(), "{\"occupied\":false}"); - await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("update telemetry and perform aggregation") + .atLeast(deduplicationInterval / 2, TimeUnit.SECONDS) + .atMost(TIMEOUT, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset.getId(), Map.of( @@ -323,14 +327,15 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr postTelemetry(device1.getId(), "{\"occupied\":false}"); - await().alias("update telemetry -> no changes").atMost(deduplicationInterval / 2, TimeUnit.SECONDS) + await().alias("update telemetry -> no changes").atMost(TIMEOUT, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(this::checkInitialCalculationValues); postTelemetry(device2.getId(), "{\"occupied\":false}"); - await().alias("create CF and perform initial calculation").atMost(deduplicationInterval, TimeUnit.SECONDS) - .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + await().alias("create CF and perform initial calculation") + .atLeast(deduplicationInterval / 2, TimeUnit.SECONDS) + .atMost(TIMEOUT, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset.getId(), Map.of( "freeSpaces", "2", @@ -517,7 +522,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr postTelemetry(device2.getId(), "{\"temperature\":19.6}"); CalculatedField cf = createAvgTemperatureCF(asset.getId()); - await().alias("create avg temp cf and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("create avg temp cf and perform initial aggregation").atMost(TIMEOUT, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset.getId(), Map.of("avgTemperature", "24")); @@ -531,7 +536,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr configuration.setMetrics(Map.of("maxTemperature", aggMetric)); saveCalculatedField(cf); - await().alias("update metrics and perform aggregation").atMost(deduplicationInterval / 2, TimeUnit.SECONDS) + await().alias("update metrics and perform aggregation").atMost(TIMEOUT, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset.getId(), Map.of("maxTemperature", "24")); @@ -540,7 +545,9 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr postTelemetry(device1.getId(), "{\"temperature\":101.3}"); postTelemetry(device2.getId(), "{\"temperature\":25.8}"); - await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("update telemetry and perform aggregation") + .atLeast(deduplicationInterval / 2, TimeUnit.SECONDS) + .atMost(TIMEOUT, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset.getId(), Map.of("maxTemperature", "26")); diff --git a/application/src/test/resources/logback-test.xml b/application/src/test/resources/logback-test.xml index 56dbbfc125..13c93da411 100644 --- a/application/src/test/resources/logback-test.xml +++ b/application/src/test/resources/logback-test.xml @@ -17,8 +17,6 @@ - - diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/alarm/AlarmService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/alarm/AlarmService.java index 82ef7f4e8d..05abc4b0c7 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/alarm/AlarmService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/alarm/AlarmService.java @@ -51,10 +51,6 @@ import java.util.UUID; public interface AlarmService extends EntityDaoService { - /* - * New API, since 3.5. - */ - /** * Designed for atomic operations over active alarms. * Only one active alarm may exist for the pair {originatorId, alarmType} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/AlarmRule.java b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/AlarmRule.java index ab7adcbd48..9a4e875154 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/AlarmRule.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/AlarmRule.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.common.data.alarm.rule; +import com.fasterxml.jackson.annotation.JsonIgnore; import jakarta.validation.Valid; import jakarta.validation.constraints.NotNull; import lombok.Data; @@ -30,6 +31,7 @@ public class AlarmRule { private String alarmDetails; private DashboardId dashboardId; + @JsonIgnore public boolean requiresScheduledReevaluation() { return condition.hasSchedule(); } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/AlarmCondition.java b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/AlarmCondition.java index c9280151d1..9bb549994b 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/AlarmCondition.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/AlarmCondition.java @@ -45,6 +45,7 @@ public abstract class AlarmCondition { @Valid private AlarmConditionValue schedule; + @JsonIgnore public boolean hasSchedule() { return schedule != null && !(schedule.getStaticValue() instanceof AnyTimeSchedule); } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/AlarmConditionValue.java b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/AlarmConditionValue.java index 84a1498ef6..fab3a78ab3 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/AlarmConditionValue.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/AlarmConditionValue.java @@ -15,6 +15,8 @@ */ package org.thingsboard.server.common.data.alarm.rule.condition; +import com.fasterxml.jackson.annotation.JsonIgnore; +import jakarta.validation.constraints.AssertTrue; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; @@ -27,4 +29,10 @@ public class AlarmConditionValue { private T staticValue; private String dynamicValueArgument; + @JsonIgnore + @AssertTrue(message = "Either staticValue or dynamicValueArgument must be set") + public boolean isValid() { + return staticValue != null ^ dynamicValueArgument != null; + } + } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/AlarmConditionFilter.java b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/AlarmConditionFilter.java index aa70feca13..e9785d675b 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/AlarmConditionFilter.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/AlarmConditionFilter.java @@ -15,7 +15,6 @@ */ package org.thingsboard.server.common.data.alarm.rule.condition.expression; -import io.swagger.v3.oas.annotations.media.Schema; import jakarta.validation.Valid; import jakarta.validation.constraints.NotBlank; import jakarta.validation.constraints.NotNull; @@ -24,7 +23,6 @@ import org.thingsboard.server.common.data.alarm.rule.condition.expression.predic import java.io.Serializable; -@Schema @Data public class AlarmConditionFilter implements Serializable { diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/BooleanFilterPredicate.java b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/BooleanFilterPredicate.java index 8a57aba3e0..94dced5fe4 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/BooleanFilterPredicate.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/BooleanFilterPredicate.java @@ -15,13 +15,18 @@ */ package org.thingsboard.server.common.data.alarm.rule.condition.expression.predicate; +import jakarta.validation.Valid; +import jakarta.validation.constraints.NotNull; import lombok.Data; import org.thingsboard.server.common.data.alarm.rule.condition.AlarmConditionValue; @Data public class BooleanFilterPredicate implements SimpleKeyFilterPredicate { + @NotNull private BooleanOperation operation; + @Valid + @NotNull private AlarmConditionValue value; @Override diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/NumericFilterPredicate.java b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/NumericFilterPredicate.java index 30a82e06bb..65316eda88 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/NumericFilterPredicate.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/NumericFilterPredicate.java @@ -15,13 +15,18 @@ */ package org.thingsboard.server.common.data.alarm.rule.condition.expression.predicate; +import jakarta.validation.Valid; +import jakarta.validation.constraints.NotNull; import lombok.Data; import org.thingsboard.server.common.data.alarm.rule.condition.AlarmConditionValue; @Data public class NumericFilterPredicate implements SimpleKeyFilterPredicate { + @NotNull private NumericOperation operation; + @Valid + @NotNull private AlarmConditionValue value; @Override diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/StringFilterPredicate.java b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/StringFilterPredicate.java index ccc263611f..913c12ca1c 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/StringFilterPredicate.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/StringFilterPredicate.java @@ -15,13 +15,18 @@ */ package org.thingsboard.server.common.data.alarm.rule.condition.expression.predicate; +import jakarta.validation.Valid; +import jakarta.validation.constraints.NotNull; import lombok.Data; import org.thingsboard.server.common.data.alarm.rule.condition.AlarmConditionValue; @Data public class StringFilterPredicate implements SimpleKeyFilterPredicate { + @NotNull private StringOperation operation; + @Valid + @NotNull private AlarmConditionValue value; private boolean ignoreCase; @@ -40,4 +45,5 @@ public class StringFilterPredicate implements SimpleKeyFilterPredicate { IN, NOT_IN } + } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/AlarmCalculatedFieldConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/AlarmCalculatedFieldConfiguration.java index 900973ac1a..d36ba33849 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/AlarmCalculatedFieldConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/AlarmCalculatedFieldConfiguration.java @@ -15,15 +15,24 @@ */ package org.thingsboard.server.common.data.cf.configuration; +import com.fasterxml.jackson.annotation.JsonIgnore; import jakarta.validation.Valid; import jakarta.validation.constraints.NotEmpty; import lombok.Data; +import org.apache.commons.lang3.tuple.Pair; import org.thingsboard.server.common.data.alarm.AlarmSeverity; import org.thingsboard.server.common.data.alarm.rule.AlarmRule; import org.thingsboard.server.common.data.cf.CalculatedFieldType; +import org.thingsboard.server.common.data.util.CollectionsUtil; +import java.util.Comparator; import java.util.List; import java.util.Map; +import java.util.Objects; +import java.util.function.BiPredicate; +import java.util.stream.Stream; + +import static java.util.Map.Entry.comparingByKey; @Data public class AlarmCalculatedFieldConfiguration implements ArgumentsBasedCalculatedFieldConfiguration { @@ -51,15 +60,31 @@ public class AlarmCalculatedFieldConfiguration implements ArgumentsBasedCalculat return null; } + @JsonIgnore @Override - public void validate() { + public boolean requiresScheduledReevaluation() { + return getAllRules().anyMatch(entry -> entry.getValue().requiresScheduledReevaluation()); + } + @JsonIgnore + public Stream> getAllRules() { + Stream> rules = createRules.entrySet().stream() + .map(entry -> Pair.of(entry.getKey(), entry.getValue())); + if (clearRule != null) { + rules = Stream.concat(rules, Stream.of(Pair.of(null, clearRule))); + } + return rules.sorted(comparingByKey(Comparator.nullsLast(Comparator.naturalOrder()))); } - @Override - public boolean requiresScheduledReevaluation() { - return createRules.values().stream().anyMatch(AlarmRule::requiresScheduledReevaluation) || - (clearRule != null && clearRule.requiresScheduledReevaluation()); + public boolean rulesEqual(AlarmCalculatedFieldConfiguration other, BiPredicate equalityCheck) { + List> thisRules = this.getAllRules().toList(); + List> otherRules = other.getAllRules().toList(); + return CollectionsUtil.elementsEqual(thisRules, otherRules, (thisRule, otherRule) -> { + if (!Objects.equals(thisRule.getKey(), otherRule.getKey())) { + return false; + } + return equalityCheck.test(thisRule.getValue(), otherRule.getValue()); + }); } } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/CalculatedFieldConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/CalculatedFieldConfiguration.java index ca0a3c1a54..3df9a32dcc 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/CalculatedFieldConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/CalculatedFieldConfiguration.java @@ -52,7 +52,7 @@ public interface CalculatedFieldConfiguration { Output getOutput(); - void validate(); + default void validate() {} @JsonIgnore default List getReferencedEntities() { @@ -74,6 +74,7 @@ public interface CalculatedFieldConfiguration { .collect(Collectors.toList()); } + @JsonIgnore default boolean requiresScheduledReevaluation() { return false; } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/event/CalculatedFieldDebugEvent.java b/common/data/src/main/java/org/thingsboard/server/common/data/event/CalculatedFieldDebugEvent.java index 0424eabeb6..acc5cf6205 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/event/CalculatedFieldDebugEvent.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/event/CalculatedFieldDebugEvent.java @@ -29,7 +29,7 @@ import org.thingsboard.server.common.data.id.TenantId; import java.util.UUID; -@ToString +@ToString(callSuper = true) @EqualsAndHashCode(callSuper = true) public class CalculatedFieldDebugEvent extends Event { diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/util/CollectionsUtil.java b/common/data/src/main/java/org/thingsboard/server/common/data/util/CollectionsUtil.java index 71c5256203..e92c62242c 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/util/CollectionsUtil.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/util/CollectionsUtil.java @@ -18,9 +18,11 @@ package org.thingsboard.server.common.data.util; import java.util.Collection; import java.util.HashMap; import java.util.HashSet; +import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.function.BiPredicate; import java.util.stream.Collectors; public class CollectionsUtil { @@ -95,4 +97,31 @@ public class CollectionsUtil { return false; } + public static boolean elementsEqual(Iterable iterable1, Iterable iterable2, BiPredicate equalityCheck) { + if (iterable1 instanceof Collection collection1 && iterable2 instanceof Collection collection2) { + if (collection1.size() != collection2.size()) { + return false; + } + } + + Iterator iterator1 = iterable1.iterator(); + Iterator iterator2 = iterable2.iterator(); + while (true) { + if (iterator1.hasNext()) { + if (!iterator2.hasNext()) { + return false; + } + + T o1 = iterator1.next(); + T o2 = iterator2.next(); + if (equalityCheck.test(o1, o2)) { + continue; + } else { + return false; + } + } + return !iterator2.hasNext(); + } + } + } diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index b060b47429..d455a8f0e6 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -1728,12 +1728,10 @@ message ToCalculatedFieldMsg { CalculatedFieldTelemetryMsgProto telemetryMsg = 1; CalculatedFieldLinkedTelemetryMsgProto linkedTelemetryMsg = 2; EntityActionEventProto eventMsg = 3; - repeated string cfTypes = 4; } message ToCalculatedFieldNotificationMsg { CalculatedFieldLinkedTelemetryMsgProto linkedTelemetryMsg = 1; - repeated string cfTypes = 2; } /* Messages that are handled by ThingsBoard RuleEngine Service */ diff --git a/dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmService.java b/dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmService.java index 3df4a64e73..413bf5ffb2 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmService.java @@ -159,7 +159,7 @@ public class BaseAlarmService extends AbstractCachedEntityService