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 2da53757ff..a53e77d241 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 @@ -24,11 +24,13 @@ import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.actors.TbActorCtx; import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor; import org.thingsboard.server.common.data.AttributeScope; +import org.thingsboard.server.common.data.cf.configuration.Argument; import org.thingsboard.server.common.data.cf.configuration.ArgumentType; import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey; 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.StringDataEntry; import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg; import org.thingsboard.server.common.msg.queue.TbCallback; @@ -164,7 +166,11 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM if (cfIds.contains(ctx.getCfId())) { callback.onSuccess(CALLBACKS_PER_CF); } else { - if (proto.getTsDataCount() > 0) { + if (proto.getRemovedTsKeysCount() > 0) { + processArgumentValuesUpdate(ctx, cfIds, callback, mapDeletedAttributesToDefault(ctx, msg.getEntityId(), proto.getScope(), proto.getRemovedTsKeysList()), toTbMsgId(proto), toTbMsgType(proto)); + } else if (proto.getRemovedAttrKeysCount() > 0) { + processArgumentValuesUpdate(ctx, cfIds, callback, mapDeletedAttributesToDefault(ctx, msg.getEntityId(), proto.getScope(), proto.getRemovedAttrKeysList()), toTbMsgId(proto), toTbMsgType(proto)); + } else if (proto.getTsDataCount() > 0) { processArgumentValuesUpdate(ctx, cfIds, callback, mapToArguments(ctx, msg.getEntityId(), proto.getTsDataList()), toTbMsgId(proto), toTbMsgType(proto)); } else if (proto.getAttrDataCount() > 0) { processArgumentValuesUpdate(ctx, cfIds, callback, mapToArguments(ctx, msg.getEntityId(), proto.getScope(), proto.getAttrDataList()), toTbMsgId(proto), toTbMsgType(proto)); @@ -182,7 +188,11 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM if (cfIds.contains(ctx.getCfId())) { callback.onSuccess(CALLBACKS_PER_CF); } else { - if (proto.getTsDataCount() > 0) { + if (proto.getRemovedTsKeysCount() > 0) { + processRemovedTelemetry(ctx, proto, cfIdList, callback); + } else if (proto.getRemovedAttrKeysCount() > 0) { + processRemovedAttributes(ctx, proto, cfIdList, callback); + } else if (proto.getTsDataCount() > 0) { processTelemetry(ctx, proto, cfIdList, callback); } else if (proto.getAttrDataCount() > 0) { processAttributes(ctx, proto, cfIdList, callback); @@ -191,23 +201,43 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } } } catch (Exception e) { + if (e instanceof CalculatedFieldException) { + throw (CalculatedFieldException) e; + } throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build(); } } - @SneakyThrows - private void processTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, MultipleTbCallback callback) { + private void processRemovedTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, MultipleTbCallback callback) throws CalculatedFieldException { + // reinit cf (consider fetching only removed ts) + log.info("Force reinitialization of CF: [{}].", ctx.getCfId()); + states.remove(ctx.getCfId()); + try { + var state = getOrInitState(ctx); + if (state.isSizeOk()) { + processStateIfReady(ctx, Collections.singletonList(ctx.getCfId()), state, null, null, callback); + } else { + throw new RuntimeException(ctx.getSizeExceedsLimitMessage()); + } + } catch (Exception e) { + throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build(); + } + } + + private void processRemovedAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, MultipleTbCallback callback) throws CalculatedFieldException { + processArgumentValuesUpdate(ctx, cfIdList, callback, mapDeletedAttributesToDefault(ctx, proto.getScope(), proto.getRemovedAttrKeysList()), toTbMsgId(proto), toTbMsgType(proto)); + } + + private void processTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, MultipleTbCallback callback) throws CalculatedFieldException { processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArguments(ctx, proto.getTsDataList()), toTbMsgId(proto), toTbMsgType(proto)); } - @SneakyThrows - private void processAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, MultipleTbCallback callback) { + private void processAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, MultipleTbCallback callback) throws CalculatedFieldException { processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArguments(ctx, proto.getScope(), proto.getAttrDataList()), toTbMsgId(proto), toTbMsgType(proto)); } - @SneakyThrows private void processArgumentValuesUpdate(CalculatedFieldCtx ctx, List cfIdList, MultipleTbCallback callback, - Map newArgValues, UUID tbMsgId, TbMsgType tbMsgType) { + Map newArgValues, UUID tbMsgId, TbMsgType tbMsgType) throws CalculatedFieldException { if (newArgValues.isEmpty()) { log.info("[{}] No new argument values to process for CF.", ctx.getCfId()); callback.onSuccess(CALLBACKS_PER_CF); @@ -348,6 +378,35 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM return arguments; } + private Map mapDeletedAttributesToDefault(CalculatedFieldCtx ctx, EntityId entityId, AttributeScopeProto scope, List removedAttrKeys) { + var argNames = ctx.getLinkedEntityArguments().get(entityId); + if (argNames.isEmpty()) { + return Collections.emptyMap(); + } + return mapToArgumentsDefaultValue(argNames, ctx.getArguments(), scope, removedAttrKeys); + } + + private Map mapDeletedAttributesToDefault(CalculatedFieldCtx ctx, AttributeScopeProto scope, List removedAttrKeys) { + return mapToArgumentsDefaultValue(ctx.getMainEntityArguments(), ctx.getArguments(), scope, removedAttrKeys); + } + + private static Map mapToArgumentsDefaultValue(Map argNames, Map configArguments, AttributeScopeProto scope, List removedAttrKeys) { + Map arguments = new HashMap<>(); + for (String removedKey : removedAttrKeys) { + ReferencedEntityKey key = new ReferencedEntityKey(removedKey, ArgumentType.ATTRIBUTE, AttributeScope.valueOf(scope.name())); + String argName = argNames.get(key); + if (argName != null) { + Argument argument = configArguments.get(argName); + String defaultValue = (argument != null) ? argument.getDefaultValue() : null; + arguments.put(argName, (defaultValue != null) + ? new SingleValueArgumentEntry(System.currentTimeMillis(), new StringDataEntry(removedKey, defaultValue), null) + : new SingleValueArgumentEntry()); + + } + } + return arguments; + } + private static List getCalculatedFieldIds(CalculatedFieldTelemetryMsgProto proto) { List cfIds = new LinkedList<>(); for (var cfId : proto.getPreviousCalculatedFieldsList()) { 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 b84b54af81..1b0d1462f2 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 @@ -16,7 +16,9 @@ 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.TimeseriesDeleteRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.common.data.kv.TimeseriesSaveResult; @@ -34,4 +36,8 @@ public interface CalculatedFieldQueueService { void pushRequestToQueue(AttributesSaveRequest request, List result, FutureCallback callback); + void pushRequestToQueue(AttributesDeleteRequest request, List result, FutureCallback callback); + + void pushRequestToQueue(TimeseriesDeleteRequest request, List result, FutureCallback callback); + } 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 9d1f9b6db5..13905dbcce 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 @@ -21,7 +21,9 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; +import org.thingsboard.rule.engine.api.AttributesDeleteRequest; import org.thingsboard.rule.engine.api.AttributesSaveRequest; +import org.thingsboard.rule.engine.api.TimeseriesDeleteRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.EntityType; @@ -48,6 +50,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.ArrayList; import java.util.EnumSet; import java.util.List; import java.util.Set; @@ -104,6 +107,23 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS () -> toCalculatedFieldTelemetryMsgProto(request, result), callback); } + @Override + public void pushRequestToQueue(AttributesDeleteRequest request, List result, FutureCallback callback) { + var tenantId = request.getTenantId(); + var entityId = request.getEntityId(); + checkEntityAndPushToQueue(tenantId, entityId, cf -> cf.matchesKeys(result, request.getScope()), cf -> cf.linkMatchesAttrKeys(entityId, result, request.getScope()), + () -> toCalculatedFieldTelemetryMsgProto(request, result), callback); + } + + @Override + public void pushRequestToQueue(TimeseriesDeleteRequest request, List result, FutureCallback callback) { + var tenantId = request.getTenantId(); + var entityId = request.getEntityId(); + + checkEntityAndPushToQueue(tenantId, entityId, cf -> cf.matchesKeys(result), cf -> cf.linkMatchesTsKeys(entityId, result), + () -> toCalculatedFieldTelemetryMsgProto(request, result), callback); + } + private void checkEntityAndPushToQueue(TenantId tenantId, EntityId entityId, Predicate mainEntityFilter, Predicate linkedEntityFilter, Supplier msg, FutureCallback callback) { @@ -174,6 +194,23 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS return msg.build(); } + private ToCalculatedFieldMsg toCalculatedFieldTelemetryMsgProto(AttributesDeleteRequest request, List removedKeys) { + CalculatedFieldTelemetryMsgProto telemetryMsg = buildTelemetryMsgProto(request.getTenantId(), request.getEntityId(), new ArrayList<>(), request.getTbMsgId(), request.getTbMsgType()) + .setScope(AttributeScopeProto.valueOf(request.getScope().name())) + .addAllRemovedAttrKeys(removedKeys).build(); + return ToCalculatedFieldMsg.newBuilder() + .setTelemetryMsg(telemetryMsg) + .build(); + } + + private ToCalculatedFieldMsg toCalculatedFieldTelemetryMsgProto(TimeseriesDeleteRequest request, List removedKeys) { + CalculatedFieldTelemetryMsgProto telemetryMsg = buildTelemetryMsgProto(request.getTenantId(), request.getEntityId(), new ArrayList<>(), request.getTbMsgId(), request.getTbMsgType()) + .addAllRemovedTsKeys(removedKeys).build(); + return ToCalculatedFieldMsg.newBuilder() + .setTelemetryMsg(telemetryMsg) + .build(); + } + private CalculatedFieldTelemetryMsgProto.Builder buildTelemetryMsgProto(TenantId tenantId, EntityId entityId, List calculatedFieldIds, UUID tbMsgId, TbMsgType tbMsgType) { CalculatedFieldTelemetryMsgProto.Builder telemetryMsg = CalculatedFieldTelemetryMsgProto.newBuilder(); 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 4e4844dbe8..e254680019 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 @@ -151,10 +151,28 @@ public class CalculatedFieldCtx { } } + public boolean matchesKeys(List keys, AttributeScope scope) { + return matchesAttributesKeys(mainEntityArguments, keys, scope); + } + + public boolean matchesKeys(List keys) { + return matchesTimeSeriesKeys(mainEntityArguments, keys); + } + public boolean matches(List values, AttributeScope scope) { return matchesAttributes(mainEntityArguments, values, scope); } + public boolean linkMatchesAttrKeys(EntityId entityId, List keys, AttributeScope scope) { + var map = linkedEntityArguments.get(entityId); + return map != null && matchesAttributesKeys(map, keys, scope); + } + + public boolean linkMatchesTsKeys(EntityId entityId, List keys) { + var map = linkedEntityArguments.get(entityId); + return map != null && matchesTimeSeriesKeys(map, keys); + } + public boolean linkMatches(EntityId entityId, List values, AttributeScope scope) { var map = linkedEntityArguments.get(entityId); return map != null && matchesAttributes(map, values, scope); @@ -179,6 +197,16 @@ public class CalculatedFieldCtx { return false; } + private boolean matchesAttributesKeys(Map argMap, List keys, AttributeScope scope) { + for (String key : keys) { + ReferencedEntityKey attrKey = new ReferencedEntityKey(key, ArgumentType.ATTRIBUTE, scope); + if (argMap.containsKey(attrKey)) { + return true; + } + } + return false; + } + private boolean matchesTimeSeries(Map argMap, List values) { for (TsKvEntry tsKv : values) { ReferencedEntityKey latestKey = new ReferencedEntityKey(tsKv.getKey(), ArgumentType.TS_LATEST, null); @@ -193,8 +221,26 @@ public class CalculatedFieldCtx { return false; } + private boolean matchesTimeSeriesKeys(Map argMap, List keys) { + for (String key : keys) { + ReferencedEntityKey latestKey = new ReferencedEntityKey(key, ArgumentType.TS_LATEST, null); + if (argMap.containsKey(latestKey)) { + return true; + } + ReferencedEntityKey rollingKey = new ReferencedEntityKey(key, ArgumentType.TS_ROLLING, null); + if (argMap.containsKey(rollingKey)) { + return true; + } + } + return false; + } + public boolean linkMatches(EntityId entityId, CalculatedFieldTelemetryMsgProto proto) { - if (!proto.getTsDataList().isEmpty()) { + if (!proto.getRemovedTsKeysList().isEmpty()) { + return linkMatchesTsKeys(entityId, proto.getRemovedTsKeysList()); + } else if (!proto.getRemovedAttrKeysList().isEmpty()) { + return linkMatchesAttrKeys(entityId, proto.getRemovedAttrKeysList(), AttributeScope.valueOf(proto.getScope().name())); + } else if (!proto.getTsDataList().isEmpty()) { List updatedTelemetry = proto.getTsDataList().stream() .map(ProtoUtils::fromProto) .toList(); diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java index 79222ca002..3f9b88b8d6 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java @@ -186,7 +186,9 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer @Override public void deleteAttributesInternal(AttributesDeleteRequest request) { ListenableFuture> deleteFuture = attrService.removeAll(request.getTenantId(), request.getEntityId(), request.getScope(), request.getKeys()); - addMainCallback(deleteFuture, request.getCallback()); + DonAsynchron.withCallback(deleteFuture, result -> { + calculatedFieldQueueService.pushRequestToQueue(request, result, request.getCallback()); + }, safeCallback(request.getCallback()), tsCallBackExecutor); addWsCallback(deleteFuture, success -> onAttributesDelete(request.getTenantId(), request.getEntityId(), request.getScope().name(), request.getKeys(), request.isNotifyDevice())); } @@ -206,7 +208,10 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer deleteFuture = tsService.remove(request.getTenantId(), request.getEntityId(), request.getDeleteHistoryQueries()); addWsCallback(deleteFuture, result -> onTimeSeriesDelete(request.getTenantId(), request.getEntityId(), request.getKeys(), result)); } - addMainCallback(deleteFuture, __ -> request.getCallback().onSuccess(request.getKeys()), request.getCallback()::onFailure); + DonAsynchron.withCallback(deleteFuture, result -> { + calculatedFieldQueueService.pushRequestToQueue(request, request.getKeys(), getCalculatedFieldCallback(request.getCallback(), request.getKeys())); + }, safeCallback(getCalculatedFieldCallback(request.getCallback(), request.getKeys())), tsCallBackExecutor); +// addMainCallback(deleteFuture, __ -> request.getCallback().onSuccess(request.getKeys()), request.getCallback()::onFailure); } else { ListenableFuture> deleteFuture = tsService.removeAllLatest(request.getTenantId(), request.getEntityId()); addMainCallback(deleteFuture, request.getCallback()::onSuccess, request.getCallback()::onFailure); @@ -338,4 +343,18 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer }; } + private FutureCallback getCalculatedFieldCallback(FutureCallback> originalCallback, List keys) { + return new FutureCallback() { + @Override + public void onSuccess(Void unused) { + originalCallback.onSuccess(keys); + } + + @Override + public void onFailure(Throwable t) { + originalCallback.onFailure(t); + } + }; + } + } diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 5fc7e7c59c..432edd2201 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -783,9 +783,11 @@ message CalculatedFieldTelemetryMsgProto { repeated TsKvProto tsData = 9; AttributeScopeProto scope = 10; repeated AttributeValueProto attrData = 11; - int64 tbMsgIdMSB = 12; - int64 tbMsgIdLSB = 13; - string tbMsgType = 14; + repeated string removedTsKeys = 12; + repeated string removedAttrKeys = 13; + int64 tbMsgIdMSB = 14; + int64 tbMsgIdLSB = 15; + string tbMsgType = 16; } message CalculatedFieldLinkedTelemetryMsgProto { diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesDeleteRequest.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesDeleteRequest.java index 118c62e78c..e161c0f165 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesDeleteRequest.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesDeleteRequest.java @@ -24,8 +24,10 @@ import lombok.ToString; import org.thingsboard.server.common.data.AttributeScope; 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 java.util.List; +import java.util.UUID; @Getter @ToString @@ -37,6 +39,8 @@ public class AttributesDeleteRequest { private final AttributeScope scope; private final List keys; private final boolean notifyDevice; + private final UUID tbMsgId; + private final TbMsgType tbMsgType; private final FutureCallback callback; public static Builder builder() { @@ -50,9 +54,12 @@ public class AttributesDeleteRequest { private AttributeScope scope; private List keys; private boolean notifyDevice; + private UUID tbMsgId; + private TbMsgType tbMsgType; private FutureCallback callback; - Builder() {} + Builder() { + } public Builder tenantId(TenantId tenantId) { this.tenantId = tenantId; @@ -89,6 +96,16 @@ public class AttributesDeleteRequest { return this; } + public Builder tbMsgId(UUID tbMsgId) { + this.tbMsgId = tbMsgId; + return this; + } + + public Builder tbMsgType(TbMsgType tbMsgType) { + this.tbMsgType = tbMsgType; + return this; + } + public Builder callback(FutureCallback callback) { this.callback = callback; return this; @@ -109,7 +126,7 @@ public class AttributesDeleteRequest { } public AttributesDeleteRequest build() { - return new AttributesDeleteRequest(tenantId, entityId, scope, keys, notifyDevice, callback); + return new AttributesDeleteRequest(tenantId, entityId, scope, keys, notifyDevice, tbMsgId, tbMsgType, callback); } } diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesDeleteRequest.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesDeleteRequest.java index b124806fff..62fc28737c 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesDeleteRequest.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesDeleteRequest.java @@ -23,8 +23,10 @@ import lombok.ToString; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.DeleteTsKvQuery; +import org.thingsboard.server.common.data.msg.TbMsgType; import java.util.List; +import java.util.UUID; @Getter @ToString @@ -35,6 +37,8 @@ public class TimeseriesDeleteRequest { private final EntityId entityId; private final List keys; private final List deleteHistoryQueries; + private final UUID tbMsgId; + private final TbMsgType tbMsgType; private final FutureCallback> callback; public static Builder builder() { @@ -47,9 +51,12 @@ public class TimeseriesDeleteRequest { private EntityId entityId; private List keys; private List deleteHistoryQueries; + private UUID tbMsgId; + private TbMsgType tbMsgType; private FutureCallback> callback; - Builder() {} + Builder() { + } public Builder tenantId(TenantId tenantId) { this.tenantId = tenantId; @@ -71,13 +78,23 @@ public class TimeseriesDeleteRequest { return this; } + public Builder tbMsgId(UUID tbMsgId) { + this.tbMsgId = tbMsgId; + return this; + } + + public Builder tbMsgType(TbMsgType tbMsgType) { + this.tbMsgType = tbMsgType; + return this; + } + public Builder callback(FutureCallback> callback) { this.callback = callback; return this; } public TimeseriesDeleteRequest build() { - return new TimeseriesDeleteRequest(tenantId, entityId, keys, deleteHistoryQueries, callback); + return new TimeseriesDeleteRequest(tenantId, entityId, keys, deleteHistoryQueries, tbMsgId, tbMsgType, callback); } } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesNode.java index a89a4f37d8..87a8b73c49 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesNode.java @@ -76,6 +76,8 @@ public class TbMsgDeleteAttributesNode implements TbNode { .scope(scope) .keys(keysToDelete) .notifyDevice(checkNotifyDevice(msg.getMetaData().getValue(NOTIFY_DEVICE_METADATA_KEY), scope)) + .tbMsgId(msg.getId()) + .tbMsgType(msg.getInternalType()) .callback(config.isSendAttributesDeletedNotification() ? new AttributesDeleteNodeCallback(ctx, msg, scope.name(), keysToDelete) : new TelemetryNodeCallback(ctx, msg))