diff --git a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java index f953e57cc3..ad683c324c 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java @@ -34,6 +34,8 @@ public interface CalculatedFieldCache { List getCalculatedFieldLinksByEntityId(TenantId tenantId, EntityId entityId); + void updateCalculatedFieldLinks(TenantId tenantId, CalculatedFieldId calculatedFieldId); + CalculatedFieldCtx getCalculatedFieldCtx(TenantId tenantId, CalculatedFieldId calculatedFieldId, TbelInvokeService tbelInvokeService); Set getEntitiesByProfile(TenantId tenantId, EntityId entityId); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java index dd2ab3857c..a4077b599b 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java @@ -141,6 +141,31 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache { return cfLinks; } + + @Override + public void updateCalculatedFieldLinks(TenantId tenantId, CalculatedFieldId calculatedFieldId) { + log.debug("Update calculated field links per entity for calculated field: [{}]", calculatedFieldId); + calculatedFieldFetchLock.lock(); + try { + List cfLinks = getCalculatedFieldLinks(tenantId, calculatedFieldId); + if (cfLinks != null && !cfLinks.isEmpty()) { + cfLinks.forEach(link -> { + entityIdCalculatedFieldLinks.compute(link.getEntityId(), (id, existingList) -> { + if (existingList == null) { + existingList = new ArrayList<>(); + } else if (!(existingList instanceof ArrayList)) { + existingList = new ArrayList<>(existingList); + } + existingList.add(link); + return existingList; + }); + }); + } + } finally { + calculatedFieldFetchLock.unlock(); + } + } + @Override public CalculatedFieldCtx getCalculatedFieldCtx(TenantId tenantId, CalculatedFieldId calculatedFieldId, TbelInvokeService tbelInvokeService) { CalculatedFieldCtx ctx = calculatedFieldsCtx.get(calculatedFieldId); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java index 976139e8bd..f7bcb4e678 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java @@ -35,7 +35,6 @@ import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.script.api.tbel.TbelInvokeService; import org.thingsboard.server.cluster.TbClusterService; -import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.cf.CalculatedField; import org.thingsboard.server.common.data.cf.CalculatedFieldLink; @@ -71,7 +70,6 @@ import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.cf.CalculatedFieldService; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtx; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; @@ -99,13 +97,11 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.function.Consumer; import java.util.stream.Collectors; -import java.util.stream.Stream; import static org.thingsboard.server.common.data.DataConstants.SCOPE; import static org.thingsboard.server.common.util.ProtoUtils.fromObjectProto; import static org.thingsboard.server.common.util.ProtoUtils.toObjectProto; -@TbCoreComponent @Service @Slf4j @RequiredArgsConstructor @@ -170,35 +166,38 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas protected Map>> onAddedPartitions(Set addedPartitions) { var result = new HashMap>>(); PageDataIterable cfs = new PageDataIterable<>(calculatedFieldService::findAllCalculatedFields, initFetchPackSize); - Map> tpiCalculatedFieldMap = new HashMap<>(); + Map> tpiTargetEntityMap = new HashMap<>(); for (CalculatedField cf : cfs) { - TopicPartitionInfo tpi; - try { - tpi = partitionService.resolve(ServiceType.TB_CORE, cf.getTenantId(), cf.getId()); - } catch (Exception e) { - log.warn("Failed to resolve partition for CalculatedField [{}], tenant id [{}]. Reason: {}", - cf.getId(), cf.getTenantId(), e.getMessage()); - continue; - } - if (addedPartitions.contains(tpi) && states.keySet().stream().noneMatch(ctxId -> ctxId.cfId().equals(cf.getId().getId()))) { - tpiCalculatedFieldMap.computeIfAbsent(tpi, k -> new ArrayList<>()).add(cf); + + Consumer resolvePartition = entityId -> { + TopicPartitionInfo tpi; + try { + tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, cf.getTenantId(), entityId); + if (addedPartitions.contains(tpi) && states.keySet().stream().noneMatch(ctxId -> ctxId.cfId().equals(cf.getId().getId()))) { + tpiTargetEntityMap.computeIfAbsent(tpi, k -> new ArrayList<>()).add(new CalculatedFieldEntityCtxId(cf.getId().getId(), entityId.getId())); + } + } catch (Exception e) { + log.warn("Failed to resolve partition for CalculatedFieldEntityCtxId: entityId=[{}], tenantId=[{}]. Reason: {}", + entityId, cf.getTenantId(), e.getMessage()); + } + }; + + EntityId cfEntityId = cf.getEntityId(); + if (isProfileEntity(cfEntityId)) { + calculatedFieldCache.getEntitiesByProfile(cf.getTenantId(), cfEntityId).forEach(resolvePartition); + } else { + resolvePartition.accept(cfEntityId); } } - for (var entry : tpiCalculatedFieldMap.entrySet()) { - for (List partition : Lists.partition(entry.getValue(), 1000)) { + for (var entry : tpiTargetEntityMap.entrySet()) { + for (List partition : Lists.partition(entry.getValue(), 1000)) { log.info("[{}] Submit task for CalculatedFields: {}", entry.getKey(), partition.size()); var future = calculatedFieldExecutor.submit(() -> { try { - for (CalculatedField cf : partition) { - EntityId cfEntityId = cf.getEntityId(); - if (isProfileEntity(cfEntityId)) { - calculatedFieldCache.getEntitiesByProfile(cf.getTenantId(), cfEntityId) - .forEach(entityId -> restoreState(cf, entityId)); - } else { - restoreState(cf, cfEntityId); - } + for (CalculatedFieldEntityCtxId ctxId : partition) { + restoreState(ctxId.cfId(), ctxId.entityId()); } } catch (Throwable t) { log.error("Unexpected exception while restoring CalculatedField states", t); @@ -211,16 +210,16 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas return result; } - private void restoreState(CalculatedField cf, EntityId entityId) { - CalculatedFieldEntityCtxId ctxId = new CalculatedFieldEntityCtxId(cf.getId().getId(), entityId.getId()); + private void restoreState(UUID calculatedFieldId, UUID entityId) { + CalculatedFieldEntityCtxId ctxId = new CalculatedFieldEntityCtxId(calculatedFieldId, entityId); String storedState = rocksDBService.get(JacksonUtil.writeValueAsString(ctxId)); if (storedState != null) { CalculatedFieldEntityCtx restoredCtx = JacksonUtil.fromString(storedState, CalculatedFieldEntityCtx.class); states.put(ctxId, restoredCtx); - log.info("Restored state for CalculatedField [{}]", cf.getId()); + log.info("Restored state for CalculatedField [{}]", calculatedFieldId); } else { - log.warn("No state found for CalculatedField [{}], entity [{}].", cf.getId(), entityId); + log.warn("No state found for CalculatedField [{}], entity [{}].", calculatedFieldId, entityId); } } @@ -239,12 +238,6 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas TenantId tenantId = TenantId.fromUUID(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())); CalculatedFieldId calculatedFieldId = new CalculatedFieldId(new UUID(proto.getCalculatedFieldIdMSB(), proto.getCalculatedFieldIdLSB())); log.info("Received CalculatedFieldMsgProto for processing: tenantId=[{}], calculatedFieldId=[{}]", tenantId, calculatedFieldId); - TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, calculatedFieldId); - if (!tpi.isMyPartition()) { - clusterService.pushMsgToCore(tenantId, calculatedFieldId, TransportProtos.ToCoreMsg.newBuilder().setCalculatedFieldMsg(proto).build(), null); - log.debug("[{}][{}] Calculated field belongs to external partition. Probably rebalancing is in progress. Topic: {}", tenantId, calculatedFieldId, tpi.getFullTopicName()); - callback.onFailure(new RuntimeException("Calculated field belongs to external partition " + tpi.getFullTopicName() + "!")); - } if (proto.getDeleted()) { log.warn("Executing onCalculatedFieldDelete, calculatedFieldId=[{}]", calculatedFieldId); onCalculatedFieldDelete(tenantId, calculatedFieldId, callback); @@ -308,12 +301,6 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas private void onCalculatedFieldDelete(TenantId tenantId, CalculatedFieldId calculatedFieldId, TbCallback callback) { try { cleanupEntity(calculatedFieldId); - TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, calculatedFieldId); - Set calculatedFieldIds = partitionedEntities.get(tpi); - if (calculatedFieldIds != null) { - calculatedFieldIds.remove(calculatedFieldId); - } - calculatedFieldCache.evict(calculatedFieldId); states.keySet().removeIf(ctxId -> ctxId.cfId().equals(calculatedFieldId.getId())); List statesToRemove = states.keySet().stream() .filter(ctxId -> ctxId.cfId().equals(calculatedFieldId.getId())) @@ -346,22 +333,14 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas try { TenantId tenantId = calculatedFieldTelemetryUpdateRequest.getTenantId(); EntityId entityId = calculatedFieldTelemetryUpdateRequest.getEntityId(); - AttributeScope scope = calculatedFieldTelemetryUpdateRequest.getScope(); - List telemetry = calculatedFieldTelemetryUpdateRequest.getKvEntries(); - List calculatedFieldIds = calculatedFieldTelemetryUpdateRequest.getCalculatedFieldIds(); if (supportedReferencedEntities.contains(entityId.getEntityType())) { EntityId profileId = getProfileId(tenantId, entityId); - List cfLinks = Stream.concat( - calculatedFieldCache.getCalculatedFieldLinksByEntityId(tenantId, entityId).stream(), - profileId != null ? calculatedFieldCache.getCalculatedFieldLinksByEntityId(tenantId, profileId).stream() : Stream.empty() - ).toList(); - - cfLinks.forEach(link -> { + getCalculatedFieldLinks(tenantId, entityId, profileId).forEach(link -> { CalculatedFieldId calculatedFieldId = link.getCalculatedFieldId(); - Map telemetryKeys = getTelemetryKeysFromLink(link, scope); - Map updatedTelemetry = telemetry.stream() + Map telemetryKeys = calculatedFieldTelemetryUpdateRequest.getTelemetryKeysFromLink(link); + Map updatedTelemetry = calculatedFieldTelemetryUpdateRequest.getKvEntries().stream() .filter(entry -> telemetryKeys.containsValue(entry.getKey())) .collect(Collectors.toMap( entry -> getMappedKey(entry, telemetryKeys), @@ -370,7 +349,8 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas )); if (!updatedTelemetry.isEmpty()) { - executeTelemetryUpdate(tenantId, entityId, calculatedFieldId, calculatedFieldIds, updatedTelemetry); + List previousCalculatedFieldIds = calculatedFieldTelemetryUpdateRequest.getPreviousCalculatedFieldIds(); + executeTelemetryUpdate(tenantId, entityId, calculatedFieldId, previousCalculatedFieldIds, updatedTelemetry); } }); } @@ -379,14 +359,6 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas } } - private Map getTelemetryKeysFromLink(CalculatedFieldLink link, AttributeScope scope) { - return scope == null ? link.getConfiguration().getTimeSeries() : switch (scope) { - case CLIENT_SCOPE -> link.getConfiguration().getClientAttributes(); - case SERVER_SCOPE -> link.getConfiguration().getServerAttributes(); - case SHARED_SCOPE -> link.getConfiguration().getSharedAttributes(); - }; - } - private String getMappedKey(KvEntry entry, Map telemetry) { return telemetry.entrySet().stream() .filter(kvEntry -> kvEntry.getValue().equals(entry.getKey())) @@ -395,7 +367,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas .orElse(entry.getKey()); } - private void executeTelemetryUpdate(TenantId tenantId, EntityId entityId, CalculatedFieldId calculatedFieldId, List calculatedFieldIds, Map updatedTelemetry) { + private void executeTelemetryUpdate(TenantId tenantId, EntityId entityId, CalculatedFieldId calculatedFieldId, List previousCalculatedFieldIds, Map updatedTelemetry) { log.info("Received telemetry update msg: tenantId=[{}], entityId=[{}], calculatedFieldId=[{}]", tenantId, entityId, calculatedFieldId); CalculatedField calculatedField = calculatedFieldCache.getCalculatedField(tenantId, calculatedFieldId); CalculatedFieldCtx calculatedFieldCtx = calculatedFieldCache.getCalculatedFieldCtx(tenantId, calculatedFieldId, tbelInvokeService); @@ -407,14 +379,14 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas case ASSET_PROFILE, DEVICE_PROFILE -> { boolean isCommonEntity = calculatedField.getConfiguration().getReferencedEntities().contains(entityId); if (isCommonEntity) { - calculatedFieldCache.getEntitiesByProfile(tenantId, cfEntityId).forEach(id -> updateOrInitializeState(calculatedFieldCtx, id, argumentValues, calculatedFieldIds)); + calculatedFieldCache.getEntitiesByProfile(tenantId, cfEntityId).forEach(id -> updateOrInitializeState(calculatedFieldCtx, id, argumentValues, previousCalculatedFieldIds)); } else { - updateOrInitializeState(calculatedFieldCtx, entityId, argumentValues, calculatedFieldIds); + updateOrInitializeState(calculatedFieldCtx, entityId, argumentValues, previousCalculatedFieldIds); } } - default -> updateOrInitializeState(calculatedFieldCtx, cfEntityId, argumentValues, calculatedFieldIds); + default -> + updateOrInitializeState(calculatedFieldCtx, cfEntityId, argumentValues, previousCalculatedFieldIds); } - log.info("Successfully updated telemetry for calculatedFieldId: [{}]", calculatedFieldId); } @Override @@ -423,20 +395,20 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas TenantId tenantId = TenantId.fromUUID(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())); CalculatedFieldId calculatedFieldId = new CalculatedFieldId(new UUID(proto.getCalculatedFieldIdMSB(), proto.getCalculatedFieldIdLSB())); EntityId entityId = EntityIdFactory.getByTypeAndUuid(proto.getEntityType(), new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB())); - + log.info("Received CalculatedFieldStateMsgProto for processing: tenantId=[{}], calculatedFieldId=[{}], entityId=[{}]", tenantId, calculatedFieldId, entityId); if (proto.getClear()) { clearState(tenantId, calculatedFieldId, entityId); return; } - List calculatedFieldIds = proto.getCalculatedFieldsList().stream() + List previousCalculatedFieldIds = proto.getPreviousCalculatedFieldsList().stream() .map(cfIdProto -> new CalculatedFieldId(new UUID(cfIdProto.getCalculatedFieldIdMSB(), cfIdProto.getCalculatedFieldIdLSB()))) - .toList(); + .collect(Collectors.toCollection(ArrayList::new)); Map argumentsMap = proto.getArgumentsMap().entrySet().stream() .collect(Collectors.toMap(Map.Entry::getKey, entry -> fromArgumentEntryProto(entry.getValue()))); CalculatedFieldCtx calculatedFieldCtx = calculatedFieldCache.getCalculatedFieldCtx(tenantId, calculatedFieldId, tbelInvokeService); - updateOrInitializeState(calculatedFieldCtx, entityId, argumentsMap, calculatedFieldIds); + updateOrInitializeState(calculatedFieldCtx, entityId, argumentsMap, previousCalculatedFieldIds); } catch (Exception e) { log.trace("Failed to process calculated field update state msg: [{}]", proto, e); } @@ -451,9 +423,6 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas EntityId newProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getNewProfileIdMSB(), proto.getNewProfileIdLSB())); log.info("Received EntityProfileUpdateMsgProto for processing: tenantId=[{}], entityId=[{}]", tenantId, entityId); - calculatedFieldCache.getEntitiesByProfile(tenantId, oldProfileId).remove(entityId); - calculatedFieldCache.getEntitiesByProfile(tenantId, newProfileId).add(entityId); - calculatedFieldService.findCalculatedFieldIdsByEntityId(tenantId, oldProfileId) .forEach(cfId -> clearState(tenantId, cfId, entityId)); @@ -472,15 +441,11 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas log.info("Received ProfileEntityMsgProto for processing: tenantId=[{}], entityId=[{}]", tenantId, entityId); if (proto.getDeleted()) { log.info("Executing profile entity deleted msg, tenantId=[{}], entityId=[{}]", tenantId, entityId); - calculatedFieldCache.getEntitiesByProfile(tenantId, profileId).remove(entityId); - List calculatedFieldIds = Stream.concat( - calculatedFieldCache.getCalculatedFieldLinksByEntityId(tenantId, entityId).stream().map(CalculatedFieldLink::getCalculatedFieldId), - calculatedFieldCache.getCalculatedFieldLinksByEntityId(tenantId, profileId).stream().map(CalculatedFieldLink::getCalculatedFieldId) - ).toList(); - calculatedFieldIds.forEach(cfId -> clearState(tenantId, cfId, entityId)); + + getCalculatedFieldLinks(tenantId, entityId, profileId) + .forEach(link -> clearState(tenantId, link.getCalculatedFieldId(), entityId)); } else { log.info("Executing profile entity added msg, tenantId=[{}], entityId=[{}]", tenantId, entityId); - calculatedFieldCache.getEntitiesByProfile(tenantId, profileId).add(entityId); initializeStateForEntityByProfile(tenantId, entityId, profileId, callback); } } catch (Exception e) { @@ -489,7 +454,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas } private void clearState(TenantId tenantId, CalculatedFieldId calculatedFieldId, EntityId entityId) { - TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, calculatedFieldId); + TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tenantId, entityId); if (tpi.isMyPartition()) { log.warn("Executing clearState, calculatedFieldId=[{}], entityId=[{}]", calculatedFieldId, entityId); CalculatedFieldEntityCtxId ctxId = new CalculatedFieldEntityCtxId(calculatedFieldId.getId(), entityId.getId()); @@ -540,12 +505,12 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas }, calculatedFieldCallbackExecutor); } - private void updateOrInitializeState(CalculatedFieldCtx calculatedFieldCtx, EntityId entityId, Map argumentValues, List calculatedFieldIds) { + private void updateOrInitializeState(CalculatedFieldCtx calculatedFieldCtx, EntityId entityId, Map argumentValues, List previousCalculatedFieldIds) { TenantId tenantId = calculatedFieldCtx.getTenantId(); CalculatedFieldId cfId = calculatedFieldCtx.getCfId(); Map argumentsMap = new HashMap<>(argumentValues); - TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, cfId); + TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tenantId, entityId); if (tpi.isMyPartition()) { CalculatedFieldEntityCtxId entityCtxId = new CalculatedFieldEntityCtxId(cfId.getId(), entityId.getId()); @@ -563,8 +528,9 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas boolean allArgsPresent = arguments.keySet().containsAll(calculatedFieldCtx.getArguments().keySet()) && !arguments.containsValue(SingleValueArgumentEntry.EMPTY) && !arguments.containsValue(TsRollingArgumentEntry.EMPTY); if (allArgsPresent) { - performCalculation(calculatedFieldCtx, state, entityId, calculatedFieldIds); + performCalculation(calculatedFieldCtx, state, entityId, previousCalculatedFieldIds); } + log.info("Successfully updated state: calculatedFieldId=[{}], entityId=[{}]", calculatedFieldCtx.getCfId(), entityId); } updateFuture.complete(null); }; @@ -597,17 +563,17 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas return calculatedFieldEntityCtx; }); } else { - sendUpdateCalculatedFieldStateMsg(tenantId, cfId, entityId, calculatedFieldIds, argumentsMap); + sendUpdateCalculatedFieldStateMsg(tenantId, cfId, entityId, previousCalculatedFieldIds, argumentsMap); } } - private void performCalculation(CalculatedFieldCtx calculatedFieldCtx, CalculatedFieldState state, EntityId entityId, List calculatedFieldIds) { + private void performCalculation(CalculatedFieldCtx calculatedFieldCtx, CalculatedFieldState state, EntityId entityId, List previousCalculatedFieldIds) { ListenableFuture resultFuture = state.performCalculation(calculatedFieldCtx); Futures.addCallback(resultFuture, new FutureCallback<>() { @Override public void onSuccess(CalculatedFieldResult result) { if (result != null) { - pushMsgToRuleEngine(calculatedFieldCtx.getTenantId(), calculatedFieldCtx.getCfId(), entityId, result, calculatedFieldIds); + pushMsgToRuleEngine(calculatedFieldCtx.getTenantId(), calculatedFieldCtx.getCfId(), entityId, result, previousCalculatedFieldIds); } } @@ -618,26 +584,35 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas }, MoreExecutors.directExecutor()); } - private void pushMsgToRuleEngine(TenantId tenantId, CalculatedFieldId calculatedFieldId, EntityId originatorId, CalculatedFieldResult calculatedFieldResult, List calculatedFieldIds) { + private void pushMsgToRuleEngine(TenantId tenantId, CalculatedFieldId calculatedFieldId, EntityId originatorId, CalculatedFieldResult calculatedFieldResult, List previousCalculatedFieldIds) { try { OutputType type = calculatedFieldResult.getType(); TbMsgType msgType = OutputType.ATTRIBUTES.equals(type) ? TbMsgType.POST_ATTRIBUTES_REQUEST : TbMsgType.POST_TELEMETRY_REQUEST; TbMsgMetaData md = OutputType.ATTRIBUTES.equals(type) ? new TbMsgMetaData(Map.of(SCOPE, calculatedFieldResult.getScope().name())) : TbMsgMetaData.EMPTY; ObjectNode payload = createJsonPayload(calculatedFieldResult); - if (calculatedFieldIds == null) { - calculatedFieldIds = new ArrayList<>(); - } - if (calculatedFieldIds.contains(calculatedFieldId)) { + if (previousCalculatedFieldIds != null && previousCalculatedFieldIds.contains(calculatedFieldId)) { throw new IllegalArgumentException("Calculated field [" + calculatedFieldId.getId() + "] refers to itself, causing an infinite loop."); } + List calculatedFieldIds = previousCalculatedFieldIds != null + ? new ArrayList<>(previousCalculatedFieldIds) + : new ArrayList<>(); calculatedFieldIds.add(calculatedFieldId); - TbMsg msg = TbMsg.newMsg().type(msgType).originator(originatorId).calculatedFieldIds(calculatedFieldIds).metaData(md).data(JacksonUtil.writeValueAsString(payload)).build(); + TbMsg msg = TbMsg.newMsg().type(msgType).originator(originatorId).previousCalculatedFieldIds(calculatedFieldIds).metaData(md).data(JacksonUtil.writeValueAsString(payload)).build(); clusterService.pushMsgToRuleEngine(tenantId, originatorId, msg, null); + log.info("Pushed message to rule engine: originatorId=[{}]", originatorId); } catch (Exception e) { log.warn("[{}] Failed to push message to rule engine. CalculatedFieldResult: {}", originatorId, calculatedFieldResult, e); } } + private List getCalculatedFieldLinks(TenantId tenantId, EntityId entityId, EntityId profileId) { + List links = new ArrayList<>(calculatedFieldCache.getCalculatedFieldLinksByEntityId(tenantId, entityId)); + if (profileId != null) { + links.addAll(calculatedFieldCache.getCalculatedFieldLinksByEntityId(tenantId, profileId)); + } + return links; + } + private ListenableFuture fetchArguments(TenantId tenantId, EntityId entityId, Map necessaryArguments, Consumer> onComplete) { Map argumentValues = new HashMap<>(); List> futures = new ArrayList<>(); @@ -701,13 +676,13 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas return Futures.transform(tsRollingFuture, tsRolling -> tsRolling == null ? TsRollingArgumentEntry.EMPTY : ArgumentEntry.createTsRollingArgument(tsRolling), calculatedFieldCallbackExecutor); } - private void sendUpdateCalculatedFieldStateMsg(TenantId tenantId, CalculatedFieldId calculatedFieldId, EntityId entityId, List calculatedFieldIds, Map argumentValues) { + private void sendUpdateCalculatedFieldStateMsg(TenantId tenantId, CalculatedFieldId calculatedFieldId, EntityId entityId, List previousCalculatedFieldIds, Map argumentValues) { TransportProtos.CalculatedFieldStateMsgProto.Builder msgBuilder = createBaseCalculatedFieldStateMsg(tenantId, calculatedFieldId, entityId); if (argumentValues != null) { argumentValues.forEach((key, argumentEntry) -> msgBuilder.putArguments(key, toArgumentEntryProto(argumentEntry))); } - if (calculatedFieldIds != null) { - calculatedFieldIds.forEach(cfId -> msgBuilder.addCalculatedFields( + if (previousCalculatedFieldIds != null) { + previousCalculatedFieldIds.forEach(cfId -> msgBuilder.addPreviousCalculatedFields( TransportProtos.CalculatedFieldIdProto.newBuilder() .setCalculatedFieldIdMSB(cfId.getId().getMostSignificantBits()) .setCalculatedFieldIdLSB(cfId.getId().getLeastSignificantBits()) @@ -715,6 +690,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas )); } + log.info("Sending calculated field state msg from entityId [{}]", entityId); clusterService.pushMsgToCore(tenantId, calculatedFieldId, TransportProtos.ToCoreMsg.newBuilder().setCalculatedFieldStateMsg(msgBuilder).build(), null); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldAttributeUpdateRequest.java b/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldAttributeUpdateRequest.java index 8479ff37d7..c56217b2ce 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldAttributeUpdateRequest.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldAttributeUpdateRequest.java @@ -18,12 +18,14 @@ package org.thingsboard.server.service.cf.telemetry; import lombok.AllArgsConstructor; import lombok.Data; import org.thingsboard.server.common.data.AttributeScope; +import org.thingsboard.server.common.data.cf.CalculatedFieldLink; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.AttributeKvEntry; import java.util.List; +import java.util.Map; @Data @AllArgsConstructor @@ -33,6 +35,15 @@ public class CalculatedFieldAttributeUpdateRequest implements CalculatedFieldTel private EntityId entityId; private AttributeScope scope; private List kvEntries; - private List calculatedFieldIds; + private List previousCalculatedFieldIds; + + @Override + public Map getTelemetryKeysFromLink(CalculatedFieldLink link) { + return switch (scope) { + case CLIENT_SCOPE -> link.getConfiguration().getClientAttributes(); + case SERVER_SCOPE -> link.getConfiguration().getServerAttributes(); + case SHARED_SCOPE -> link.getConfiguration().getSharedAttributes(); + }; + } } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldTelemetryUpdateRequest.java b/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldTelemetryUpdateRequest.java index 3c28833f31..98062a08db 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldTelemetryUpdateRequest.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldTelemetryUpdateRequest.java @@ -15,13 +15,14 @@ */ package org.thingsboard.server.service.cf.telemetry; -import org.thingsboard.server.common.data.AttributeScope; +import org.thingsboard.server.common.data.cf.CalculatedFieldLink; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.KvEntry; import java.util.List; +import java.util.Map; public interface CalculatedFieldTelemetryUpdateRequest { @@ -29,10 +30,10 @@ public interface CalculatedFieldTelemetryUpdateRequest { EntityId getEntityId(); - AttributeScope getScope(); - List getKvEntries(); - List getCalculatedFieldIds(); + List getPreviousCalculatedFieldIds(); + + Map getTelemetryKeysFromLink(CalculatedFieldLink link); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldTimeSeriesUpdateRequest.java b/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldTimeSeriesUpdateRequest.java index 987d899465..bd2161dca1 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldTimeSeriesUpdateRequest.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldTimeSeriesUpdateRequest.java @@ -17,13 +17,14 @@ package org.thingsboard.server.service.cf.telemetry; import lombok.AllArgsConstructor; import lombok.Data; -import org.thingsboard.server.common.data.AttributeScope; +import org.thingsboard.server.common.data.cf.CalculatedFieldLink; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.TsKvEntry; import java.util.List; +import java.util.Map; @Data @AllArgsConstructor @@ -32,11 +33,11 @@ public class CalculatedFieldTimeSeriesUpdateRequest implements CalculatedFieldTe private TenantId tenantId; private EntityId entityId; private List kvEntries; - private List calculatedFieldIds; + private List previousCalculatedFieldIds; @Override - public AttributeScope getScope() { - return null; + public Map getTelemetryKeysFromLink(CalculatedFieldLink link) { + return link.getConfiguration().getTimeSeries(); } } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java index 6316d1fdb2..551b0ddbc2 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java @@ -391,7 +391,7 @@ public class DefaultTbClusterService implements TbClusterService { public void onDeviceDeleted(TenantId tenantId, Device device, TbQueueCallback callback) { DeviceId deviceId = device.getId(); gatewayNotificationsService.onDeviceDeleted(device); - sendProfileEntityEvent(tenantId, deviceId, device.getDeviceProfileId(), false, true); + handleProfileEntityEvent(tenantId, deviceId, device.getDeviceProfileId(), false, true); broadcastEntityDeleteToTransport(tenantId, deviceId, device.getName(), callback); sendDeviceStateServiceEvent(tenantId, deviceId, false, false, true); broadcastEntityStateChangeEvent(tenantId, deviceId, ComponentLifecycleEvent.DELETED); @@ -400,7 +400,7 @@ public class DefaultTbClusterService implements TbClusterService { @Override public void onAssetDeleted(TenantId tenantId, Asset asset, TbQueueCallback callback) { AssetId assetId = asset.getId(); - sendProfileEntityEvent(tenantId, assetId, asset.getAssetProfileId(), false, true); + handleProfileEntityEvent(tenantId, assetId, asset.getAssetProfileId(), true, true); broadcastEntityStateChangeEvent(tenantId, assetId, ComponentLifecycleEvent.DELETED); } @@ -563,7 +563,9 @@ public class DefaultTbClusterService implements TbClusterService { || entityType.equals(EntityType.API_USAGE_STATE) || (entityType.equals(EntityType.DEVICE) && msg.getEvent() == ComponentLifecycleEvent.UPDATED) || entityType.equals(EntityType.ENTITY_VIEW) - || entityType.equals(EntityType.NOTIFICATION_RULE)) { + || entityType.equals(EntityType.NOTIFICATION_RULE) + || entityType.equals(EntityType.CALCULATED_FIELD) + ) { TbQueueProducer> toCoreNfProducer = producerProvider.getTbCoreNotificationsMsgProducer(); Set tbCoreServices = partitionService.getAllServiceIds(ServiceType.TB_CORE); for (String serviceId : tbCoreServices) { @@ -624,13 +626,13 @@ public class DefaultTbClusterService implements TbClusterService { } boolean deviceTypeChanged = !device.getType().equals(old.getType()); if (deviceTypeChanged) { - sendEntityProfileUpdatedEvent(device.getTenantId(), device.getId(), old.getDeviceProfileId(), device.getDeviceProfileId()); + handleEntityProfileUpdatedEvent(device.getTenantId(), device.getId(), old.getDeviceProfileId(), device.getDeviceProfileId()); } if (deviceNameChanged || deviceTypeChanged) { pushMsgToCore(new DeviceNameOrTypeUpdateMsg(device.getTenantId(), device.getId(), device.getName(), device.getType()), null); } } else { - sendProfileEntityEvent(device.getTenantId(), device.getId(), device.getDeviceProfileId(), true, false); + handleProfileEntityEvent(device.getTenantId(), device.getId(), device.getDeviceProfileId(), true, false); } broadcastEntityStateChangeEvent(device.getTenantId(), device.getId(), created ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED); sendDeviceStateServiceEvent(device.getTenantId(), device.getId(), created, !created, false); @@ -644,10 +646,10 @@ public class DefaultTbClusterService implements TbClusterService { if (old != null) { boolean assetTypeChanged = !asset.getType().equals(old.getType()); if (assetTypeChanged) { - sendEntityProfileUpdatedEvent(asset.getTenantId(), asset.getId(), old.getAssetProfileId(), asset.getAssetProfileId()); + handleEntityProfileUpdatedEvent(asset.getTenantId(), asset.getId(), old.getAssetProfileId(), asset.getAssetProfileId()); } } else { - sendProfileEntityEvent(asset.getTenantId(), asset.getId(), asset.getAssetProfileId(), true, false); + handleProfileEntityEvent(asset.getTenantId(), asset.getId(), asset.getAssetProfileId(), true, false); } broadcastEntityStateChangeEvent(asset.getTenantId(), asset.getId(), created ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED); } @@ -792,8 +794,8 @@ public class DefaultTbClusterService implements TbClusterService { public void onCalculatedFieldDeleted(TenantId tenantId, CalculatedField calculatedField, TbQueueCallback callback) { CalculatedFieldId calculatedFieldId = calculatedField.getId(); broadcastEntityDeleteToTransport(tenantId, calculatedFieldId, calculatedField.getName(), callback); - sendCalculatedFieldEvent(tenantId, calculatedFieldId, false, false, true); broadcastEntityStateChangeEvent(tenantId, calculatedFieldId, ComponentLifecycleEvent.DELETED); + sendCalculatedFieldEvent(tenantId, calculatedFieldId, false, false, true); } private void sendCalculatedFieldEvent(TenantId tenantId, CalculatedFieldId calculatedFieldId, boolean added, boolean updated, boolean deleted) { @@ -809,7 +811,7 @@ public class DefaultTbClusterService implements TbClusterService { pushMsgToCore(tenantId, calculatedFieldId, ToCoreMsg.newBuilder().setCalculatedFieldMsg(msg).build(), null); } - private void sendEntityProfileUpdatedEvent(TenantId tenantId, EntityId entityId, EntityId oldProfileId, EntityId newProfileId) { + private void handleEntityProfileUpdatedEvent(TenantId tenantId, EntityId entityId, EntityId oldProfileId, EntityId newProfileId) { TransportProtos.EntityProfileUpdateMsgProto.Builder builder = TransportProtos.EntityProfileUpdateMsgProto.newBuilder(); builder.setTenantIdMSB(tenantId.getId().getMostSignificantBits()); builder.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()); @@ -822,10 +824,12 @@ public class DefaultTbClusterService implements TbClusterService { builder.setNewProfileIdMSB(newProfileId.getId().getMostSignificantBits()); builder.setNewProfileIdLSB(newProfileId.getId().getLeastSignificantBits()); TransportProtos.EntityProfileUpdateMsgProto msg = builder.build(); + + broadcastToCore(ToCoreNotificationMsg.newBuilder().setEntityProfileUpdateMsg(msg).build()); pushMsgToCore(tenantId, entityId, ToCoreMsg.newBuilder().setEntityProfileUpdateMsg(msg).build(), null); } - private void sendProfileEntityEvent(TenantId tenantId, EntityId entityId, EntityId profileId, boolean added, boolean deleted) { + private void handleProfileEntityEvent(TenantId tenantId, EntityId entityId, EntityId profileId, boolean added, boolean deleted) { TransportProtos.ProfileEntityMsgProto.Builder builder = TransportProtos.ProfileEntityMsgProto.newBuilder(); builder.setTenantIdMSB(tenantId.getId().getMostSignificantBits()); builder.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()); @@ -838,6 +842,8 @@ public class DefaultTbClusterService implements TbClusterService { builder.setAdded(added); builder.setDeleted(deleted); TransportProtos.ProfileEntityMsgProto msg = builder.build(); + + broadcastToCore(ToCoreNotificationMsg.newBuilder().setProfileEntityMsg(msg).build()); pushMsgToCore(tenantId, entityId, ToCoreMsg.newBuilder().setProfileEntityMsg(msg).build(), null); } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index 1be25b308f..6baa75b3ef 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -40,6 +40,7 @@ import org.thingsboard.server.common.data.event.Event; import org.thingsboard.server.common.data.event.LifecycleEvent; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.id.NotificationRequestId; import org.thingsboard.server.common.data.id.TenantId; @@ -87,6 +88,7 @@ import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; import org.thingsboard.server.queue.provider.TbCoreQueueFactory; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.apiusage.TbApiUsageStateService; +import org.thingsboard.server.service.cf.CalculatedFieldCache; import org.thingsboard.server.service.cf.CalculatedFieldExecutionService; import org.thingsboard.server.service.notification.NotificationSchedulerService; import org.thingsboard.server.service.ota.OtaPackageStateService; @@ -109,6 +111,7 @@ import org.thingsboard.server.service.ws.notification.sub.NotificationRequestUpd import org.thingsboard.server.service.ws.notification.sub.NotificationUpdate; import java.util.List; +import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -181,8 +184,9 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService entitiesByProfile = calculatedFieldCache.getEntitiesByProfile(tenantId, profileId); + if (added) { + entitiesByProfile.add(entityId); + } else { + entitiesByProfile.remove(entityId); + } + } + private void forwardToSubMgrService(SubscriptionMgrMsgProto msg, TbCallback callback) { if (msg.hasSubEvent()) { TbEntitySubEventProto subEvent = msg.getSubEvent(); @@ -688,12 +718,12 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService future = calculatedFieldsExecutor.submit(() -> calculatedFieldExecutionService.onEntityProfileChangedMsg(profileUpdateMsg, callback)); DonAsynchron.withCallback(future, __ -> callback.onSuccess(), t -> { - log.warn("[{}] Failed to process device type updated message for device [{}]", tenantId.getId(), entityId.getId(), t); + log.warn("[{}] Failed to process entity profile updated message for entity [{}]", tenantId.getId(), entityId.getId(), t); callback.onFailure(t); }); } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java index f219d7ae69..35c7894f2f 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java @@ -91,7 +91,7 @@ public class DefaultTbEdgeConsumerService extends AbstractConsumerService future = edgeCtx.getTenantProfileProcessor().processEntityNotification(tenantId, edgeNotificationMsg); case NOTIFICATION_RULE, NOTIFICATION_TARGET, NOTIFICATION_TEMPLATE -> future = edgeCtx.getNotificationEdgeProcessor().processEntityNotification(tenantId, edgeNotificationMsg); - case TB_RESOURCE -> future = edgeCtx.getResourceProcessor().processEntityNotification(tenantId, edgeNotificationMsg); - case DOMAIN, OAUTH2_CLIENT -> future = edgeCtx.getOAuth2EdgeProcessor().processEntityNotification(tenantId, edgeNotificationMsg); + case TB_RESOURCE -> + future = edgeCtx.getResourceProcessor().processEntityNotification(tenantId, edgeNotificationMsg); + case DOMAIN, OAUTH2_CLIENT -> + future = edgeCtx.getOAuth2EdgeProcessor().processEntityNotification(tenantId, edgeNotificationMsg); default -> { future = Futures.immediateFuture(null); log.warn("[{}] Edge event type [{}] is not designed to be pushed to edge", tenantId, type); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java index 73810949b3..7d4d975cb4 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java @@ -46,6 +46,7 @@ import org.thingsboard.server.queue.discovery.QueueKey; import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; import org.thingsboard.server.queue.util.TbRuleEngineComponent; import org.thingsboard.server.service.apiusage.TbApiUsageStateService; +import org.thingsboard.server.service.cf.CalculatedFieldCache; import org.thingsboard.server.service.profile.TbAssetProfileCache; import org.thingsboard.server.service.profile.TbDeviceProfileCache; import org.thingsboard.server.service.queue.processing.AbstractConsumerService; @@ -83,8 +84,9 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< TbApiUsageStateService apiUsageStateService, PartitionService partitionService, ApplicationEventPublisher eventPublisher, - JwtSettingsService jwtSettingsService) { - super(actorContext, tenantProfileCache, deviceProfileCache, assetProfileCache, apiUsageStateService, partitionService, eventPublisher, jwtSettingsService); + JwtSettingsService jwtSettingsService, + CalculatedFieldCache calculatedFieldCache) { + super(actorContext, tenantProfileCache, deviceProfileCache, assetProfileCache, calculatedFieldCache, apiUsageStateService, partitionService, eventPublisher, jwtSettingsService); this.ctx = ctx; this.tbDeviceRpcService = tbDeviceRpcService; this.queueService = queueService; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java index 95fc8cb843..2aaed13ec1 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java @@ -25,6 +25,7 @@ import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.AssetProfileId; +import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceProfileId; @@ -43,6 +44,7 @@ import org.thingsboard.server.queue.discovery.TbApplicationEventListener; import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.service.apiusage.TbApiUsageStateService; +import org.thingsboard.server.service.cf.CalculatedFieldCache; import org.thingsboard.server.service.profile.TbAssetProfileCache; import org.thingsboard.server.service.profile.TbDeviceProfileCache; import org.thingsboard.server.service.queue.TbPackCallback; @@ -68,6 +70,7 @@ public abstract class AbstractConsumerService calculatedFieldExecutionService.onTelemetryUpdate(new CalculatedFieldTimeSeriesUpdateRequest(tenantId, entityId, request.getEntries(), request.getPreviousCalculatedFieldIds()))); return saveFuture; } @@ -170,7 +170,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer ListenableFuture> saveFuture = attrService.save(request.getTenantId(), request.getEntityId(), request.getScope(), request.getEntries()); addMainCallback(saveFuture, request.getCallback()); addWsCallback(saveFuture, success -> onAttributesUpdate(request.getTenantId(), request.getEntityId(), request.getScope().name(), request.getEntries(), request.isNotifyDevice())); - calculatedFieldExecutionService.onTelemetryUpdate(new CalculatedFieldAttributeUpdateRequest(request.getTenantId(), request.getEntityId(), request.getScope(), request.getEntries(), request.getCalculatedFieldIds())); + addCalculatedFieldCallback(saveFuture, success -> calculatedFieldExecutionService.onTelemetryUpdate(new CalculatedFieldAttributeUpdateRequest(request.getTenantId(), request.getEntityId(), request.getScope(), request.getEntries(), request.getPreviousCalculatedFieldIds()))); } @Override @@ -243,7 +243,8 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer .onlyLatest(true) .callback(new FutureCallback<>() { @Override - public void onSuccess(@Nullable Void tmp) {} + public void onSuccess(@Nullable Void tmp) { + } @Override public void onFailure(Throwable t) { @@ -342,4 +343,17 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer }; } + protected void addCalculatedFieldCallback(ListenableFuture saveFuture, Consumer callback) { + Futures.addCallback(saveFuture, new FutureCallback() { + @Override + public void onSuccess(@Nullable T result) { + callback.accept(result); + } + + @Override + public void onFailure(Throwable t) { + } + }, tsCallBackExecutor); + } + } diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java index 0805175f77..9d99f38c60 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java @@ -35,10 +35,10 @@ import org.thingsboard.server.common.msg.gen.MsgProtos; import org.thingsboard.server.common.msg.queue.TbMsgCallback; import java.io.Serializable; -import java.util.ArrayList; import java.util.List; import java.util.Objects; import java.util.UUID; +import java.util.concurrent.CopyOnWriteArrayList; /** * Created by ashvayka on 13.01.18. @@ -67,7 +67,7 @@ public final class TbMsg implements Serializable { private final UUID correlationId; private final Integer partition; - private final List calculatedFieldIds; + private final List previousCalculatedFieldIds; @Getter(value = AccessLevel.NONE) @JsonIgnore @@ -117,7 +117,7 @@ public final class TbMsg implements Serializable { } private TbMsg(String queueName, UUID id, long ts, TbMsgType internalType, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, TbMsgDataType dataType, String data, - RuleChainId ruleChainId, RuleNodeId ruleNodeId, UUID correlationId, Integer partition, List calculatedFieldIds, TbMsgProcessingCtx ctx, TbMsgCallback callback) { + RuleChainId ruleChainId, RuleNodeId ruleNodeId, UUID correlationId, Integer partition, List previousCalculatedFieldIds, TbMsgProcessingCtx ctx, TbMsgCallback callback) { this.id = id != null ? id : UUID.randomUUID(); this.queueName = queueName; if (ts > 0) { @@ -144,7 +144,9 @@ public final class TbMsg implements Serializable { this.ruleNodeId = ruleNodeId; this.correlationId = correlationId; this.partition = partition; - this.calculatedFieldIds = calculatedFieldIds; + this.previousCalculatedFieldIds = previousCalculatedFieldIds != null + ? new CopyOnWriteArrayList<>(previousCalculatedFieldIds) + : new CopyOnWriteArrayList<>(); this.ctx = ctx != null ? ctx : new TbMsgProcessingCtx(); this.callback = Objects.requireNonNullElse(callback, TbMsgCallback.EMPTY); } @@ -192,8 +194,8 @@ public final class TbMsg implements Serializable { builder.setPartition(msg.getPartition()); } - if (msg.getCalculatedFieldIds() != null) { - for (CalculatedFieldId calculatedFieldId : msg.getCalculatedFieldIds()) { + if (msg.getPreviousCalculatedFieldIds() != null) { + for (CalculatedFieldId calculatedFieldId : msg.getPreviousCalculatedFieldIds()) { MsgProtos.CalculatedFieldIdProto calculatedFieldIdProto = MsgProtos.CalculatedFieldIdProto.newBuilder() .setCalculatedFieldIdMSB(calculatedFieldId.getId().getMostSignificantBits()) .setCalculatedFieldIdLSB(calculatedFieldId.getId().getLeastSignificantBits()) @@ -216,7 +218,7 @@ public final class TbMsg implements Serializable { RuleNodeId ruleNodeId = null; UUID correlationId = null; Integer partition = null; - List calculatedFieldIds = new ArrayList<>(); + List calculatedFieldIds = new CopyOnWriteArrayList<>(); if (proto.getCustomerIdMSB() != 0L && proto.getCustomerIdLSB() != 0L) { customerId = new CustomerId(new UUID(proto.getCustomerIdMSB(), proto.getCustomerIdLSB())); } @@ -274,6 +276,7 @@ public final class TbMsg implements Serializable { /** * Checks if the message is still valid for processing. May be invalid if the message pack is timed-out or canceled. + * * @return 'true' if message is valid for processing, 'false' otherwise. */ public boolean isValid() { @@ -368,7 +371,7 @@ public final class TbMsg implements Serializable { protected RuleNodeId ruleNodeId; protected UUID correlationId; protected Integer partition; - protected List calculatedFieldIds; + protected List previousCalculatedFieldIds; protected TbMsgProcessingCtx ctx; protected TbMsgCallback callback; @@ -390,7 +393,7 @@ public final class TbMsg implements Serializable { this.ruleNodeId = tbMsg.ruleNodeId; this.correlationId = tbMsg.correlationId; this.partition = tbMsg.partition; - this.calculatedFieldIds = tbMsg.calculatedFieldIds; + this.previousCalculatedFieldIds = tbMsg.previousCalculatedFieldIds; this.ctx = tbMsg.ctx; this.callback = tbMsg.callback; } @@ -413,8 +416,7 @@ public final class TbMsg implements Serializable { /** *

Deprecated: This should only be used when you need to specify a custom message type that doesn't exist in the {@link TbMsgType} enum. * Prefer using {@link #type(TbMsgType)} instead. - * - * */ + */ @Deprecated public TbMsgBuilder type(String type) { this.type = type; @@ -482,8 +484,8 @@ public final class TbMsg implements Serializable { return this; } - public TbMsgBuilder calculatedFieldIds(List calculatedFieldIds) { - this.calculatedFieldIds = calculatedFieldIds; + public TbMsgBuilder previousCalculatedFieldIds(List previousCalculatedFieldIds) { + this.previousCalculatedFieldIds = previousCalculatedFieldIds; return this; } @@ -498,7 +500,7 @@ public final class TbMsg implements Serializable { } public TbMsg build() { - return new TbMsg(queueName, id, ts, internalType, type, originator, customerId, metaData, dataType, data, ruleChainId, ruleNodeId, correlationId, partition, calculatedFieldIds, ctx, callback); + return new TbMsg(queueName, id, ts, internalType, type, originator, customerId, metaData, dataType, data, ruleChainId, ruleNodeId, correlationId, partition, previousCalculatedFieldIds, ctx, callback); } public String toString() { @@ -506,7 +508,7 @@ public final class TbMsg implements Serializable { ", type=" + this.type + ", internalType=" + this.internalType + ", originator=" + this.originator + ", customerId=" + this.customerId + ", metaData=" + this.metaData + ", dataType=" + this.dataType + ", data=" + this.data + ", ruleChainId=" + this.ruleChainId + ", ruleNodeId=" + this.ruleNodeId + - ", correlationId=" + this.correlationId + ", partition=" + this.partition + ", calculatedFields=" + this.calculatedFieldIds + + ", correlationId=" + this.correlationId + ", partition=" + this.partition + ", previousCalculatedFields=" + this.previousCalculatedFieldIds + ", ctx=" + this.ctx + ", callback=" + this.callback + ")"; } diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 1f66621a5b..9cba62eecc 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -818,7 +818,7 @@ message CalculatedFieldStateMsgProto { int64 entityIdMSB = 6; int64 entityIdLSB = 7; bool clear = 8; - repeated CalculatedFieldIdProto calculatedFields = 9; + repeated CalculatedFieldIdProto previousCalculatedFields = 9; map arguments = 10; } @@ -1620,6 +1620,8 @@ message ToCoreNotificationMsg { FromEdgeSyncResponseMsgProto fromEdgeSyncResponse = 12 [deprecated = true]; ResourceCacheInvalidateMsg resourceCacheInvalidateMsg = 13; RestApiCallResponseMsgProto restApiCallResponseMsg = 50; + EntityProfileUpdateMsgProto entityProfileUpdateMsg = 51; + ProfileEntityMsgProto profileEntityMsg = 52; } /* Messages to Edge queue that are handled by ThingsBoard Core Service */ diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java index 9747a6033e..406b2d0d5c 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java @@ -41,7 +41,7 @@ public class AttributesSaveRequest { private final AttributeScope scope; private final List entries; private final boolean notifyDevice; - private final List calculatedFieldIds; + private final List previousCalculatedFieldIds; private final FutureCallback callback; public static Builder builder() { @@ -55,7 +55,7 @@ public class AttributesSaveRequest { private AttributeScope scope; private List entries; private boolean notifyDevice = true; - private List calculatedFieldIds; + private List previousCalculatedFieldIds; private FutureCallback callback; Builder() {} @@ -103,8 +103,8 @@ public class AttributesSaveRequest { return this; } - public Builder calculatedFieldIds(List calculatedFieldIds) { - this.calculatedFieldIds = calculatedFieldIds; + public Builder previousCalculatedFieldIds(List previousCalculatedFieldIds) { + this.previousCalculatedFieldIds = previousCalculatedFieldIds; return this; } @@ -128,7 +128,7 @@ public class AttributesSaveRequest { } public AttributesSaveRequest build() { - return new AttributesSaveRequest(tenantId, entityId, scope, entries, notifyDevice, calculatedFieldIds, callback); + return new AttributesSaveRequest(tenantId, entityId, scope, entries, notifyDevice, previousCalculatedFieldIds, callback); } } diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java index 12afa2d939..95eb788e5f 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java @@ -41,7 +41,7 @@ public class TimeseriesSaveRequest { private final long ttl; private final boolean saveLatest; private final boolean onlyLatest; - private final List calculatedFieldIds; + private final List previousCalculatedFieldIds; private final FutureCallback callback; public static Builder builder() { @@ -58,7 +58,7 @@ public class TimeseriesSaveRequest { private FutureCallback callback; private boolean saveLatest = true; private boolean onlyLatest; - private List calculatedFieldIds; + private List previousCalculatedFieldIds; Builder() {} @@ -106,8 +106,8 @@ public class TimeseriesSaveRequest { return this; } - public Builder calculatedFieldIds(List calculatedFieldIds) { - this.calculatedFieldIds = calculatedFieldIds; + public Builder previousCalculatedFieldIds(List previousCalculatedFieldIds) { + this.previousCalculatedFieldIds = previousCalculatedFieldIds; return this; } @@ -131,7 +131,7 @@ public class TimeseriesSaveRequest { } public TimeseriesSaveRequest build() { - return new TimeseriesSaveRequest(tenantId, customerId, entityId, entries, ttl, saveLatest, onlyLatest, calculatedFieldIds, callback); + return new TimeseriesSaveRequest(tenantId, customerId, entityId, entries, ttl, saveLatest, onlyLatest, previousCalculatedFieldIds, callback); } } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java index 20d0dda42f..ae2fce6575 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java @@ -125,7 +125,7 @@ public class TbMsgAttributesNode implements TbNode { .scope(scope) .entries(attributes) .notifyDevice(config.isNotifyDevice() || checkNotifyDeviceMdValue(msg.getMetaData().getValue(NOTIFY_DEVICE_METADATA_KEY))) - .calculatedFieldIds(msg.getCalculatedFieldIds()) + .previousCalculatedFieldIds(msg.getPreviousCalculatedFieldIds()) .callback(callback) .build()); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java index 386e56320a..89f7844313 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java @@ -112,7 +112,7 @@ public class TbMsgTimeseriesNode implements TbNode { .entries(tsKvEntryList) .ttl(ttl) .saveLatest(!config.isSkipLatestPersistence()) - .calculatedFieldIds(msg.getCalculatedFieldIds()) + .previousCalculatedFieldIds(msg.getPreviousCalculatedFieldIds()) .callback(new TelemetryNodeCallback(ctx, msg)) .build()); }