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 c8ccd91dbb..2c22941c2f 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -109,6 +109,7 @@ import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.service.apiusage.TbApiUsageStateService; import org.thingsboard.server.service.cf.CalculatedFieldProcessingService; +import org.thingsboard.server.service.cf.CalculatedFieldQueueService; import org.thingsboard.server.service.cf.CalculatedFieldStateService; import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache; import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; @@ -544,6 +545,11 @@ public class ActorSystemContext { @Getter private CalculatedFieldStateService calculatedFieldStateService; + @Lazy + @Autowired(required = false) + @Getter + private CalculatedFieldQueueService calculatedFieldQueueService; + @Lazy @Autowired(required = false) @Getter diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index 1cd0a4df63..ee6758df6a 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -28,6 +28,7 @@ import org.thingsboard.rule.engine.api.NotificationCenter; import org.thingsboard.rule.engine.api.RuleEngineAlarmService; import org.thingsboard.rule.engine.api.RuleEngineApiUsageStateService; import org.thingsboard.rule.engine.api.RuleEngineAssetProfileCache; +import org.thingsboard.rule.engine.api.RuleEngineCalculatedFieldQueueService; import org.thingsboard.rule.engine.api.RuleEngineDeviceProfileCache; import org.thingsboard.rule.engine.api.DeviceStateManager; import org.thingsboard.rule.engine.api.RuleEngineRpcService; @@ -902,6 +903,11 @@ public class DefaultTbContext implements TbContext { return mainCtx.getCalculatedFieldService(); } + @Override + public RuleEngineCalculatedFieldQueueService getCalculatedFieldQueueService() { + return mainCtx.getCalculatedFieldQueueService(); + } + @Override public boolean isExternalNodeForceAck() { return mainCtx.isExternalNodeForceAck(); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldQueueService.java b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldQueueService.java index b688993a8d..eb86220361 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldQueueService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldQueueService.java @@ -18,13 +18,14 @@ package org.thingsboard.server.service.cf; import com.google.common.util.concurrent.FutureCallback; import org.thingsboard.rule.engine.api.AttributesDeleteRequest; import org.thingsboard.rule.engine.api.AttributesSaveRequest; +import org.thingsboard.rule.engine.api.RuleEngineCalculatedFieldQueueService; import org.thingsboard.rule.engine.api.TimeseriesDeleteRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.common.data.kv.TimeseriesSaveResult; import java.util.List; -public interface CalculatedFieldQueueService { +public interface CalculatedFieldQueueService extends RuleEngineCalculatedFieldQueueService { /** * Filter CFs based on the request entity. Push to the queue if any matching CF exist; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java index 3ca0b220ff..ba4be6ace6 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java @@ -47,6 +47,7 @@ import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import org.thingsboard.server.service.profile.TbAssetProfileCache; import org.thingsboard.server.service.profile.TbDeviceProfileCache; +import java.util.Collections; import java.util.EnumSet; import java.util.List; import java.util.Set; @@ -89,6 +90,11 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS () -> toCalculatedFieldTelemetryMsgProto(request, result), callback); } + @Override + public void pushRequestToQueue(TimeseriesSaveRequest request, FutureCallback callback) { + pushRequestToQueue(request, null, callback); + } + @Override public void pushRequestToQueue(AttributesSaveRequest request, List result, FutureCallback callback) { var tenantId = request.getTenantId(); @@ -97,6 +103,11 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS () -> toCalculatedFieldTelemetryMsgProto(request, result), callback); } + @Override + public void pushRequestToQueue(AttributesSaveRequest request, FutureCallback callback) { + pushRequestToQueue(request, null, callback); + } + @Override public void pushRequestToQueue(AttributesDeleteRequest request, List result, FutureCallback callback) { var tenantId = request.getTenantId(); @@ -156,15 +167,19 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS ToCalculatedFieldMsg.Builder msg = ToCalculatedFieldMsg.newBuilder(); CalculatedFieldTelemetryMsgProto.Builder telemetryMsg = buildTelemetryMsgProto(request.getTenantId(), request.getEntityId(), request.getPreviousCalculatedFieldIds(), request.getTbMsgId(), request.getTbMsgType()); + List entries = request.getEntries(); - List versions = result.getVersions(); + List versions = result != null ? result.getVersions() : Collections.emptyList(); + for (int i = 0; i < entries.size(); i++) { - long tsVersion = versions.get(i); - TsKvProto tsProto = toTsKvProto(entries.get(i)).toBuilder().setVersion(tsVersion).build(); - telemetryMsg.addTsData(tsProto); + TsKvProto.Builder tsProtoBuilder = toTsKvProto(entries.get(i)).toBuilder(); + if (result != null) { + tsProtoBuilder.setVersion(versions.get(i)); + } + telemetryMsg.addTsData(tsProtoBuilder.build()); } - msg.setTelemetryMsg(telemetryMsg.build()); + msg.setTelemetryMsg(telemetryMsg.build()); return msg.build(); } @@ -175,9 +190,11 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS telemetryMsg.setScope(AttributeScopeProto.valueOf(request.getScope().name())); List entries = request.getEntries(); for (int i = 0; i < entries.size(); i++) { - long attrVersion = versions.get(i); - AttributeValueProto attrProto = ProtoUtils.toProto(entries.get(i)).toBuilder().setVersion(attrVersion).build(); - telemetryMsg.addAttrData(attrProto); + AttributeValueProto.Builder attrProtoBuilder = ProtoUtils.toProto(entries.get(i)).toBuilder(); + if (versions != null) { + attrProtoBuilder.setVersion(versions.get(i)); + } + telemetryMsg.addAttrData(attrProtoBuilder.build()); } msg.setTelemetryMsg(telemetryMsg.build()); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java index a237f3d022..bfe9eed24f 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java @@ -42,13 +42,17 @@ public class SingleValueArgumentEntry implements ArgumentEntry { public SingleValueArgumentEntry(TsKvProto entry) { this.ts = entry.getTs(); - this.version = entry.getVersion(); + if (entry.hasVersion()) { + this.version = entry.getVersion(); + } this.kvEntryValue = ProtoUtils.fromProto(entry.getKv()); } public SingleValueArgumentEntry(AttributeValueProto entry) { this.ts = entry.getLastUpdateTs(); - this.version = entry.getVersion(); + if (entry.hasVersion()) { + this.version = entry.getVersion(); + } this.kvEntryValue = ProtoUtils.basicKvEntryFromProto(entry); } diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineCalculatedFieldQueueService.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineCalculatedFieldQueueService.java new file mode 100644 index 0000000000..6ab40b79c2 --- /dev/null +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineCalculatedFieldQueueService.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2025 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.rule.engine.api; + +import com.google.common.util.concurrent.FutureCallback; + +public interface RuleEngineCalculatedFieldQueueService { + + void pushRequestToQueue(TimeseriesSaveRequest request, FutureCallback callback); + + void pushRequestToQueue(AttributesSaveRequest request, FutureCallback callback); + +} diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java index d0a0abf3d1..e473df4c54 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java @@ -360,6 +360,8 @@ public interface TbContext { CalculatedFieldService getCalculatedFieldService(); + RuleEngineCalculatedFieldQueueService getCalculatedFieldQueueService(); + boolean isExternalNodeForceAck(); /** diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbCalculatedFieldsNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbCalculatedFieldsNode.java new file mode 100644 index 0000000000..b2cce52c87 --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbCalculatedFieldsNode.java @@ -0,0 +1,125 @@ +/** + * Copyright © 2016-2025 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.rule.engine.telemetry; + +import com.google.gson.JsonParser; +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.rule.engine.api.AttributesSaveRequest; +import org.thingsboard.rule.engine.api.EmptyNodeConfiguration; +import org.thingsboard.rule.engine.api.RuleNode; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.rule.engine.api.TbNode; +import org.thingsboard.rule.engine.api.TbNodeConfiguration; +import org.thingsboard.rule.engine.api.TbNodeException; +import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; +import org.thingsboard.rule.engine.api.util.TbNodeUtils; +import org.thingsboard.server.common.adaptor.JsonConverter; +import org.thingsboard.server.common.data.AttributeScope; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.BasicTsKvEntry; +import org.thingsboard.server.common.data.kv.KvEntry; +import org.thingsboard.server.common.data.kv.TsKvEntry; +import org.thingsboard.server.common.data.plugin.ComponentType; +import org.thingsboard.server.common.msg.TbMsg; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +import static org.thingsboard.server.common.data.DataConstants.SCOPE; + +@Slf4j +@RuleNode( + type = ComponentType.ACTION, + name = "calculated fields", + configClazz = EmptyNodeConfiguration.class, + nodeDescription = "Pushes incoming messages to calculated fields service", + nodeDetails = "Node enables the processing of calculated fields without persisting incoming messages to the database. " + + "By default, the processing of calculated fields is triggered by the save attributes and save time series nodes. " + + "This rule node accepts the same messages as these nodes but allows you to trigger the processing of calculated " + + "fields independently, ensuring that derived data can be computed and utilized in real time without storing the original message in the database.", + configDirective = "tbNodeEmptyConfig", + icon = "published_with_changes" +) +public class TbCalculatedFieldsNode implements TbNode { + + private EmptyNodeConfiguration config; + + @Override + public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { + this.config = TbNodeUtils.convert(configuration, EmptyNodeConfiguration.class); + } + + @Override + public void onMsg(TbContext ctx, TbMsg msg) { + switch (msg.getInternalType()) { + case POST_TELEMETRY_REQUEST -> processPostTelemetryRequest(ctx, msg); + case POST_ATTRIBUTES_REQUEST -> processPostAttributesRequest(ctx, msg); + default -> ctx.tellFailure(msg, new IllegalArgumentException("Unsupported msg type: " + msg.getType())); + } + } + + private void processPostTelemetryRequest(TbContext ctx, TbMsg msg) { + Map> tsKvMap = JsonConverter.convertToTelemetry(JsonParser.parseString(msg.getData()), System.currentTimeMillis()); + + if (tsKvMap.isEmpty()) { + ctx.tellSuccess(msg); + return; + } + + List tsKvEntryList = new ArrayList<>(); + for (Map.Entry> tsKvEntry : tsKvMap.entrySet()) { + for (KvEntry kvEntry : tsKvEntry.getValue()) { + tsKvEntryList.add(new BasicTsKvEntry(tsKvEntry.getKey(), kvEntry)); + } + } + + TimeseriesSaveRequest timeseriesSaveRequest = TimeseriesSaveRequest.builder() + .tenantId(ctx.getTenantId()) + .customerId(msg.getCustomerId()) + .entityId(msg.getOriginator()) + .entries(tsKvEntryList) + .previousCalculatedFieldIds(msg.getPreviousCalculatedFieldIds()) + .tbMsgId(msg.getId()) + .tbMsgType(msg.getInternalType()) + .callback(new TelemetryNodeCallback(ctx, msg)) + .build(); + + ctx.getCalculatedFieldQueueService().pushRequestToQueue(timeseriesSaveRequest, timeseriesSaveRequest.getCallback()); + } + + private void processPostAttributesRequest(TbContext ctx, TbMsg msg) { + List newAttributes = new ArrayList<>(JsonConverter.convertToAttributes(JsonParser.parseString(msg.getData()))); + + if (newAttributes.isEmpty()) { + ctx.tellSuccess(msg); + return; + } + + AttributesSaveRequest attributesSaveRequest = AttributesSaveRequest.builder() + .tenantId(ctx.getTenantId()) + .entityId(msg.getOriginator()) + .scope(AttributeScope.valueOf(msg.getMetaData().getValue(SCOPE))) + .entries(newAttributes) + .previousCalculatedFieldIds(msg.getPreviousCalculatedFieldIds()) + .tbMsgId(msg.getId()) + .tbMsgType(msg.getInternalType()) + .callback(new TelemetryNodeCallback(ctx, msg)) + .build(); + ctx.getCalculatedFieldQueueService().pushRequestToQueue(attributesSaveRequest, attributesSaveRequest.getCallback()); + } + +}