diff --git a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java index 9124dc0a9a..8a6907c107 100644 --- a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java @@ -115,12 +115,13 @@ public class AppActor extends ContextAwareActor { case CF_INIT_MSG: case CF_LINK_INIT_MSG: case CF_STATE_RESTORE_MSG: - case CF_UPDATE_MSG: + case CF_ENTITY_LIFECYCLE_MSG: + //TODO: use priority from the message body. For example, messages about CF lifecycle are important and Device lifecycle are not. + // same for the Linked telemetry. onToCalculatedFieldSystemActorMsg((ToCalculatedFieldSystemMsg) msg, true); break; case CF_TELEMETRY_MSG: case CF_LINKED_TELEMETRY_MSG: - case CF_ENTITY_UPDATE_MSG: onToCalculatedFieldSystemActorMsg((ToCalculatedFieldSystemMsg) msg, false); break; default: diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java index a5461c6152..cebe6b6a60 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java @@ -53,6 +53,12 @@ public class CalculatedFieldEntityActor extends ContextAwareActor { case CF_STATE_RESTORE_MSG: processor.process((CalculatedFieldStateRestoreMsg) msg); break; + case CF_ENTITY_INIT_CF_MSG: + processor.process((EntityInitCalculatedFieldMsg) msg); + break; + case CF_ENTITY_DELETE_MSG: + processor.process((CalculatedFieldEntityDeleteMsg) msg); + break; case CF_ENTITY_TELEMETRY_MSG: processor.process((EntityCalculatedFieldTelemetryMsg) msg); break; diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityDeleteMsg.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityDeleteMsg.java new file mode 100644 index 0000000000..8df6a63a82 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityDeleteMsg.java @@ -0,0 +1,44 @@ +/** + * 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.actors.calculatedField; + +import lombok.Data; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.msg.MsgType; +import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg; +import org.thingsboard.server.common.msg.queue.TbCallback; + +@Data +public class CalculatedFieldEntityDeleteMsg implements ToCalculatedFieldSystemMsg { + + private final TenantId tenantId; + private final EntityId entityId; + private final TbCallback callback; + + public CalculatedFieldEntityDeleteMsg(TenantId tenantId, + EntityId entityId, + TbCallback callback) { + this.tenantId = tenantId; + this.entityId = entityId; + this.callback = callback; + } + + @Override + public MsgType getMsgType() { + return MsgType.CF_ENTITY_DELETE_MSG; + } +} 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 6426be8a3c..f0064ff459 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 @@ -30,6 +30,7 @@ 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.msg.TbMsgType; +import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.gen.transport.TransportProtos.AttributeScopeProto; import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; @@ -85,6 +86,29 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM states.put(msg.getId().cfId(), msg.getState()); } + public void process(EntityInitCalculatedFieldMsg msg) { + var cfCtx = msg.getCtx(); + if (msg.isForceReinit()) { + states.remove(cfCtx.getCfId()); + } + var cfState = getOrInitState(cfCtx); + processStateIfReady(cfCtx, Collections.singletonList(cfCtx.getCfId()), cfState, null, null, msg.getCallback()); + } + + public void process(CalculatedFieldEntityDeleteMsg msg) { + 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)); + 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()); + } + } + } + public void process(EntityCalculatedFieldTelemetryMsg msg) { var proto = msg.getProto(); var numberOfCallbacks = CALLBACKS_PER_CF * (msg.getEntityIdFields().size() + msg.getProfileIdFields().size()); @@ -149,18 +173,9 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } CalculatedFieldState state = getOrInitState(ctx); if (state.updateState(newArgValues)) { - if (state.isReady() && ctx.isInitialized()) { - CalculatedFieldResult calculationResult = state.performCalculation(ctx).get(5, TimeUnit.SECONDS); - cfIdList = new ArrayList<>(cfIdList); - cfIdList.add(ctx.getCfId()); - cfService.pushMsgToRuleEngine(tenantId, entityId, calculationResult, cfIdList, callback); - if (DebugModeUtil.isDebugAllAvailable(ctx.getCalculatedField())) { - systemContext.persistCalculatedFieldDebugEvent(tenantId, ctx.getCfId(), entityId, state.getArguments(), tbMsgId, tbMsgType, JacksonUtil.writeValueAsString(calculationResult.getResultMap()), null); - } - } else { - callback.onSuccess(); // State was updated but no calculation performed; - } - cfService.pushStateToStorage(new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId), state, callback); + cfIdList = new ArrayList<>(cfIdList); + cfIdList.add(ctx.getCfId()); + processStateIfReady(ctx, cfIdList, state, tbMsgId, tbMsgType, callback); } else { callback.onSuccess(CALLBACKS_PER_CF); } @@ -196,6 +211,20 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } return null; } + + @SneakyThrows + private void processStateIfReady(CalculatedFieldCtx ctx, List cfIdList, CalculatedFieldState state, UUID tbMsgId, TbMsgType tbMsgType, TbCallback callback) { + if (state.isReady()) { + CalculatedFieldResult calculationResult = state.performCalculation(ctx).get(5, TimeUnit.SECONDS); + cfService.pushMsgToRuleEngine(tenantId, entityId, calculationResult, cfIdList, callback); + if (DebugModeUtil.isDebugAllAvailable(ctx.getCalculatedField())) { + systemContext.persistCalculatedFieldDebugEvent(tenantId, ctx.getCfId(), entityId, state.getArguments(), tbMsgId, tbMsgType, JacksonUtil.writeValueAsString(calculationResult.getResultMap()), null); + } + } else { + callback.onSuccess(); // State was updated but no calculation performed; + } + cfService.pushStateToStorage(new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId), state, callback); + } private Map mapToArguments(CalculatedFieldCtx ctx, List data) { return mapToArguments(ctx.getMainEntityArguments(), data); diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerActor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerActor.java index 1c198c660d..22909dd5af 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerActor.java @@ -22,6 +22,7 @@ import org.thingsboard.server.actors.TbActorException; import org.thingsboard.server.actors.service.ContextAwareActor; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.TbActorMsg; +import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldInitMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldLinkInitMsg; @@ -66,8 +67,8 @@ public class CalculatedFieldManagerActor extends ContextAwareActor { case CF_STATE_RESTORE_MSG: processor.onStateRestoreMsg((CalculatedFieldStateRestoreMsg) msg); break; - case CF_UPDATE_MSG: -// processor.onToCalculatedFieldSystemActorMsg((ToCalculatedFieldSystemMsg) msg); + case CF_ENTITY_LIFECYCLE_MSG: + processor.onEntityLifecycleMsg((CalculatedFieldEntityLifecycleMsg) msg); break; case CF_TELEMETRY_MSG: processor.onTelemetryMsg((CalculatedFieldTelemetryMsg) msg); @@ -75,9 +76,6 @@ public class CalculatedFieldManagerActor extends ContextAwareActor { case CF_LINKED_TELEMETRY_MSG: processor.onLinkedTelemetryMsg((CalculatedFieldLinkedTelemetryMsg) msg); break; - case CF_ENTITY_UPDATE_MSG: -// processor.onToCalculatedFieldSystemActorMsg((ToCalculatedFieldSystemMsg) msg); - break; default: return 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 603ab96a8e..a2577a3f27 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 @@ -34,8 +34,12 @@ 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.page.PageDataIterable; +import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldInitMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldLinkInitMsg; +import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; +import org.thingsboard.server.common.msg.queue.TbCallback; +import org.thingsboard.server.dao.cf.CalculatedFieldService; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; import org.thingsboard.server.service.cf.CalculatedFieldExecutionService; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; @@ -67,7 +71,8 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private final ConcurrentMap> entityIdCalculatedFieldLinks = new ConcurrentHashMap<>(); private final ConcurrentMap> profileEntities = new ConcurrentHashMap<>(); - private final CalculatedFieldExecutionService cfService; + private final CalculatedFieldExecutionService cfExecService; + private final CalculatedFieldService cfDaoService; private final TbAssetProfileCache assetProfileCache; private final TbDeviceProfileCache deviceProfileCache; @@ -77,7 +82,8 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware CalculatedFieldManagerMessageProcessor(ActorSystemContext systemContext, TenantId tenantId) { super(systemContext); - this.cfService = systemContext.getCalculatedFieldExecutionService(); + this.cfExecService = systemContext.getCalculatedFieldExecutionService(); + this.cfDaoService = systemContext.getCalculatedFieldService(); this.assetProfileCache = systemContext.getAssetProfileCache(); this.deviceProfileCache = systemContext.getDeviceProfileCache(); this.tenantId = tenantId; @@ -100,7 +106,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware calculatedFields.put(cf.getId(), cfCtx); // We use copy on write lists to safely pass the reference to another actor for the iteration. // Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead) - entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(cfCtx); + entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new ArrayList<>()).add(cfCtx); msg.getCallback().onSuccess(); } @@ -108,7 +114,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware var link = msg.getLink(); // We use copy on write lists to safely pass the reference to another actor for the iteration. // Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead) - entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(link); + entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new ArrayList<>()).add(link); msg.getCallback().onSuccess(); } @@ -120,6 +126,168 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } } + public void onEntityLifecycleMsg(CalculatedFieldEntityLifecycleMsg msg) { + var entityType = msg.getData().getEntityId().getEntityType(); + var event = msg.getData().getEvent(); + switch (entityType) { + case CALCULATED_FIELD: { + switch (event) { + case CREATED: + onCfCreated(msg.getData(), msg.getCallback()); + break; + case UPDATED: + onCfUpdated(msg.getData(), msg.getCallback()); + break; + case DELETED: + onCfDeleted(msg.getData(), msg.getCallback()); + break; + default: + msg.getCallback().onSuccess(); + break; + } + break; + } + case DEVICE: + case ASSET: { + switch (event) { + case CREATED: + onEntityCreated(msg.getData(), msg.getCallback()); + break; + case UPDATED: + onEntityUpdated(msg.getData(), msg.getCallback()); + break; + case DELETED: + onEntityDeleted(msg.getData(), msg.getCallback()); + break; + default: + msg.getCallback().onSuccess(); + break; + } + break; + } + default: { + msg.getCallback().onSuccess(); + } + } + } + + private void onEntityCreated(ComponentLifecycleMsg msg, TbCallback callback) { + EntityId entityId = msg.getEntityId(); + var entityIdFields = getCalculatedFieldsByEntityId(entityId); + var profileIdFields = getCalculatedFieldsByEntityId(getProfileId(tenantId, entityId)); + var fieldsCount = entityIdFields.size() + profileIdFields.size(); + if (fieldsCount > 0) { + MultipleTbCallback multiCallback = new MultipleTbCallback(fieldsCount, callback); + entityIdFields.forEach(ctx -> initCfForEntity(entityId, ctx, true, multiCallback)); + profileIdFields.forEach(ctx -> initCfForEntity(entityId, ctx, true, multiCallback)); + } else { + callback.onSuccess(); + } + } + + private void onEntityUpdated(ComponentLifecycleMsg msg, TbCallback callback) { + if (msg.getOldProfileId() != null && msg.getOldProfileId() != msg.getProfileId()) { + var oldProfileCfs = getCalculatedFieldsByEntityId(msg.getOldProfileId()); + var newProfileCfs = getCalculatedFieldsByEntityId(msg.getProfileId()); + var fieldsCount = oldProfileCfs.size() + newProfileCfs.size(); + if (fieldsCount > 0) { + MultipleTbCallback multiCallback = new MultipleTbCallback(fieldsCount, callback); + var entityId = msg.getEntityId(); + oldProfileCfs.forEach(ctx -> deleteCfForEntity(entityId, ctx.getCfId(), callback)); + newProfileCfs.forEach(ctx -> initCfForEntity(entityId, ctx, true, multiCallback)); + } else { + callback.onSuccess(); + } + } + } + + private void onEntityDeleted(ComponentLifecycleMsg msg, TbCallback callback) { + getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback)); + } + + private void onCfCreated(ComponentLifecycleMsg msg, TbCallback callback) { + var cfId = new CalculatedFieldId(msg.getEntityId().getId()); + if (calculatedFields.containsKey(cfId)) { + log.warn("[{}] CF was already initialized [{}]", tenantId, cfId); + callback.onSuccess(); + } else { + var cf = cfDaoService.findById(msg.getTenantId(), cfId); + if (cf == null) { + log.warn("[{}] Failed to lookup CF by id [{}]", tenantId, cfId); + callback.onSuccess(); + } else { + var cfCtx = new CalculatedFieldCtx(cf, systemContext.getTbelInvokeService()); + calculatedFields.put(cf.getId(), cfCtx); + // We use copy on write lists to safely pass the reference to another actor for the iteration. + // Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead) + entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(cfCtx); + initCf(cfCtx, callback, false); + } + } + } + + private void onCfUpdated(ComponentLifecycleMsg msg, TbCallback callback) { + var cfId = new CalculatedFieldId(msg.getEntityId().getId()); + var oldCfCtx = calculatedFields.get(cfId); + if (oldCfCtx == null) { + onCfCreated(msg, callback); + } else { + var newCf = cfDaoService.findById(msg.getTenantId(), cfId); + if (newCf == null) { + log.warn("[{}] Failed to lookup CF by id [{}]", tenantId, cfId); + callback.onSuccess(); + } else { + var newCfCtx = new CalculatedFieldCtx(newCf, systemContext.getTbelInvokeService()); + calculatedFields.put(newCf.getId(), newCfCtx); + List oldCfList = entityIdCalculatedFields.get(newCf.getId()); + List newCfList = new ArrayList<>(oldCfList.size()); + boolean found = false; + for (CalculatedFieldCtx oldCtx : oldCfList) { + if (oldCtx.getCfId().equals(newCf.getId())) { + newCfList.add(newCfCtx); + found = true; + } else { + newCfList.add(oldCtx); + } + } + if (!found) { + newCfList.add(newCfCtx); + } + entityIdCalculatedFields.put(newCf.getId(), newCfList); + // We use copy on write lists to safely pass the reference to another actor for the iteration. + // Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead) + if (newCfCtx.hasSignificantChanges(oldCfCtx)) { + initCf(newCfCtx, callback, true); + } + } + } + + } + + private void onCfDeleted(ComponentLifecycleMsg msg, TbCallback callback) { + var cfId = new CalculatedFieldId(msg.getEntityId().getId()); + var cfCtx = calculatedFields.remove(cfId); + if (cfCtx == null) { + log.warn("[{}] CF was already deleted [{}]", tenantId, cfId); + callback.onSuccess(); + } else { + EntityId entityId = cfCtx.getEntityId(); + EntityType entityType = cfCtx.getEntityId().getEntityType(); + if (isProfileEntity(entityType)) { + var entityIds = getEntitiesByProfile(entityId); + if (!entityIds.isEmpty()) { + //TODO: no need to do this if we cache all created actors and know which one belong to us; + var multiCallback = new MultipleTbCallback(entityIds.size(), callback); + entityIds.forEach(id -> deleteCfForEntity(entityId, cfId, multiCallback)); + } else { + callback.onSuccess(); + } + } else { + deleteCfForEntity(entityId, cfId, callback); + } + } + } + public void onTelemetryMsg(CalculatedFieldTelemetryMsg msg) { EntityId entityId = msg.getEntityId(); // 2 = 1 for CF processing + 1 for links processing @@ -136,7 +304,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware List linkedCalculatedFields = filterCalculatedFieldLinks(msg); var linksSize = linkedCalculatedFields.size(); if (linksSize > 0) { - cfService.pushMsgToLinks(msg, linkedCalculatedFields, callback); + cfExecService.pushMsgToLinks(msg, linkedCalculatedFields, callback); } else { callback.onSuccess(); } @@ -233,6 +401,33 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware return entities; } + private void initCf(CalculatedFieldCtx cfCtx, TbCallback callback, boolean forceStateReinit) { + EntityId entityId = cfCtx.getEntityId(); + EntityType entityType = cfCtx.getEntityId().getEntityType(); + if (isProfileEntity(entityType)) { + var entityIds = getEntitiesByProfile(entityId); + if (!entityIds.isEmpty()) { + var multiCallback = new MultipleTbCallback(entityIds.size(), callback); + entityIds.forEach(id -> initCfForEntity(id, cfCtx, forceStateReinit, multiCallback)); + } else { + callback.onSuccess(); + } + } else { + initCfForEntity(entityId, cfCtx, forceStateReinit, callback); + } + } + + private void deleteCfForEntity(EntityId entityId, CalculatedFieldId cfId, TbCallback callback) { + getOrCreateActor(entityId).tell(new CalculatedFieldEntityDeleteMsg(tenantId, cfId, callback)); + } + + private void initCfForEntity(EntityId entityId, CalculatedFieldCtx cfCtx, boolean forceStateReinit, TbCallback callback) { + getOrCreateActor(entityId).tell(new EntityInitCalculatedFieldMsg(tenantId, cfCtx, callback, forceStateReinit)); + } + + private static boolean isProfileEntity(EntityType entityType) { + return EntityType.DEVICE_PROFILE.equals(entityType) || EntityType.ASSET_PROFILE.equals(entityType); + } private EntityId getProfileId(TenantId tenantId, EntityId entityId) { return switch (entityId.getEntityType()) { diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityInitCalculatedFieldMsg.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityInitCalculatedFieldMsg.java new file mode 100644 index 0000000000..00a378c90a --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityInitCalculatedFieldMsg.java @@ -0,0 +1,41 @@ +/** + * 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.actors.calculatedField; + +import lombok.Data; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.msg.MsgType; +import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg; +import org.thingsboard.server.common.msg.queue.TbCallback; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; +import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; + +import java.util.List; + +@Data +public class EntityInitCalculatedFieldMsg implements ToCalculatedFieldSystemMsg { + + private final TenantId tenantId; + private final CalculatedFieldCtx ctx; + private final TbCallback callback; + private final boolean forceReinit; + + @Override + public MsgType getMsgType() { + return MsgType.CF_ENTITY_INIT_CF_MSG; + } +} diff --git a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java index 6ae10aaece..1b008f462a 100644 --- a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java @@ -170,12 +170,11 @@ public class TenantActor extends RuleChainManagerActor { case CF_INIT_MSG: case CF_LINK_INIT_MSG: case CF_STATE_RESTORE_MSG: - case CF_UPDATE_MSG: + case CF_ENTITY_LIFECYCLE_MSG: onToCalculatedFieldSystemActorMsg((ToCalculatedFieldSystemMsg) msg, true); break; case CF_TELEMETRY_MSG: case CF_LINKED_TELEMETRY_MSG: - case CF_ENTITY_UPDATE_MSG: onToCalculatedFieldSystemActorMsg((ToCalculatedFieldSystemMsg) msg, false); break; default: 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 d8daae2e64..19f60165cf 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 @@ -20,16 +20,11 @@ import com.google.common.util.concurrent.ListenableFuture; import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.actors.calculatedField.CalculatedFieldTelemetryMsg; -import org.thingsboard.server.common.data.cf.CalculatedField; 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.TimeseriesSaveResult; import org.thingsboard.server.common.msg.queue.TbCallback; -import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityUpdateMsgProto; -import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldLinkedTelemetryMsgProto; -import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; -import org.thingsboard.server.gen.transport.TransportProtos.ComponentLifecycleMsgProto; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState; @@ -50,25 +45,11 @@ public interface CalculatedFieldExecutionService { void pushStateToStorage(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback); - void pushCalculatedFieldLifecycleMsgToQueue(CalculatedField calculatedField, ComponentLifecycleMsgProto proto); - 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 pushEntityUpdateMsg(TransportProtos.CalculatedFieldEntityUpdateMsgProto proto, TbCallback callback); - - /* ===================================================== */ - - void onCalculatedFieldLifecycleMsg(ComponentLifecycleMsgProto proto, TbCallback callback); - - void onTelemetryUpdate(CalculatedFieldTelemetryMsgProto proto, TbCallback callback); - - void onTelemetryUpdate(CalculatedFieldLinkedTelemetryMsgProto proto, TbCallback callback); - - void onEntityUpdateMsg(CalculatedFieldEntityUpdateMsgProto proto, 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 7367eb611f..b9a18084b1 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 @@ -39,7 +39,6 @@ import org.thingsboard.server.actors.calculatedField.MultipleTbCallback; 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; import org.thingsboard.server.common.data.cf.configuration.Argument; import org.thingsboard.server.common.data.cf.configuration.OutputType; @@ -64,7 +63,6 @@ import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; -import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.common.util.ProtoUtils; @@ -74,13 +72,10 @@ import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.gen.transport.TransportProtos.AttributeScopeProto; import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; -import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityUpdateMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldIdProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldLinkedTelemetryMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldLinkedTelemetryMsgProto.Builder; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; -import org.thingsboard.server.gen.transport.TransportProtos.ComponentLifecycleEvent; -import org.thingsboard.server.gen.transport.TransportProtos.ComponentLifecycleMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto; @@ -273,6 +268,11 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas stateService.persistState(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>>(); @@ -330,271 +330,44 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas states.keySet().removeIf(ctxId -> ctxId.cfId().equals(calculatedFieldId)); } - @Override - public void pushCalculatedFieldLifecycleMsgToQueue(CalculatedField calculatedField, ComponentLifecycleMsgProto proto) { - EntityId entityId = calculatedField.getEntityId(); - ToCalculatedFieldMsg msg = ToCalculatedFieldMsg.newBuilder().setComponentLifecycleMsg(proto).build(); - switch (entityId.getEntityType()) { - case ASSET, DEVICE -> { - TopicPartitionInfo tpi = partitionService.resolve(CALCULATED_FIELD_QUEUE_KEY, entityId); - clusterService.pushMsgToCalculatedFields(tpi, UUID.randomUUID(), msg, null); - } - case ASSET_PROFILE, DEVICE_PROFILE -> { - Set tpiSet = calculatedFieldCache.getEntitiesByProfile(calculatedField.getTenantId(), entityId).stream() - .map(targetEntityId -> partitionService.resolve(CALCULATED_FIELD_QUEUE_KEY, targetEntityId)) - .collect(Collectors.toSet()); - tpiSet.forEach(tpi -> clusterService.pushMsgToCalculatedFields(tpi, UUID.randomUUID(), msg, null)); - } - default -> throw new IllegalArgumentException("Entity type '" + calculatedField.getId().getEntityType() - + "' does not support calculated fields."); - } - } - - @Override - public void onCalculatedFieldLifecycleMsg(ComponentLifecycleMsgProto proto, TbCallback callback) { - try { - TenantId tenantId = TenantId.fromUUID(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())); - CalculatedFieldId calculatedFieldId = new CalculatedFieldId(new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB())); - log.info("Received CalculatedFieldMsgProto for processing: tenantId=[{}], calculatedFieldId=[{}]", tenantId, calculatedFieldId); - ComponentLifecycleEvent event = proto.getEvent(); - if (ComponentLifecycleEvent.DELETED.equals(event)) { - log.warn("Executing onCalculatedFieldDelete, calculatedFieldId=[{}]", calculatedFieldId); - calculatedFieldCache.evict(calculatedFieldId); - onCalculatedFieldDelete(calculatedFieldId, callback); - callback.onSuccess(); - } - CalculatedField cf = calculatedFieldService.findById(tenantId, calculatedFieldId); - if (ComponentLifecycleEvent.UPDATED.equals(event)) { - log.info("Executing onCalculatedFieldUpdate, calculatedFieldId=[{}]", calculatedFieldId); - boolean shouldReinit = onCalculatedFieldUpdate(cf, callback); - if (!shouldReinit) { - return; - } - } - if (cf != null) { - calculatedFieldCache.addCalculatedField(tenantId, calculatedFieldId); - EntityId entityId = cf.getEntityId(); - CalculatedFieldCtx calculatedFieldCtx = calculatedFieldCache.getCalculatedFieldCtx(calculatedFieldId); - switch (entityId.getEntityType()) { - case ASSET, DEVICE -> { - log.info("Initializing state for entity: tenantId=[{}], entityId=[{}]", tenantId, entityId); - initializeStateForEntity(calculatedFieldCtx, entityId, callback); - } - case ASSET_PROFILE, DEVICE_PROFILE -> { - log.info("Initializing state for all entities in profile: tenantId=[{}], profileId=[{}]", tenantId, entityId); - Map commonArguments = calculatedFieldCtx.getArguments().entrySet().stream() - .filter(entry -> entry.getValue().getRefEntityId() != null) - .collect(Collectors.toMap(Entry::getKey, Entry::getValue)); - fetchArguments(tenantId, entityId, commonArguments, commonArgs -> { - calculatedFieldCache.getEntitiesByProfile(tenantId, entityId).forEach(targetEntityId -> { - initializeStateForEntity(calculatedFieldCtx, targetEntityId, commonArgs, callback); - }); - }); - } - default -> - throw new IllegalArgumentException("Entity type '" + calculatedFieldId.getEntityType() + "' does not support calculated fields."); - } - } else { - //Calculated field was probably deleted while message was in queue; - log.warn("Calculated field not found, possibly deleted: {}", calculatedFieldId); - callback.onSuccess(); - } - callback.onSuccess(); - log.info("Successfully processed calculated field message for calculatedFieldId: [{}]", calculatedFieldId); - } catch (Exception e) { - log.trace("Failed to process calculated field msg: [{}]", proto, e); - callback.onFailure(e); - } - } - - private boolean onCalculatedFieldUpdate(CalculatedField updatedCalculatedField, TbCallback callback) { - CalculatedField oldCalculatedField = calculatedFieldCache.getCalculatedField(updatedCalculatedField.getId()); - boolean shouldReinit = true; - if (hasSignificantChanges(oldCalculatedField, updatedCalculatedField)) { - onCalculatedFieldDelete(updatedCalculatedField.getId(), callback); - } else { - calculatedFieldCache.updateCalculatedField(updatedCalculatedField.getTenantId(), updatedCalculatedField.getId()); - callback.onSuccess(); - shouldReinit = false; - } - return shouldReinit; - } - - private void onCalculatedFieldDelete(CalculatedFieldId calculatedFieldId, TbCallback callback) { - try { - cleanupEntity(calculatedFieldId); - states.keySet().removeIf(ctxId -> { - if (ctxId.cfId().equals(calculatedFieldId)) { - stateService.removeState(ctxId); - return true; - } - return false; - }); - } catch (Exception e) { - log.trace("Failed to delete calculated field: [{}]", calculatedFieldId, e); - callback.onFailure(e); - } - } - - private boolean hasSignificantChanges(CalculatedField oldCalculatedField, CalculatedField newCalculatedField) { - if (oldCalculatedField == null) { - return true; - } - boolean entityIdChanged = !oldCalculatedField.getEntityId().equals(newCalculatedField.getEntityId()); - boolean typeChanged = !oldCalculatedField.getType().equals(newCalculatedField.getType()); - boolean argumentsChanged = !oldCalculatedField.getConfiguration().getArguments().equals(newCalculatedField.getConfiguration().getArguments()); - - return entityIdChanged || typeChanged || argumentsChanged; - } - - @Override - public void onTelemetryUpdate(CalculatedFieldTelemetryMsgProto proto, TbCallback callback) { - try { - CalculatedFieldTelemetryUpdateRequest request = fromProto(proto); - EntityId entityId = request.getEntityId(); - - if (supportedReferencedEntities.contains(entityId.getEntityType())) { - TenantId tenantId = request.getTenantId(); - TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tenantId, entityId); - - if (tpi.isMyPartition()) { - - processCalculatedFields(request, entityId); - processCalculatedFields(request, getProfileId(tenantId, entityId)); - - Map> tpiStatesToUpdate = new HashMap<>(); - processCalculatedFieldLinks(request, tpiStatesToUpdate); - if (!tpiStatesToUpdate.isEmpty()) { - tpiStatesToUpdate.forEach((topicPartitionInfo, ctxIds) -> { - CalculatedFieldLinkedTelemetryMsgProto linkedTelemetryMsgProto = buildLinkedTelemetryMsgProto(proto, ctxIds); - clusterService.pushMsgToCalculatedFields(topicPartitionInfo, UUID.randomUUID(), ToCalculatedFieldMsg.newBuilder().setLinkedTelemetryMsg(linkedTelemetryMsgProto).build(), null); - }); - } - } else { - clusterService.pushMsgToCalculatedFields(tpi, UUID.randomUUID(), ToCalculatedFieldMsg.newBuilder().setTelemetryMsg(proto).build(), null); - } - } - } catch (Exception e) { - log.trace("Failed to update telemetry.", e); - } - } - - private void processCalculatedFields(CalculatedFieldTelemetryUpdateRequest request, EntityId cfTargetEntityId) { - if (cfTargetEntityId != null) { - calculatedFieldCache.getCalculatedFieldCtxsByEntityId(cfTargetEntityId).forEach(ctx -> { - Map updatedTelemetry = request.getMappedTelemetry(ctx, cfTargetEntityId); - if (!updatedTelemetry.isEmpty()) { - EntityId targetEntityId = isProfileEntity(cfTargetEntityId) ? request.getEntityId() : cfTargetEntityId; - executeTelemetryUpdate(ctx, targetEntityId, request.getPreviousCalculatedFieldIds(), updatedTelemetry); - } - }); - } - } - - private void processCalculatedFieldLinks(CalculatedFieldTelemetryUpdateRequest request, Map> tpiStates) { - TenantId tenantId = request.getTenantId(); - EntityId entityId = request.getEntityId(); - - calculatedFieldCache.getCalculatedFieldLinksByEntityId(entityId) - .forEach(link -> { - CalculatedFieldId calculatedFieldId = link.getCalculatedFieldId(); - CalculatedFieldCtx ctx = calculatedFieldCache.getCalculatedFieldCtx(calculatedFieldId); - EntityId targetEntityId = ctx.getEntityId(); - - if (isProfileEntity(targetEntityId)) { - calculatedFieldCache.getEntitiesByProfile(tenantId, targetEntityId).forEach(entityByProfile -> { - processCalculatedFieldLink(request, entityByProfile, ctx, tpiStates); - }); - } else { - processCalculatedFieldLink(request, targetEntityId, ctx, tpiStates); - } - }); - } - - private void processCalculatedFieldLink(CalculatedFieldTelemetryUpdateRequest request, EntityId targetEntity, CalculatedFieldCtx ctx, Map> tpiStates) { - TopicPartitionInfo targetEntityTpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, request.getTenantId(), targetEntity); - if (targetEntityTpi.isMyPartition()) { - Map updatedTelemetry = request.getMappedTelemetry(ctx, request.getEntityId()); - if (!updatedTelemetry.isEmpty()) { - executeTelemetryUpdate(ctx, targetEntity, request.getPreviousCalculatedFieldIds(), updatedTelemetry); - } - } else { - List ctxIds = tpiStates.computeIfAbsent(targetEntityTpi, k -> new ArrayList<>()); - ctxIds.add(new CalculatedFieldEntityCtxId(ctx.getTenantId(), ctx.getCfId(), targetEntity)); - } - } - - @Override - public void onTelemetryUpdate(CalculatedFieldLinkedTelemetryMsgProto proto, TbCallback callback) { - try { - CalculatedFieldTelemetryUpdateRequest request = fromProto(proto.getMsg()); - - if (proto.getLinksList().isEmpty()) { - onTelemetryUpdate(proto.getMsg(), callback); - return; - } - - proto.getLinksList().forEach(ctxIdProto -> { - CalculatedFieldId calculatedFieldId = new CalculatedFieldId(new UUID(ctxIdProto.getCalculatedFieldIdMSB(), ctxIdProto.getCalculatedFieldIdLSB())); - CalculatedFieldCtx ctx = calculatedFieldCache.getCalculatedFieldCtx(calculatedFieldId); - - Map updatedTelemetry = request.getMappedTelemetry(ctx, request.getEntityId()); - if (!updatedTelemetry.isEmpty()) { - EntityId targetEntityId = EntityIdFactory.getByTypeAndUuid(ctxIdProto.getEntityType(), new UUID(ctxIdProto.getEntityIdMSB(), ctxIdProto.getEntityIdLSB())); - executeTelemetryUpdate(ctx, targetEntityId, request.getPreviousCalculatedFieldIds(), updatedTelemetry); - } - }); - } catch (Exception e) { - log.trace("Failed to process telemetry update msg: [{}]", proto, e); - } - } - - private void executeTelemetryUpdate(CalculatedFieldCtx cfCtx, EntityId entityId, List previousCalculatedFieldIds, Map updatedTelemetry) { - log.info("Received telemetry update msg: tenantId=[{}], entityId=[{}], calculatedFieldId=[{}]", cfCtx.getTenantId(), entityId, cfCtx.getCfId()); - Map argumentValues = updatedTelemetry.entrySet().stream() - .collect(Collectors.toMap(Entry::getKey, entry -> ArgumentEntry.createSingleValueArgument(entry.getValue()))); - -// updateOrInitializeState(cfCtx, entityId, argumentValues, previousCalculatedFieldIds); - } - - @Override - public void onEntityUpdateMsg(CalculatedFieldEntityUpdateMsgProto proto, TbCallback callback) { - try { - TenantId tenantId = TenantId.fromUUID(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())); - EntityId entityId = EntityIdFactory.getByTypeAndUuid(proto.getEntityType(), new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB())); - - TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tenantId, entityId); - if (tpi.isMyPartition()) { - log.info("Received CalculatedFieldEntityUpdateMsgProto for processing: tenantId=[{}], entityId=[{}]", tenantId, entityId); - if (proto.getDeleted()) { - log.info("Executing CalculatedFieldEntityUpdateMsgProto msg: entity deleted from profile, tenantId=[{}], entityId=[{}]", tenantId, entityId); - - EntityId oldProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getOldProfileIdMSB(), proto.getOldProfileIdLSB())); - calculatedFieldCache.getCalculatedFieldsByEntityId(entityId).forEach(cf -> clearState(cf.getId(), entityId)); - calculatedFieldCache.getCalculatedFieldsByEntityId(oldProfileId).forEach(cf -> clearState(cf.getId(), entityId)); - } - if (proto.getAdded()) { - log.info("Executing CalculatedFieldEntityUpdateMsgProto msg: entity added to profile, tenantId=[{}], entityId=[{}]", tenantId, entityId); - - EntityId newProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getNewProfileIdMSB(), proto.getNewProfileIdLSB())); - initializeStateForEntityByProfile(entityId, newProfileId, callback); - } - if (proto.getUpdated()) { - log.info("Executing CalculatedFieldEntityUpdateMsgProto msg: entity changed the profile, tenantId=[{}], entityId=[{}]", tenantId, entityId); - - EntityId oldProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getOldProfileIdMSB(), proto.getOldProfileIdLSB())); - EntityId newProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getNewProfileIdMSB(), proto.getNewProfileIdLSB())); - - calculatedFieldCache.getCalculatedFieldsByEntityId(oldProfileId).forEach(cf -> clearState(cf.getId(), entityId)); - initializeStateForEntityByProfile(entityId, newProfileId, callback); - } - } else { - clusterService.pushNotificationToCalculatedFields(tenantId, entityId, ToCalculatedFieldNotificationMsg.newBuilder().setEntityUpdateMsg(proto).build(), null); - } - } catch (Exception e) { - log.trace("Failed to process entity update msg: [{}]", proto, e); - } - } +// @Override +// public void onEntityUpdateMsg(CalculatedFieldEntityUpdateMsgProto proto, TbCallback callback) { +// try { +// TenantId tenantId = TenantId.fromUUID(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())); +// EntityId entityId = EntityIdFactory.getByTypeAndUuid(proto.getEntityType(), new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB())); +// +// TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tenantId, entityId); +// if (tpi.isMyPartition()) { +// log.info("Received CalculatedFieldEntityUpdateMsgProto for processing: tenantId=[{}], entityId=[{}]", tenantId, entityId); +// if (proto.getDeleted()) { +// log.info("Executing CalculatedFieldEntityUpdateMsgProto msg: entity deleted from profile, tenantId=[{}], entityId=[{}]", tenantId, entityId); +// +// EntityId oldProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getOldProfileIdMSB(), proto.getOldProfileIdLSB())); +// calculatedFieldCache.getCalculatedFieldsByEntityId(entityId).forEach(cf -> clearState(cf.getId(), entityId)); +// calculatedFieldCache.getCalculatedFieldsByEntityId(oldProfileId).forEach(cf -> clearState(cf.getId(), entityId)); +// } +// if (proto.getAdded()) { +// log.info("Executing CalculatedFieldEntityUpdateMsgProto msg: entity added to profile, tenantId=[{}], entityId=[{}]", tenantId, entityId); +// +// EntityId newProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getNewProfileIdMSB(), proto.getNewProfileIdLSB())); +// initializeStateForEntityByProfile(entityId, newProfileId, callback); +// } +// if (proto.getUpdated()) { +// log.info("Executing CalculatedFieldEntityUpdateMsgProto msg: entity changed the profile, tenantId=[{}], entityId=[{}]", tenantId, entityId); +// +// EntityId oldProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getOldProfileIdMSB(), proto.getOldProfileIdLSB())); +// EntityId newProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getNewProfileIdMSB(), proto.getNewProfileIdLSB())); +// +// calculatedFieldCache.getCalculatedFieldsByEntityId(oldProfileId).forEach(cf -> clearState(cf.getId(), entityId)); +// initializeStateForEntityByProfile(entityId, newProfileId, callback); +// } +// } else { +// clusterService.pushNotificationToCalculatedFields(tenantId, entityId, ToCalculatedFieldNotificationMsg.newBuilder().setEntityUpdateMsg(proto).build(), null); +// } +// } catch (Exception e) { +// log.trace("Failed to process entity update msg: [{}]", proto, e); +// } +// } private void clearState(CalculatedFieldId calculatedFieldId, EntityId entityId) { log.warn("Executing clearState, calculatedFieldId=[{}], entityId=[{}]", calculatedFieldId, entityId); 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 4f9a998937..c670e97580 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 @@ -26,6 +26,6 @@ public interface CalculatedFieldStateService { void persistState(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback); - void removeState(CalculatedFieldEntityCtxId ctxId); + void removeState(CalculatedFieldEntityCtxId stateId, TbCallback callback); } 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 33421a7d10..caeeabea52 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 @@ -192,4 +192,12 @@ public class CalculatedFieldCtx { public CalculatedFieldEntityCtxId toCalculatedFieldEntityCtxId() { return new CalculatedFieldEntityCtxId(tenantId, cfId, entityId); } + + public boolean hasSignificantChanges(CalculatedFieldCtx other) { + boolean entityIdChanged = !entityId.equals(other.entityId); + boolean typeChanged = !cfType.equals(other.cfType); + boolean argumentsChanged = !arguments.equals(other.arguments); + return entityIdChanged || typeChanged || argumentsChanged; + } + } 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/RocksDBStateService.java index 4de15f70bf..eaa1b08f5e 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/RocksDBStateService.java @@ -66,8 +66,9 @@ public class RocksDBStateService implements CalculatedFieldStateService { } @Override - public void removeState(CalculatedFieldEntityCtxId ctxId) { + public void removeState(CalculatedFieldEntityCtxId ctxId, TbCallback callback) { rocksDBService.delete(JacksonUtil.writeValueAsString(ctxId)); + callback.onSuccess(); } private CalculatedFieldEntityCtxIdProto toProto(CalculatedFieldEntityCtxId ctxId) { 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 95b340c362..d40bf7b130 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 @@ -56,6 +56,7 @@ import org.thingsboard.server.dao.eventsourcing.ActionEntityEvent; import org.thingsboard.server.dao.eventsourcing.DeleteEntityEvent; import org.thingsboard.server.dao.eventsourcing.SaveEntityEvent; import org.thingsboard.server.dao.tenant.TenantService; +import org.thingsboard.server.queue.TbQueueCallback; import java.util.Set; @@ -124,7 +125,7 @@ public class EntityStateSourcingListener { tbClusterService.onApiStateChange(apiUsageState, null); } case CALCULATED_FIELD -> { - onCalculatedFieldUpdate(event.getEntity(), event.getOldEntity(), lifecycleEvent); + onCalculatedFieldUpdate(event.getEntity(), event.getOldEntity()); } default -> { } @@ -193,8 +194,7 @@ public class EntityStateSourcingListener { } case CALCULATED_FIELD -> { CalculatedField calculatedField = (CalculatedField) event.getEntity(); - ComponentLifecycleMsg lifecycleMsg = new ComponentLifecycleMsg(calculatedField.getTenantId(), calculatedField.getId(), ComponentLifecycleEvent.DELETED); - tbClusterService.onCalculatedFieldDeleted(calculatedField.getTenantId(), calculatedField, lifecycleMsg); + tbClusterService.onCalculatedFieldDeleted(calculatedField, null); } default -> { } @@ -276,13 +276,13 @@ public class EntityStateSourcingListener { } } - private void onCalculatedFieldUpdate(Object entity, Object oldEntity, ComponentLifecycleEvent lifecycleEvent) { + private void onCalculatedFieldUpdate(Object entity, Object oldEntity) { CalculatedField calculatedField = (CalculatedField) entity; CalculatedField oldCalculatedField = null; if (oldEntity instanceof CalculatedField) { oldCalculatedField = (CalculatedField) oldEntity; } - tbClusterService.onCalculatedFieldUpdated(calculatedField, oldCalculatedField, new ComponentLifecycleMsg(calculatedField.getTenantId(), calculatedField.getId(), lifecycleEvent)); + tbClusterService.onCalculatedFieldUpdated(calculatedField, oldCalculatedField, TbQueueCallback.EMPTY); } private void pushAssignedFromNotification(Tenant currentTenant, TenantId newTenantId, Device assignedDevice) { 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 0fcc04bd5d..3d8bbddc80 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.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListeningExecutorService; import com.google.common.util.concurrent.MoreExecutors; import jakarta.annotation.PostConstruct; @@ -25,20 +24,17 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.ApplicationEventPublisher; import org.springframework.stereotype.Service; -import org.thingsboard.common.util.DonAsynchron; import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.actors.calculatedField.CalculatedFieldLinkedTelemetryMsg; import org.thingsboard.server.actors.calculatedField.CalculatedFieldTelemetryMsg; -import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.queue.QueueConfig; +import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TbCallback; -import org.thingsboard.server.common.util.ProtoUtils; import org.thingsboard.server.dao.tenant.TbTenantProfileCache; -import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityUpdateMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldLinkedTelemetryMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.ComponentLifecycleMsgProto; @@ -70,6 +66,8 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; +import static org.thingsboard.server.common.util.ProtoUtils.fromProto; + @Service @TbRuleEngineComponent @Slf4j @@ -86,8 +84,6 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer private final TbRuleEngineQueueFactory queueFactory; - private final CalculatedFieldExecutionService calculatedFieldExecutionService; - private MainQueueConsumerManager, CalculatedFieldQueueConfig> mainConsumer; private volatile ListeningExecutorService calculatedFieldsExecutor; @@ -101,12 +97,10 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer PartitionService partitionService, ApplicationEventPublisher eventPublisher, JwtSettingsService jwtSettingsService, - CalculatedFieldExecutionService calculatedFieldExecutionService, CalculatedFieldCache calculatedFieldCache) { super(actorContext, tenantProfileCache, deviceProfileCache, assetProfileCache, calculatedFieldCache, apiUsageStateService, partitionService, eventPublisher, jwtSettingsService); this.queueFactory = tbQueueFactory; - this.calculatedFieldExecutionService = calculatedFieldExecutionService; } @PostConstruct @@ -168,12 +162,7 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer forwardToActorSystem(toCfMsg.getLinkedTelemetryMsg(), callback); } else if (toCfMsg.hasComponentLifecycleMsg()) { log.trace("[{}] Forwarding component lifecycle message for processing {}", id, toCfMsg.getComponentLifecycleMsg()); - /// TODO: forward to Actor system - forwardToCalculatedFieldService(toCfMsg.getComponentLifecycleMsg(), callback); - } else if (toCfMsg.hasEntityUpdateMsg()) { - log.trace("[{}] Forwarding entity update message for processing {}", id, toCfMsg.getEntityUpdateMsg()); - /// TODO: forward to Actor system - forwardToCalculatedFieldService(toCfMsg.getEntityUpdateMsg(), callback); + forwardToActorSystem(toCfMsg.getComponentLifecycleMsg(), callback); } } catch (Throwable e) { log.warn("[{}] Failed to process message: {}", id, msg, e); @@ -222,14 +211,10 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer @Override protected void handleNotification(UUID id, TbProtoQueueMsg msg, TbCallback callback) { ToCalculatedFieldNotificationMsg toCfNotification = msg.getValue(); - if (toCfNotification.hasComponentLifecycle()) { + if (toCfNotification.hasComponentLifecycleMsg()) { // from upstream (maybe removed since we don't need to init state for each partition) - forwardToActorSystem(toCfNotification.getComponentLifecycle(), callback); - handleComponentLifecycleMsg(id, ProtoUtils.fromProto(toCfNotification.getComponentLifecycle())); - } else if (toCfNotification.hasEntityUpdateMsg()) { - processEntityUpdateMsg(toCfNotification.getEntityUpdateMsg()); - // from upstream (maybe removed since we don't need to update state for each partition) - forwardToActorSystem(toCfNotification.getEntityUpdateMsg(), callback); + log.trace("[{}] Forwarding component lifecycle message for processing {}", id, toCfNotification.getComponentLifecycleMsg()); + forwardToActorSystem(toCfNotification.getComponentLifecycleMsg(), callback); } else if (toCfNotification.hasLinkedTelemetryMsg()) { forwardToActorSystem(toCfNotification.getLinkedTelemetryMsg(), callback); } @@ -248,74 +233,9 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer actorContext.tell(new CalculatedFieldLinkedTelemetryMsg(tenantId, entityId, linkedMsg, callback)); } - private void forwardToCalculatedFieldService(ComponentLifecycleMsgProto msg, TbCallback callback) { - var tenantId = toTenantId(msg.getTenantIdMSB(), msg.getTenantIdLSB()); - var calculatedFieldId = new CalculatedFieldId(new UUID(msg.getEntityIdMSB(), msg.getEntityIdLSB())); - ListenableFuture future = calculatedFieldsExecutor.submit(() -> calculatedFieldExecutionService.onCalculatedFieldLifecycleMsg(msg, callback)); - DonAsynchron.withCallback(future, - __ -> callback.onSuccess(), - t -> { - log.warn("[{}] Failed to process calculated field message for calculated field [{}]", tenantId.getId(), calculatedFieldId.getId(), t); - callback.onFailure(t); - }); - } - - private void forwardToActorSystem(ComponentLifecycleMsgProto msg, TbCallback callback) { - var tenantId = toTenantId(msg.getTenantIdMSB(), msg.getTenantIdLSB()); - var calculatedFieldId = new CalculatedFieldId(new UUID(msg.getEntityIdMSB(), msg.getEntityIdLSB())); - ListenableFuture future = calculatedFieldsExecutor.submit(() -> calculatedFieldExecutionService.onCalculatedFieldLifecycleMsg(msg, callback)); - DonAsynchron.withCallback(future, - __ -> callback.onSuccess(), - t -> { - log.warn("[{}] Failed to process calculated field message for calculated field [{}]", tenantId.getId(), calculatedFieldId.getId(), t); - callback.onFailure(t); - }); - } - - private void forwardToCalculatedFieldService(CalculatedFieldEntityUpdateMsgProto msg, TbCallback callback) { - var tenantId = toTenantId(msg.getTenantIdMSB(), msg.getTenantIdLSB()); - var entityId = EntityIdFactory.getByTypeAndUuid(msg.getEntityType(), new UUID(msg.getEntityIdMSB(), msg.getEntityIdLSB())); - ListenableFuture future = calculatedFieldsExecutor.submit(() -> calculatedFieldExecutionService.onEntityUpdateMsg(msg, callback)); - DonAsynchron.withCallback(future, - __ -> callback.onSuccess(), - t -> { - log.warn("[{}] Failed to process entity updated message for entity [{}]", tenantId.getId(), entityId.getId(), t); - callback.onFailure(t); - }); - } - - private void forwardToActorSystem(CalculatedFieldEntityUpdateMsgProto msg, TbCallback callback) { - var tenantId = toTenantId(msg.getTenantIdMSB(), msg.getTenantIdLSB()); - var entityId = EntityIdFactory.getByTypeAndUuid(msg.getEntityType(), new UUID(msg.getEntityIdMSB(), msg.getEntityIdLSB())); - ListenableFuture future = calculatedFieldsExecutor.submit(() -> calculatedFieldExecutionService.onEntityUpdateMsg(msg, callback)); - DonAsynchron.withCallback(future, - __ -> callback.onSuccess(), - t -> { - log.warn("[{}] Failed to process entity updated message for entity [{}]", tenantId.getId(), entityId.getId(), t); - callback.onFailure(t); - }); - } - - private void processEntityUpdateMsg(CalculatedFieldEntityUpdateMsgProto entityUpdateMsg) { - var tenantId = toTenantId(entityUpdateMsg.getTenantIdMSB(), entityUpdateMsg.getTenantIdLSB()); - var entityId = EntityIdFactory.getByTypeAndUuid(entityUpdateMsg.getEntityType(), new UUID(entityUpdateMsg.getEntityIdMSB(), entityUpdateMsg.getEntityIdLSB())); - if (entityUpdateMsg.getAdded()) { - var newProfile = EntityIdFactory.getByTypeAndUuid(entityUpdateMsg.getEntityProfileType(), new UUID(entityUpdateMsg.getNewProfileIdMSB(), entityUpdateMsg.getNewProfileIdLSB())); - calculatedFieldCache.getEntitiesByProfile(tenantId, newProfile).add(entityId); - } else if (entityUpdateMsg.getDeleted()) { - var oldProfile = EntityIdFactory.getByTypeAndUuid(entityUpdateMsg.getEntityProfileType(), new UUID(entityUpdateMsg.getOldProfileIdMSB(), entityUpdateMsg.getOldProfileIdLSB())); - calculatedFieldCache.getEntitiesByProfile(tenantId, oldProfile).remove(entityId); - } else if (entityUpdateMsg.getUpdated()) { - var oldProfile = EntityIdFactory.getByTypeAndUuid(entityUpdateMsg.getEntityProfileType(), new UUID(entityUpdateMsg.getOldProfileIdMSB(), entityUpdateMsg.getOldProfileIdLSB())); - var newProfile = EntityIdFactory.getByTypeAndUuid(entityUpdateMsg.getEntityProfileType(), new UUID(entityUpdateMsg.getNewProfileIdMSB(), entityUpdateMsg.getNewProfileIdLSB())); - calculatedFieldCache.getEntitiesByProfile(tenantId, oldProfile).remove(entityId); - calculatedFieldCache.getEntitiesByProfile(tenantId, newProfile).add(entityId); - } - } - - private void throwNotHandled(Object msg, TbCallback callback) { - log.warn("Message not handled: {}", msg); - callback.onFailure(new RuntimeException("Message not handled!")); + private void forwardToActorSystem(ComponentLifecycleMsgProto proto, TbCallback callback) { + var msg = fromProto(proto); + actorContext.tell(new CalculatedFieldEntityLifecycleMsg(msg.getTenantId(), msg, callback)); } private TenantId toTenantId(long tenantIdMSB, long tenantIdLSB) { 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 2783e826fe..b272694b16 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 @@ -70,7 +70,6 @@ import org.thingsboard.server.common.msg.rule.engine.DeviceNameOrTypeUpdateMsg; import org.thingsboard.server.common.util.ProtoUtils; import org.thingsboard.server.dao.edge.EdgeService; import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityUpdateMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.ComponentLifecycleMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.DeviceStateServiceMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.EdgeNotificationMsgProto; @@ -362,8 +361,7 @@ public class DefaultTbClusterService implements TbClusterService { @Override public void pushMsgToCalculatedFields(TenantId tenantId, EntityId entityId, ToCalculatedFieldMsg msg, TbQueueCallback callback) { TopicPartitionInfo tpi = partitionService.resolve(CALCULATED_FIELD_QUEUE_KEY, entityId); - producerProvider.getCalculatedFieldsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), msg), callback); - toRuleEngineMsgs.incrementAndGet(); + pushMsgToCalculatedFields(tpi, UUID.randomUUID(), msg, callback); } @Override @@ -435,7 +433,7 @@ public class DefaultTbClusterService implements TbClusterService { public void onDeviceDeleted(TenantId tenantId, Device device, TbQueueCallback callback) { DeviceId deviceId = device.getId(); gatewayNotificationsService.onDeviceDeleted(device); - handleCalculatedFieldEntityDeleted(tenantId, deviceId, device.getDeviceProfileId()); + handleCalculatedFieldEntityDeleted(tenantId, deviceId); broadcastEntityDeleteToTransport(tenantId, deviceId, device.getName(), callback); sendDeviceStateServiceEvent(tenantId, deviceId, false, false, true); broadcastEntityStateChangeEvent(tenantId, deviceId, ComponentLifecycleEvent.DELETED); @@ -444,7 +442,7 @@ public class DefaultTbClusterService implements TbClusterService { @Override public void onAssetDeleted(TenantId tenantId, Asset asset, TbQueueCallback callback) { AssetId assetId = asset.getId(); - handleCalculatedFieldEntityDeleted(tenantId, assetId, asset.getAssetProfileId()); + handleCalculatedFieldEntityDeleted(tenantId, assetId); broadcastEntityStateChangeEvent(tenantId, assetId, ComponentLifecycleEvent.DELETED); } @@ -597,9 +595,7 @@ public class DefaultTbClusterService implements TbClusterService { private void broadcast(ComponentLifecycleMsg msg) { ComponentLifecycleMsgProto componentLifecycleMsgProto = toProto(msg); TbQueueProducer> toRuleEngineProducer = producerProvider.getRuleEngineNotificationsMsgProducer(); - TbQueueProducer> toCalculatedFieldProducer = producerProvider.getCalculatedFieldsNotificationsMsgProducer(); Set tbRuleEngineServices = partitionService.getAllServiceIds(ServiceType.TB_RULE_ENGINE); - Set tbCalculatedFieldServices = new HashSet<>(tbRuleEngineServices); EntityType entityType = msg.getEntityId().getEntityType(); if (entityType.equals(EntityType.TENANT) || entityType.equals(EntityType.TENANT_PROFILE) @@ -622,14 +618,6 @@ public class DefaultTbClusterService implements TbClusterService { // No need to push notifications twice tbRuleEngineServices.removeAll(tbCoreServices); } - if (entityType.equals(EntityType.CALCULATED_FIELD)) { - for (String serviceId : tbCalculatedFieldServices) { - TopicPartitionInfo tpi = topicService.getCalculatedFieldNotificationsTopic(serviceId); - ToCalculatedFieldNotificationMsg toCfNotificationMsg = ToCalculatedFieldNotificationMsg.newBuilder().setComponentLifecycle(componentLifecycleMsgProto).build(); - toCalculatedFieldProducer.send(tpi, new TbProtoQueueMsg<>(msg.getEntityId().getId(), toCfNotificationMsg), null); - toRuleEngineNfs.incrementAndGet(); // TODO: add separate counter when we will have new ServiceType.CALCULATED_FIELDS - } - } for (String serviceId : tbRuleEngineServices) { TopicPartitionInfo tpi = topicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceId); ToRuleEngineNotificationMsg toRuleEngineMsg = ToRuleEngineNotificationMsg.newBuilder().setComponentLifecycle(componentLifecycleMsgProto).build(); @@ -669,56 +657,86 @@ public class DefaultTbClusterService implements TbClusterService { } @Override - public void onDeviceUpdated(Device device, Device old) { + public void onDeviceUpdated(Device entity, Device old) { var created = old == null; - broadcastEntityChangeToTransport(device.getTenantId(), device.getId(), device, null); + broadcastEntityChangeToTransport(entity.getTenantId(), entity.getId(), entity, null); if (old != null) { - boolean deviceNameChanged = !device.getName().equals(old.getName()); + boolean deviceNameChanged = !entity.getName().equals(old.getName()); if (deviceNameChanged) { - gatewayNotificationsService.onDeviceUpdated(device, old); + gatewayNotificationsService.onDeviceUpdated(entity, old); } - boolean deviceProfileChanged = !device.getDeviceProfileId().equals(old.getDeviceProfileId()); + boolean deviceProfileChanged = !entity.getDeviceProfileId().equals(old.getDeviceProfileId()); if (deviceProfileChanged) { - handleCalculatedFieldEntityUpdated(device.getTenantId(), device.getId(), old.getDeviceProfileId(), device.getDeviceProfileId()); + ComponentLifecycleMsg msg = ComponentLifecycleMsg.builder() + .tenantId(entity.getTenantId()) + .entityId(entity.getId()) + .event(ComponentLifecycleEvent.UPDATED) + .oldProfileId(old.getDeviceProfileId()) + .profileId(entity.getDeviceProfileId()) + .oldName(old.getName()) + .name(entity.getName()) + .build(); + pushMsgToCalculatedFields(entity.getTenantId(), entity.getId(), ToCalculatedFieldMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY); } if (deviceNameChanged || deviceProfileChanged) { - pushMsgToCore(new DeviceNameOrTypeUpdateMsg(device.getTenantId(), device.getId(), device.getName(), device.getType()), null); + pushMsgToCore(new DeviceNameOrTypeUpdateMsg(entity.getTenantId(), entity.getId(), entity.getName(), entity.getType()), null); } } else { - handleCalculatedFieldEntityAdded(device.getTenantId(), device.getId(), device.getDeviceProfileId()); + ComponentLifecycleMsg msg = ComponentLifecycleMsg.builder() + .tenantId(entity.getTenantId()) + .entityId(entity.getId()) + .event(ComponentLifecycleEvent.CREATED) + .profileId(entity.getDeviceProfileId()) + .name(entity.getName()) + .build(); + pushMsgToCalculatedFields(entity.getTenantId(), entity.getId(), ToCalculatedFieldMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY); } - broadcastEntityStateChangeEvent(device.getTenantId(), device.getId(), created ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED); - sendDeviceStateServiceEvent(device.getTenantId(), device.getId(), created, !created, false); - otaPackageStateService.update(device, old); + broadcastEntityStateChangeEvent(entity.getTenantId(), entity.getId(), created ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED); + sendDeviceStateServiceEvent(entity.getTenantId(), entity.getId(), created, !created, false); + otaPackageStateService.update(entity, old); } @Override - public void onAssetUpdated(Asset asset, Asset old) { + public void onAssetUpdated(Asset entity, Asset old) { var created = old == null; - broadcastEntityChangeToTransport(asset.getTenantId(), asset.getId(), asset, null); + broadcastEntityChangeToTransport(entity.getTenantId(), entity.getId(), entity, null); if (old != null) { - boolean assetTypeChanged = !asset.getType().equals(old.getType()); + boolean assetTypeChanged = !entity.getAssetProfileId().equals(old.getAssetProfileId()); if (assetTypeChanged) { - handleCalculatedFieldEntityUpdated(asset.getTenantId(), asset.getId(), old.getAssetProfileId(), asset.getAssetProfileId()); + ComponentLifecycleMsg msg = ComponentLifecycleMsg.builder() + .tenantId(entity.getTenantId()) + .entityId(entity.getId()) + .event(ComponentLifecycleEvent.UPDATED) + .oldProfileId(old.getAssetProfileId()) + .profileId(entity.getAssetProfileId()) + .oldName(old.getName()) + .name(entity.getName()) + .build(); + pushMsgToCalculatedFields(entity.getTenantId(), entity.getId(), ToCalculatedFieldMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY); } } else { - handleCalculatedFieldEntityAdded(asset.getTenantId(), asset.getId(), asset.getAssetProfileId()); + ComponentLifecycleMsg msg = ComponentLifecycleMsg.builder() + .tenantId(entity.getTenantId()) + .entityId(entity.getId()) + .event(ComponentLifecycleEvent.CREATED) + .profileId(entity.getAssetProfileId()) + .name(entity.getName()) + .build(); + pushMsgToCalculatedFields(entity.getTenantId(), entity.getId(), ToCalculatedFieldMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY); } - broadcastEntityStateChangeEvent(asset.getTenantId(), asset.getId(), created ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED); + broadcastEntityStateChangeEvent(entity.getTenantId(), entity.getId(), created ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED); } @Override - public void onCalculatedFieldUpdated(CalculatedField calculatedField, CalculatedField oldCalculatedField, ComponentLifecycleMsg lifecycleMsg) { - var created = oldCalculatedField == null; - calculatedFieldExecutionService.pushCalculatedFieldLifecycleMsgToQueue(calculatedField, toProto(lifecycleMsg)); - broadcastEntityStateChangeEvent(calculatedField.getTenantId(), calculatedField.getId(), created ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED); + public void onCalculatedFieldUpdated(CalculatedField calculatedField, CalculatedField oldCalculatedField, TbQueueCallback callback) { + var msg = new ComponentLifecycleMsg(calculatedField.getTenantId(), calculatedField.getEntityId(), oldCalculatedField == null ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED); + broadcastToCalculatedFields(ToCalculatedFieldNotificationMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), callback); } @Override - public void onCalculatedFieldDeleted(TenantId tenantId, CalculatedField calculatedField, ComponentLifecycleMsg lifecycleMsg) { - CalculatedFieldId calculatedFieldId = calculatedField.getId(); - calculatedFieldExecutionService.pushCalculatedFieldLifecycleMsgToQueue(calculatedField, toProto(lifecycleMsg)); - broadcastEntityStateChangeEvent(tenantId, calculatedFieldId, ComponentLifecycleEvent.DELETED); + public void onCalculatedFieldDeleted(CalculatedField calculatedField, TbQueueCallback callback) { + var msg = new ComponentLifecycleMsg(calculatedField.getTenantId(), calculatedField.getEntityId(), ComponentLifecycleEvent.DELETED); + broadcastToCalculatedFields(ToCalculatedFieldNotificationMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), callback); } @Override @@ -849,54 +867,8 @@ public class DefaultTbClusterService implements TbClusterService { } } - private void handleCalculatedFieldEntityAdded(TenantId tenantId, EntityId entityId, EntityId newProfileId) { - handleCalculatedFieldEntityUpdateEvent(tenantId, entityId, null, newProfileId, true, false, false); + 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); } - - private void handleCalculatedFieldEntityUpdated(TenantId tenantId, EntityId entityId, EntityId oldProfileId, EntityId newProfileId) { - handleCalculatedFieldEntityUpdateEvent(tenantId, entityId, oldProfileId, newProfileId, false, true, true); - } - - private void handleCalculatedFieldEntityDeleted(TenantId tenantId, EntityId entityId, EntityId oldProfileId) { - handleCalculatedFieldEntityUpdateEvent(tenantId, entityId, oldProfileId, null, false, false, true); - } - - private void handleCalculatedFieldEntityUpdateEvent(TenantId tenantId, EntityId entityId, EntityId oldProfileId, EntityId newProfileId, boolean added, boolean updated, boolean deleted) { - CalculatedFieldEntityUpdateMsgProto.Builder builder = CalculatedFieldEntityUpdateMsgProto.newBuilder(); - builder.setTenantIdMSB(tenantId.getId().getMostSignificantBits()); - builder.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()); - builder.setEntityType(entityId.getEntityType().name()); - builder.setEntityIdMSB(entityId.getId().getMostSignificantBits()); - builder.setEntityIdLSB(entityId.getId().getLeastSignificantBits()); - if (oldProfileId != null) { - builder.setEntityProfileType(oldProfileId.getEntityType().name()); - builder.setOldProfileIdMSB(oldProfileId.getId().getMostSignificantBits()); - builder.setOldProfileIdLSB(oldProfileId.getId().getLeastSignificantBits()); - } - if (newProfileId != null) { - builder.setEntityProfileType(newProfileId.getEntityType().name()); - builder.setNewProfileIdMSB(newProfileId.getId().getMostSignificantBits()); - builder.setNewProfileIdLSB(newProfileId.getId().getLeastSignificantBits()); - } - builder.setAdded(added); - builder.setUpdated(updated); - builder.setDeleted(deleted); - CalculatedFieldEntityUpdateMsgProto msg = builder.build(); - - broadcastEntityUpdateEvent(msg); - pushMsgToCalculatedFields(tenantId, entityId, ToCalculatedFieldMsg.newBuilder().setEntityUpdateMsg(msg).build(), null); - } - - private void broadcastEntityUpdateEvent(CalculatedFieldEntityUpdateMsgProto proto) { - TbQueueProducer> toCalculatedFieldProducer = producerProvider.getCalculatedFieldsNotificationsMsgProducer(); - Set tbRuleEngineServices = partitionService.getAllServiceIds(ServiceType.TB_RULE_ENGINE); - Set tbCalculatedFieldServices = new HashSet<>(tbRuleEngineServices); - for (String serviceId : tbCalculatedFieldServices) { - TopicPartitionInfo tpi = topicService.getCalculatedFieldNotificationsTopic(serviceId); - ToCalculatedFieldNotificationMsg toCfNotificationMsg = ToCalculatedFieldNotificationMsg.newBuilder().setEntityUpdateMsg(proto).build(); - toCalculatedFieldProducer.send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), toCfNotificationMsg), null); - toRuleEngineNfs.incrementAndGet(); // TODO: add separate counter when we will have new ServiceType.CALCULATED_FIELDS - } - } - } diff --git a/common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java b/common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java index 588b135891..3d81de1cc4 100644 --- a/common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java +++ b/common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java @@ -131,8 +131,8 @@ public interface TbClusterService extends TbQueueClusterService { void sendNotificationMsgToEdge(TenantId tenantId, EdgeId edgeId, EntityId entityId, String body, EdgeEventType type, EdgeEventActionType action, EdgeId sourceEdgeId); - void onCalculatedFieldUpdated(CalculatedField calculatedField, CalculatedField oldCalculatedField, ComponentLifecycleMsg lifecycleMsg); + void onCalculatedFieldUpdated(CalculatedField calculatedField, CalculatedField oldCalculatedField, TbQueueCallback callback); - void onCalculatedFieldDeleted(TenantId tenantId, CalculatedField calculatedField, ComponentLifecycleMsg lifecycleMsg); + void onCalculatedFieldDeleted(CalculatedField calculatedField, TbQueueCallback callback); } diff --git a/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueCallback.java b/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueCallback.java index 7943c9d43d..9e8f8e4a0a 100644 --- a/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueCallback.java +++ b/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueCallback.java @@ -15,8 +15,22 @@ */ package org.thingsboard.server.queue; + public interface TbQueueCallback { + TbQueueCallback EMPTY = new TbQueueCallback() { + + @Override + public void onSuccess(TbQueueMsgMetadata metadata) { + + } + + @Override + public void onFailure(Throwable t) { + + } + }; + void onSuccess(TbQueueMsgMetadata metadata); void onFailure(Throwable t); diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java index 91419bee9c..e13e2d2ecb 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java @@ -135,14 +135,18 @@ public enum MsgType { EDGE_SYNC_REQUEST_TO_EDGE_SESSION_MSG, EDGE_SYNC_RESPONSE_FROM_EDGE_SESSION_MSG, + CF_INIT_MSG, // Sent to init particular calculated field; CF_LINK_INIT_MSG, // Sent to init particular calculated field; - CF_STATE_RESTORE_MSG,// Sent to init particular calculated field entity state; - CF_TELEMETRY_MSG, + CF_STATE_RESTORE_MSG, // Sent to restore particular calculated field entity state; + CF_ENTITY_LIFECYCLE_MSG, // Sent on CF/Device/Asset create/update/delete; + CF_TELEMETRY_MSG, // Sent from queue to actor system; + CF_LINKED_TELEMETRY_MSG, // Sent from queue to actor system; + + /* CF Manager Actor -> CF Entity actor */ CF_ENTITY_TELEMETRY_MSG, - CF_LINKED_TELEMETRY_MSG, - CF_UPDATE_MSG, - CF_ENTITY_UPDATE_MSG; + CF_ENTITY_INIT_CF_MSG, + CF_ENTITY_DELETE_MSG; @Getter private final boolean ignoreOnStart; diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/cf/CalculatedFieldEntityLifecycleMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/cf/CalculatedFieldEntityLifecycleMsg.java new file mode 100644 index 0000000000..51eea1f943 --- /dev/null +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/cf/CalculatedFieldEntityLifecycleMsg.java @@ -0,0 +1,37 @@ +/** + * 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.common.msg.cf; + +import lombok.Data; +import org.thingsboard.server.common.data.cf.CalculatedField; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.msg.MsgType; +import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg; +import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; +import org.thingsboard.server.common.msg.queue.TbCallback; + +@Data +public class CalculatedFieldEntityLifecycleMsg implements ToCalculatedFieldSystemMsg { + + private final TenantId tenantId; + private final ComponentLifecycleMsg data; + private final TbCallback callback; + + @Override + public MsgType getMsgType() { + return MsgType.CF_ENTITY_LIFECYCLE_MSG; + } +} diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/plugin/ComponentLifecycleMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/plugin/ComponentLifecycleMsg.java index 000733cfd8..208b6ab2f5 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/plugin/ComponentLifecycleMsg.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/plugin/ComponentLifecycleMsg.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.common.msg.plugin; +import lombok.Builder; import lombok.Data; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.id.EntityId; @@ -37,6 +38,25 @@ public class ComponentLifecycleMsg implements TenantAwareMsg, ToAllNodesMsg { private final TenantId tenantId; private final EntityId entityId; private final ComponentLifecycleEvent event; + private final String oldName; + private final String name; + private final EntityId oldProfileId; + private final EntityId profileId; + + public ComponentLifecycleMsg(TenantId tenantId, EntityId entityId, ComponentLifecycleEvent event) { + this(tenantId, entityId, event, null, null, null, null); + } + + @Builder + private ComponentLifecycleMsg(TenantId tenantId, EntityId entityId, ComponentLifecycleEvent event, String oldName, String name, EntityId oldProfileId, EntityId profileId) { + this.tenantId = tenantId; + this.entityId = entityId; + this.event = event; + this.oldName = oldName; + this.name = name; + this.oldProfileId = oldProfileId; + this.profileId = profileId; + } public Optional getRuleChainId() { return entityId.getEntityType() == EntityType.RULE_CHAIN ? Optional.of((RuleChainId) entityId) : Optional.empty(); diff --git a/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java b/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java index 91e3276210..b81af2bfb9 100644 --- a/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java +++ b/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java @@ -115,14 +115,28 @@ public class ProtoUtils { } public static TransportProtos.ComponentLifecycleMsgProto toProto(ComponentLifecycleMsg msg) { - return TransportProtos.ComponentLifecycleMsgProto.newBuilder() + var builder = TransportProtos.ComponentLifecycleMsgProto.newBuilder() .setTenantIdMSB(msg.getTenantId().getId().getMostSignificantBits()) .setTenantIdLSB(msg.getTenantId().getId().getLeastSignificantBits()) .setEntityType(toProto(msg.getEntityId().getEntityType())) .setEntityIdMSB(msg.getEntityId().getId().getMostSignificantBits()) .setEntityIdLSB(msg.getEntityId().getId().getLeastSignificantBits()) - .setEvent(TransportProtos.ComponentLifecycleEvent.forNumber(msg.getEvent().ordinal())) - .build(); + .setEvent(TransportProtos.ComponentLifecycleEvent.forNumber(msg.getEvent().ordinal())); + if (msg.getProfileId() != null) { + builder.setProfileIdMSB(msg.getProfileId().getId().getMostSignificantBits()); + builder.setProfileIdLSB(msg.getProfileId().getId().getLeastSignificantBits()); + } + if (msg.getOldProfileId() != null) { + builder.setProfileIdMSB(msg.getOldProfileId().getId().getMostSignificantBits()); + builder.setProfileIdLSB(msg.getOldProfileId().getId().getLeastSignificantBits()); + } + if (msg.getName() != null) { + builder.setName(msg.getName()); + } + if (msg.getOldName() != null) { + builder.setName(msg.getOldName()); + } + return builder.build(); } public static TransportProtos.EntityTypeProto toProto(EntityType entityType) { @@ -130,11 +144,22 @@ public class ProtoUtils { } public static ComponentLifecycleMsg fromProto(TransportProtos.ComponentLifecycleMsgProto proto) { - return new ComponentLifecycleMsg( - TenantId.fromUUID(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())), - EntityIdFactory.getByTypeAndUuid(fromProto(proto.getEntityType()), new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB())), - ComponentLifecycleEvent.values()[proto.getEventValue()] - ); + EntityId entityId = EntityIdFactory.getByTypeAndUuid(fromProto(proto.getEntityType()), new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB())); + var builder = ComponentLifecycleMsg.builder() + .tenantId(TenantId.fromUUID(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB()))) + .entityId(entityId) + .event(ComponentLifecycleEvent.values()[proto.getEventValue()]) + .name(proto.getName()) + .oldName(proto.getOldName()); + if (proto.getProfileIdMSB() != 0 || proto.getProfileIdLSB() != 0) { + var profileType = EntityType.DEVICE.equals(entityId.getEntityType()) ? EntityType.DEVICE_PROFILE : EntityType.ASSET_PROFILE; + builder.profileId(EntityIdFactory.getByTypeAndUuid(profileType, new UUID(proto.getProfileIdMSB(), proto.getProfileIdLSB()))); + } + if (proto.getOldProfileIdMSB() != 0 || proto.getOldProfileIdLSB() != 0) { + var profileType = EntityType.DEVICE.equals(entityId.getEntityType()) ? EntityType.DEVICE_PROFILE : EntityType.ASSET_PROFILE; + builder.oldProfileId(EntityIdFactory.getByTypeAndUuid(profileType, new UUID(proto.getOldProfileIdMSB(), proto.getOldProfileIdLSB()))); + } + return builder.build(); } public static EntityType fromProto(TransportProtos.EntityTypeProto entityType) { diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 2ed535495a..6ee4108f9a 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -773,22 +773,6 @@ message DeviceInactivityProto { int64 lastInactivityTime = 5; } -message CalculatedFieldEntityUpdateMsgProto { - int64 tenantIdMSB = 1; - int64 tenantIdLSB = 2; - string entityType = 3; - int64 entityIdMSB = 4; - int64 entityIdLSB = 5; - string entityProfileType = 6; - int64 oldProfileIdMSB = 7; - int64 oldProfileIdLSB = 8; - int64 newProfileIdMSB = 9; - int64 newProfileIdLSB = 10; - bool added = 11; - bool updated = 12; - bool deleted = 13; -} - message CalculatedFieldTelemetryMsgProto { int64 tenantIdMSB = 1; int64 tenantIdLSB = 2; @@ -1205,6 +1189,13 @@ message ComponentLifecycleMsgProto { int64 entityIdMSB = 4; int64 entityIdLSB = 5; ComponentLifecycleEvent event = 6; + //Since 4.0. To replace the + string oldName = 7; + string name = 8; + int64 oldProfileIdMSB = 9; + int64 oldProfileIdLSB = 10; + int64 profileIdMSB = 11; + int64 profileIdLSB = 12; } message EdgeEventMsgProto { @@ -1632,16 +1623,14 @@ message ToEdgeEventNotificationMsg { } message ToCalculatedFieldMsg { - CalculatedFieldTelemetryMsgProto telemetryMsg = 1; - CalculatedFieldLinkedTelemetryMsgProto linkedTelemetryMsg = 2; - ComponentLifecycleMsgProto componentLifecycleMsg = 3; - CalculatedFieldEntityUpdateMsgProto entityUpdateMsg = 4; + ComponentLifecycleMsgProto componentLifecycleMsg = 1; + CalculatedFieldTelemetryMsgProto telemetryMsg = 2; + CalculatedFieldLinkedTelemetryMsgProto linkedTelemetryMsg = 3; } message ToCalculatedFieldNotificationMsg { - ComponentLifecycleMsgProto componentLifecycle = 1; - CalculatedFieldEntityUpdateMsgProto entityUpdateMsg = 2; - CalculatedFieldLinkedTelemetryMsgProto linkedTelemetryMsg = 3; + ComponentLifecycleMsgProto componentLifecycleMsg = 1; + CalculatedFieldLinkedTelemetryMsgProto linkedTelemetryMsg = 2; } /* Messages that are handled by ThingsBoard RuleEngine Service */