tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

wip ts/attributes deletion

pull/12752/head
IrynaMatveieva 2 years ago
parent
c7d8e6e7b1
commit
49328d270b
9 changed files with 223 additions and 18 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 75
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
  2. 6
      application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldQueueService.java
  3. 37
      application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java
  4. 48
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
  5. 23
      application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java
  6. 8
      common/proto/src/main/proto/queue.proto
  7. 21
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesDeleteRequest.java
  8. 21
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesDeleteRequest.java
  9. 2
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesNode.java

75
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
View File

@ -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<CalculatedFieldId> cfIdList, MultipleTbCallback callback) {
private void processRemovedTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List<CalculatedFieldId> 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<CalculatedFieldId> 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<CalculatedFieldId> 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<CalculatedFieldId> cfIdList, MultipleTbCallback callback) {
private void processAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List<CalculatedFieldId> 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<CalculatedFieldId> cfIdList, MultipleTbCallback callback,
Map<String, ArgumentEntry> newArgValues, UUID tbMsgId, TbMsgType tbMsgType) {
Map<String, ArgumentEntry> 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<String, ArgumentEntry> mapDeletedAttributesToDefault(CalculatedFieldCtx ctx, EntityId entityId, AttributeScopeProto scope, List<String> removedAttrKeys) {
var argNames = ctx.getLinkedEntityArguments().get(entityId);
if (argNames.isEmpty()) {
return Collections.emptyMap();
}
return mapToArgumentsDefaultValue(argNames, ctx.getArguments(), scope, removedAttrKeys);
}
private Map<String, ArgumentEntry> mapDeletedAttributesToDefault(CalculatedFieldCtx ctx, AttributeScopeProto scope, List<String> removedAttrKeys) {
return mapToArgumentsDefaultValue(ctx.getMainEntityArguments(), ctx.getArguments(), scope, removedAttrKeys);
}
private static Map<String, ArgumentEntry> mapToArgumentsDefaultValue(Map<ReferencedEntityKey, String> argNames, Map<String, Argument> configArguments, AttributeScopeProto scope, List<String> removedAttrKeys) {
Map<String, ArgumentEntry> 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<CalculatedFieldId> getCalculatedFieldIds(CalculatedFieldTelemetryMsgProto proto) {
List<CalculatedFieldId> cfIds = new LinkedList<>();
for (var cfId : proto.getPreviousCalculatedFieldsList()) {

6
application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldQueueService.java
View File

@ -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<Long> result, FutureCallback<Void> callback);
void pushRequestToQueue(AttributesDeleteRequest request, List<String> result, FutureCallback<Void> callback);
void pushRequestToQueue(TimeseriesDeleteRequest request, List<String> result, FutureCallback<Void> callback);
}

37
application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java
View File

@ -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<String> result, FutureCallback<Void> 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<String> result, FutureCallback<Void> 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<CalculatedFieldCtx> mainEntityFilter, Predicate<CalculatedFieldCtx> linkedEntityFilter,
Supplier<ToCalculatedFieldMsg> msg, FutureCallback<Void> callback) {
@ -174,6 +194,23 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS
return msg.build();
}
private ToCalculatedFieldMsg toCalculatedFieldTelemetryMsgProto(AttributesDeleteRequest request, List<String> 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<String> 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<CalculatedFieldId> calculatedFieldIds, UUID tbMsgId, TbMsgType tbMsgType) {
CalculatedFieldTelemetryMsgProto.Builder telemetryMsg = CalculatedFieldTelemetryMsgProto.newBuilder();

48
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
View File

@ -151,10 +151,28 @@ public class CalculatedFieldCtx {
}
}
public boolean matchesKeys(List<String> keys, AttributeScope scope) {
return matchesAttributesKeys(mainEntityArguments, keys, scope);
}
public boolean matchesKeys(List<String> keys) {
return matchesTimeSeriesKeys(mainEntityArguments, keys);
}
public boolean matches(List<AttributeKvEntry> values, AttributeScope scope) {
return matchesAttributes(mainEntityArguments, values, scope);
}
public boolean linkMatchesAttrKeys(EntityId entityId, List<String> keys, AttributeScope scope) {
var map = linkedEntityArguments.get(entityId);
return map != null && matchesAttributesKeys(map, keys, scope);
}
public boolean linkMatchesTsKeys(EntityId entityId, List<String> keys) {
var map = linkedEntityArguments.get(entityId);
return map != null && matchesTimeSeriesKeys(map, keys);
}
public boolean linkMatches(EntityId entityId, List<AttributeKvEntry> 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<ReferencedEntityKey, String> argMap, List<String> 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<ReferencedEntityKey, String> argMap, List<TsKvEntry> 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<ReferencedEntityKey, String> argMap, List<String> 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<TsKvEntry> updatedTelemetry = proto.getTsDataList().stream()
.map(ProtoUtils::fromProto)
.toList();

23
application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java
View File

@ -186,7 +186,9 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
@Override
public void deleteAttributesInternal(AttributesDeleteRequest request) {
ListenableFuture<List<String>> 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<List<String>> 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<Void> getCalculatedFieldCallback(FutureCallback<List<String>> originalCallback, List<String> keys) {
return new FutureCallback<Void>() {
@Override
public void onSuccess(Void unused) {
originalCallback.onSuccess(keys);
}
@Override
public void onFailure(Throwable t) {
originalCallback.onFailure(t);
}
};
}
}

8
common/proto/src/main/proto/queue.proto
View File

@ -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 {

21
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesDeleteRequest.java
View File

@ -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<String> keys;
private final boolean notifyDevice;
private final UUID tbMsgId;
private final TbMsgType tbMsgType;
private final FutureCallback<Void> callback;
public static Builder builder() {
@ -50,9 +54,12 @@ public class AttributesDeleteRequest {
private AttributeScope scope;
private List<String> keys;
private boolean notifyDevice;
private UUID tbMsgId;
private TbMsgType tbMsgType;
private FutureCallback<Void> 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<Void> 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);
}
}

21
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesDeleteRequest.java
View File

@ -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<String> keys;
private final List<DeleteTsKvQuery> deleteHistoryQueries;
private final UUID tbMsgId;
private final TbMsgType tbMsgType;
private final FutureCallback<List<String>> callback;
public static Builder builder() {
@ -47,9 +51,12 @@ public class TimeseriesDeleteRequest {
private EntityId entityId;
private List<String> keys;
private List<DeleteTsKvQuery> deleteHistoryQueries;
private UUID tbMsgId;
private TbMsgType tbMsgType;
private FutureCallback<List<String>> 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<List<String>> 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);
}
}

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesNode.java
View File

@ -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))

Write Preview
Loading…
Cancel
Save