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 8f15a6766b..c714ed7dab 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -108,6 +108,7 @@ import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.service.apiusage.TbApiUsageStateService; import org.thingsboard.server.service.cf.CalculatedFieldExecutionService; import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache; +import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService; import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; import org.thingsboard.server.service.component.ComponentDiscoveryService; import org.thingsboard.server.service.edge.rpc.EdgeRpcService; @@ -527,6 +528,11 @@ public class ActorSystemContext { @Getter private CalculatedFieldExecutionService calculatedFieldExecutionService; + @Lazy + @Autowired(required = false) + @Getter + private CalculatedFieldStateService calculatedFieldStateService; + @Lazy @Autowired(required = false) @Getter 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 544dea44c9..0d327adca9 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 @@ -1,12 +1,12 @@ /** * Copyright © 2016-2024 The Thingsboard Authors - * + *

* Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

* Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. @@ -38,6 +38,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto; import org.thingsboard.server.service.cf.CalculatedFieldExecutionService; import org.thingsboard.server.service.cf.CalculatedFieldResult; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; +import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService; import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState; @@ -67,6 +68,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM final TenantId tenantId; final EntityId entityId; final CalculatedFieldExecutionService cfService; + final CalculatedFieldStateService cfStateService; TbActorCtx ctx; Map states = new HashMap<>(); @@ -76,6 +78,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM this.tenantId = tenantId; this.entityId = entityId; this.cfService = systemContext.getCalculatedFieldExecutionService(); + this.cfStateService = systemContext.getCalculatedFieldStateService(); } void init(TbActorCtx ctx) { @@ -102,13 +105,13 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM log.info("[{}] Processing CF entity delete msg.", msg.getEntityId()); if (this.entityId.equals(msg.getEntityId())) { MultipleTbCallback multipleTbCallback = new MultipleTbCallback(states.size(), msg.getCallback()); - states.forEach((cfId, state) -> cfService.deleteStateFromStorage(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), multipleTbCallback)); + states.forEach((cfId, state) -> cfStateService.removeState(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), multipleTbCallback)); ctx.stop(ctx.getSelf()); } else { var cfId = new CalculatedFieldId(msg.getEntityId().getId()); var state = states.remove(cfId); if (state != null) { - cfService.deleteStateFromStorage(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), msg.getCallback()); + cfStateService.removeState(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), msg.getCallback()); } } } @@ -178,8 +181,13 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM log.info("[{}] No new argument values to process for CF.", ctx.getCfId()); callback.onSuccess(CALLBACKS_PER_CF); } - CalculatedFieldState state = getOrInitState(ctx); - if (state.updateState(newArgValues)) { + CalculatedFieldState state = states.get(ctx.getCfId()); + boolean justRestored = false; + if (state == null) { + state = getOrInitState(ctx); + justRestored = true; + } + if (state.updateState(newArgValues) || justRestored) { cfIdList = new ArrayList<>(cfIdList); cfIdList.add(ctx.getCfId()); processStateIfReady(ctx, cfIdList, state, tbMsgId, tbMsgType, callback); @@ -222,7 +230,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } else { callback.onSuccess(); // State was updated but no calculation performed; } - cfService.pushStateToStorage(ctx, new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId), state, callback); + cfStateService.persistState(ctx, new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId), state, callback); } private Map mapToArguments(CalculatedFieldCtx ctx, List data) { 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 6a6edae768..cdc31ed93c 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 @@ -42,6 +42,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntit import org.thingsboard.server.service.cf.CalculatedFieldExecutionService; import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; +import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import org.thingsboard.server.service.profile.TbAssetProfileCache; import org.thingsboard.server.service.profile.TbDeviceProfileCache; @@ -68,6 +69,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private final ConcurrentMap> entityIdCalculatedFieldLinks = new ConcurrentHashMap<>(); private final CalculatedFieldExecutionService cfExecService; + private final CalculatedFieldStateService cfStateService; private final CalculatedFieldEntityProfileCache cfEntityCache; private final CalculatedFieldService cfDaoService; private final TbAssetProfileCache assetProfileCache; @@ -80,6 +82,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware super(systemContext); this.cfEntityCache = systemContext.getCalculatedFieldEntityProfileCache(); this.cfExecService = systemContext.getCalculatedFieldExecutionService(); + this.cfStateService = systemContext.getCalculatedFieldStateService(); this.cfDaoService = systemContext.getCalculatedFieldService(); this.assetProfileCache = systemContext.getAssetProfileCache(); this.deviceProfileCache = systemContext.getDeviceProfileCache(); @@ -127,7 +130,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware log.info("Pushing CF state restore msg to specific actor [{}]", msg.getId().entityId()); getOrCreateActor(msg.getId().entityId()).tell(msg); } else { - cfExecService.deleteStateFromStorage(msg.getId(), msg.getCallback()); + cfStateService.removeState(msg.getId(), msg.getCallback()); } } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldExecutionService.java b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldExecutionService.java index 393fbd3ec2..47fd560b4d 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldExecutionService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldExecutionService.java @@ -43,13 +43,10 @@ public interface CalculatedFieldExecutionService { void pushRequestToQueue(AttributesSaveRequest request, List result, FutureCallback callback); - void pushStateToStorage(CalculatedFieldCtx ctx, CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback); - ListenableFuture fetchStateFromDb(CalculatedFieldCtx ctx, EntityId entityId); void pushMsgToRuleEngine(TenantId tenantId, EntityId entityId, CalculatedFieldResult calculationResult, List cfIds, TbCallback callback); void pushMsgToLinks(CalculatedFieldTelemetryMsg msg, List linkedCalculatedFields, TbCallback callback); - void deleteStateFromStorage(CalculatedFieldEntityCtxId calculatedFieldEntityCtxId, TbCallback callback); } 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 43e85d87e9..e84e785a94 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 @@ -141,7 +141,6 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas private final CalculatedFieldCache calculatedFieldCache; private final AttributesService attributesService; private final TimeseriesService timeseriesService; - private final CalculatedFieldStateService stateService; private final TbClusterService clusterService; private final ApiLimitService apiLimitService; @@ -263,16 +262,6 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas }, calculatedFieldCallbackExecutor); } - @Override - public void pushStateToStorage(CalculatedFieldCtx ctx, CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) { - stateService.persistState(ctx, stateId, state, callback); - } - - @Override - public void deleteStateFromStorage(CalculatedFieldEntityCtxId calculatedFieldEntityCtxId, TbCallback callback) { - stateService.removeState(calculatedFieldEntityCtxId, callback); - } - @Override protected Map>> onAddedPartitions(Set addedPartitions) { var result = new HashMap>>(); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldInitService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldInitService.java index f71617e204..2afd6d8238 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldInitService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldInitService.java @@ -29,7 +29,6 @@ import org.thingsboard.server.dao.device.DeviceService; import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.queue.util.TbRuleEngineComponent; import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache; -import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService; @Slf4j @Service @@ -38,9 +37,6 @@ import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService; public class DefaultCalculatedFieldInitService implements CalculatedFieldInitService { private final CalculatedFieldEntityProfileCache entityProfileCache; - private final CalculatedFieldStateService stateService; - - private final ActorSystemContext actorSystemContext; private final AssetService assetService; private final DeviceService deviceService; @@ -62,9 +58,4 @@ public class DefaultCalculatedFieldInitService implements CalculatedFieldInitSer } } - @AfterStartUp(order = AfterStartUp.CF_STATE_RESTORE_SERVICE) - public void initCalculatedFieldStates() { - stateService.restoreStates().forEach((k, v) -> actorSystemContext.tell(new CalculatedFieldStateRestoreMsg(k, v))); - } - } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/RocksDBService.java b/application/src/main/java/org/thingsboard/server/service/cf/RocksDBService.java index fe800f61ad..7181cc43ed 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/RocksDBService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/RocksDBService.java @@ -21,6 +21,7 @@ import org.rocksdb.RocksDBException; import org.rocksdb.RocksIterator; import org.rocksdb.WriteOptions; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Service; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; @@ -32,7 +33,7 @@ import java.util.Map; @Service @Slf4j -@ConditionalOnExpression("'${service.type:null}'=='monolith'") +@ConditionalOnProperty(prefix = "zk", value = "enabled", havingValue = "false", matchIfMissing = true) public class RocksDBService { private final RocksDB db; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/cache/DefaultCalculatedFieldEntityProfileCache.java b/application/src/main/java/org/thingsboard/server/service/cf/cache/DefaultCalculatedFieldEntityProfileCache.java index 13f95c547d..866cc86f24 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/cache/DefaultCalculatedFieldEntityProfileCache.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/cache/DefaultCalculatedFieldEntityProfileCache.java @@ -39,6 +39,7 @@ import java.util.stream.Collectors; @Service @Slf4j @RequiredArgsConstructor +//TODO: remove and use TenantEntityProfileCache in each CalculatedFieldManagerMessageProcessor; public class DefaultCalculatedFieldEntityProfileCache extends TbApplicationEventListener implements CalculatedFieldEntityProfileCache { private static final Integer UNKNOWN = -1; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/CalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/CalculatedFieldStateService.java index e822d52767..ce1562c735 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/CalculatedFieldStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/CalculatedFieldStateService.java @@ -19,12 +19,8 @@ import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState; -import java.util.Map; - public interface CalculatedFieldStateService { - Map restoreStates(); - void persistState(CalculatedFieldCtx ctx, CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback); void removeState(CalculatedFieldEntityCtxId stateId, TbCallback callback); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java new file mode 100644 index 0000000000..9d45532eec --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java @@ -0,0 +1,68 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.cf.ctx.state; + +import lombok.RequiredArgsConstructor; +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.stereotype.Service; +import org.thingsboard.server.actors.ActorSystemContext; +import org.thingsboard.server.actors.calculatedField.CalculatedFieldStateRestoreMsg; +import org.thingsboard.server.common.data.StringUtils; +import org.thingsboard.server.common.data.cf.CalculatedFieldType; +import org.thingsboard.server.common.data.id.CalculatedFieldId; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.EntityIdFactory; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.kv.BasicKvEntry; +import org.thingsboard.server.common.msg.queue.TbCallback; +import org.thingsboard.server.common.util.KvProtoUtil; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; +import org.thingsboard.server.gen.transport.TransportProtos.SingleValueArgumentProto; +import org.thingsboard.server.gen.transport.TransportProtos.TsValueListProto; +import org.thingsboard.server.gen.transport.TransportProtos.TsValueProto; +import org.thingsboard.server.queue.util.AfterStartUp; +import org.thingsboard.server.service.cf.RocksDBService; +import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; +import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService; + +import java.util.Map; +import java.util.Optional; +import java.util.TreeMap; +import java.util.UUID; +import java.util.stream.Collectors; + +@Service +@RequiredArgsConstructor +@ConditionalOnExpression("'${zk.enabled:false}'=='true' && ('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-rule-engine')") +public class KafkaCalculatedFieldStateService implements CalculatedFieldStateService { + + @AfterStartUp(order = AfterStartUp.CF_STATE_RESTORE_SERVICE) + public void initCalculatedFieldStates() { + } + + @Override + public void persistState(CalculatedFieldCtx ctx, CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) { + callback.onSuccess(); + } + + @Override + public void removeState(CalculatedFieldEntityCtxId ctxId, TbCallback callback) { + callback.onSuccess(); + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java similarity index 92% rename from application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java rename to application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java index b2e33e1705..86556d7adc 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java @@ -16,8 +16,10 @@ package org.thingsboard.server.service.cf.ctx.state; import lombok.RequiredArgsConstructor; -import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Service; +import org.thingsboard.server.actors.ActorSystemContext; +import org.thingsboard.server.actors.calculatedField.CalculatedFieldStateRestoreMsg; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.cf.CalculatedFieldType; import org.thingsboard.server.common.data.id.CalculatedFieldId; @@ -32,6 +34,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldState import org.thingsboard.server.gen.transport.TransportProtos.SingleValueArgumentProto; import org.thingsboard.server.gen.transport.TransportProtos.TsValueListProto; import org.thingsboard.server.gen.transport.TransportProtos.TsValueProto; +import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.service.cf.RocksDBService; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService; @@ -44,12 +47,12 @@ import java.util.stream.Collectors; @Service @RequiredArgsConstructor -@ConditionalOnExpression("'${service.type:null}'=='monolith'") -public class RocksDBStateService implements CalculatedFieldStateService { +@ConditionalOnProperty(prefix = "zk", value = "enabled", havingValue = "false", matchIfMissing = true) +public class RocksDBCalculatedFieldStateService implements CalculatedFieldStateService { + private final ActorSystemContext actorSystemContext; private final RocksDBService rocksDBService; - @Override public Map restoreStates() { return rocksDBService.getAll().entrySet().stream() .collect(Collectors.toMap( @@ -58,6 +61,12 @@ public class RocksDBStateService implements CalculatedFieldStateService { )); } + @AfterStartUp(order = AfterStartUp.CF_STATE_RESTORE_SERVICE) + public void initCalculatedFieldStates() { + restoreStates().forEach((k, v) -> actorSystemContext.tell(new CalculatedFieldStateRestoreMsg(k, v))); + } + + @Override public void persistState(CalculatedFieldCtx ctx, CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) { CalculatedFieldStateProto stateProto = toProto(stateId, state); 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 fa02435ca0..2a090fe701 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 @@ -674,7 +674,7 @@ public class DefaultTbClusterService implements TbClusterService { .oldName(old.getName()) .name(entity.getName()) .build(); - pushMsgToCalculatedFields(entity.getTenantId(), entity.getId(), ToCalculatedFieldMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY); + broadcastToCalculatedFields(ToCalculatedFieldNotificationMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY); } if (deviceNameChanged || deviceProfileChanged) { pushMsgToCore(new DeviceNameOrTypeUpdateMsg(entity.getTenantId(), entity.getId(), entity.getName(), entity.getType()), null); @@ -687,7 +687,7 @@ public class DefaultTbClusterService implements TbClusterService { .profileId(entity.getDeviceProfileId()) .name(entity.getName()) .build(); - pushMsgToCalculatedFields(entity.getTenantId(), entity.getId(), ToCalculatedFieldMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY); + broadcastToCalculatedFields(ToCalculatedFieldNotificationMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY); } broadcastEntityStateChangeEvent(entity.getTenantId(), entity.getId(), created ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED); sendDeviceStateServiceEvent(entity.getTenantId(), entity.getId(), created, !created, false); @@ -710,7 +710,7 @@ public class DefaultTbClusterService implements TbClusterService { .oldName(old.getName()) .name(entity.getName()) .build(); - pushMsgToCalculatedFields(entity.getTenantId(), entity.getId(), ToCalculatedFieldMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY); + broadcastToCalculatedFields(ToCalculatedFieldNotificationMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY); } } else { ComponentLifecycleMsg msg = ComponentLifecycleMsg.builder() @@ -720,7 +720,7 @@ public class DefaultTbClusterService implements TbClusterService { .profileId(entity.getAssetProfileId()) .name(entity.getName()) .build(); - pushMsgToCalculatedFields(entity.getTenantId(), entity.getId(), ToCalculatedFieldMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY); + broadcastToCalculatedFields(ToCalculatedFieldNotificationMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY); } broadcastEntityStateChangeEvent(entity.getTenantId(), entity.getId(), created ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED); } @@ -872,6 +872,6 @@ public class DefaultTbClusterService implements TbClusterService { private void handleCalculatedFieldEntityDeleted(TenantId tenantId, EntityId entityId) { ComponentLifecycleMsg msg = new ComponentLifecycleMsg(tenantId, entityId, ComponentLifecycleEvent.DELETED); - pushMsgToCalculatedFields(tenantId, entityId, ToCalculatedFieldMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY); + broadcastToCalculatedFields(ToCalculatedFieldNotificationMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY); } }