dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

BaseTelemetryProcessor - use async instead of .get()

Browse Source
This commit is contained in:
Volodymyr Babak 2025-01-31 16:06:52 +02:00
parent 1223347587
commit 5f793b11f9
1 changed files with 95 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

161
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java
View File

@ -19,6 +19,7 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture; import com.google.common.util.concurrent.SettableFuture;
import com.google.gson.Gson; import com.google.gson.Gson;
import com.google.gson.JsonObject; import com.google.gson.JsonObject;
@ -81,6 +82,7 @@ import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Set;
import java.util.UUID; import java.util.UUID;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -262,27 +264,38 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor {
return new ImmutablePair<>(queueName, ruleChainId); return new ImmutablePair<>(queueName, ruleChainId);
} }
private ListenableFuture<Void> processPostAttributes(TenantId tenantId, CustomerId customerId, EntityId entityId, TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData, long ts) throws Exception { private ListenableFuture<Void> processPostAttributes(TenantId tenantId, CustomerId customerId, EntityId entityId,
TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData, long ts) throws Exception {
SettableFuture<Void> futureToSet = SettableFuture.create(); SettableFuture<Void> futureToSet = SettableFuture.create();
JsonObject json = JsonUtils.getJsonObject(msg.getKvList()); JsonObject json = JsonUtils.getJsonObject(msg.getKvList());
List<AttributeKvEntry> attributes = new ArrayList<>(JsonConverter.convertToAttributes(json, ts)); List<AttributeKvEntry> attributes = new ArrayList<>(JsonConverter.convertToAttributes(json, ts));
filterAttributesByTs(tenantId, entityId, AttributeScope.CLIENT_SCOPE, attributes, json); ListenableFuture<List<AttributeKvEntry>> future = filterAttributesByTs(tenantId, entityId, AttributeScope.CLIENT_SCOPE, attributes);
Futures.addCallback(future, new FutureCallback<>() {
var defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId);
TbMsg tbMsg = TbMsg.newMsg()
.queueName(defaultQueueAndRuleChain.getKey())
.type(TbMsgType.POST_ATTRIBUTES_REQUEST)
.originator(entityId)
.customerId(customerId)
.copyMetaData(metaData)
.data(gson.toJson(json))
.ruleChainId(defaultQueueAndRuleChain.getValue())
.build();
edgeCtx.getClusterService().pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() {
@Override @Override
public void onSuccess(TbQueueMsgMetadata metadata) { public void onSuccess(List<AttributeKvEntry> attributesToSave) {
futureToSet.set(null); JsonObject jsonToSave = filterAttributesFromJson(json, attributesToSave);
var defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId);
TbMsg tbMsg = TbMsg.newMsg()
.queueName(defaultQueueAndRuleChain.getKey())
.type(TbMsgType.POST_ATTRIBUTES_REQUEST)
.originator(entityId)
.customerId(customerId)
.copyMetaData(metaData)
.data(gson.toJson(jsonToSave))
.ruleChainId(defaultQueueAndRuleChain.getValue())
.build();
edgeCtx.getClusterService().pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() {
@Override
public void onSuccess(TbQueueMsgMetadata metadata) {
futureToSet.set(null);
}
@Override
public void onFailure(Throwable t) {
log.error("[{}] Can't process post attributes [{}]", tenantId, msg, t);
futureToSet.setException(t);
}
});
} }
@Override @Override
@ -290,7 +303,7 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor {
log.error("[{}] Can't process post attributes [{}]", tenantId, msg, t); log.error("[{}] Can't process post attributes [{}]", tenantId, msg, t);
futureToSet.setException(t); futureToSet.setException(t);
} }
}); }, dbCallbackExecutorService);
return futureToSet; return futureToSet;
} }
@ -299,34 +312,46 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor {
EntityId entityId, EntityId entityId,
TransportProtos.PostAttributeMsg msg, TransportProtos.PostAttributeMsg msg,
TbMsgMetaData metaData, TbMsgMetaData metaData,
long ts) throws Exception { long ts) {
SettableFuture<Void> futureToSet = SettableFuture.create(); SettableFuture<Void> futureToSet = SettableFuture.create();
JsonObject json = JsonUtils.getJsonObject(msg.getKvList()); JsonObject json = JsonUtils.getJsonObject(msg.getKvList());
AttributeScope scope = AttributeScope.valueOf(metaData.getValue("scope")); AttributeScope scope = AttributeScope.valueOf(metaData.getValue("scope"));
List<AttributeKvEntry> attributes = new ArrayList<>(JsonConverter.convertToAttributes(json, ts)); List<AttributeKvEntry> attributes = new ArrayList<>(JsonConverter.convertToAttributes(json, ts));
List<AttributeKvEntry> attributesToSave = filterAttributesByTs(tenantId, entityId, scope, attributes, json); ListenableFuture<List<AttributeKvEntry>> future = filterAttributesByTs(tenantId, entityId, scope, attributes);
tsSubService.saveAttributes(AttributesSaveRequest.builder() Futures.addCallback(future, new FutureCallback<>() {
.tenantId(tenantId) @Override
.entityId(entityId) public void onSuccess(List<AttributeKvEntry> attributesToSave) {
.scope(scope) JsonObject jsonToSave = filterAttributesFromJson(json, attributesToSave);
.entries(attributesToSave) tsSubService.saveAttributes(AttributesSaveRequest.builder()
.callback(new FutureCallback<>() { .tenantId(tenantId)
@Override .entityId(entityId)
public void onSuccess(@Nullable Void tmp) { .scope(scope)
var defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId); .entries(attributesToSave)
TbMsg tbMsg = TbMsg.newMsg() .callback(new FutureCallback<>() {
.queueName(defaultQueueAndRuleChain.getKey())
.type(TbMsgType.ATTRIBUTES_UPDATED)
.originator(entityId)
.customerId(customerId)
.copyMetaData(metaData)
.data(gson.toJson(json))
.ruleChainId(defaultQueueAndRuleChain.getValue())
.build();
edgeCtx.getClusterService().pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() {
@Override @Override
public void onSuccess(TbQueueMsgMetadata metadata) { public void onSuccess(@Nullable Void tmp) {
futureToSet.set(null); var defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId);
TbMsg tbMsg = TbMsg.newMsg()
.queueName(defaultQueueAndRuleChain.getKey())
.type(TbMsgType.ATTRIBUTES_UPDATED)
.originator(entityId)
.customerId(customerId)
.copyMetaData(metaData)
.data(gson.toJson(jsonToSave))
.ruleChainId(defaultQueueAndRuleChain.getValue())
.build();
edgeCtx.getClusterService().pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() {
@Override
public void onSuccess(TbQueueMsgMetadata metadata) {
futureToSet.set(null);
}
@Override
public void onFailure(Throwable t) {
log.error("[{}] Can't process attributes update [{}]", tenantId, msg, t);
futureToSet.setException(t);
}
});
} }
@Override @Override
@ -334,22 +359,28 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor {
log.error("[{}] Can't process attributes update [{}]", tenantId, msg, t); log.error("[{}] Can't process attributes update [{}]", tenantId, msg, t);
futureToSet.setException(t); futureToSet.setException(t);
} }
}); })
} .build());
}
@Override @Override
public void onFailure(Throwable t) { public void onFailure(Throwable t) {
log.error("[{}] Can't process attributes update [{}]", tenantId, msg, t); log.error("[{}] Can't process attributes update [{}]", tenantId, msg, t);
futureToSet.setException(t); futureToSet.setException(t);
} }
}) }, dbCallbackExecutorService);
.build());
return futureToSet; return futureToSet;
} }
private ListenableFuture<Void> processAttributeDeleteMsg(TenantId tenantId, EntityId entityId, AttributeDeleteMsg attributeDeleteMsg, private JsonObject filterAttributesFromJson(JsonObject json, List<AttributeKvEntry> attributesToSave) {
String entityType) { Set<String> keysToSave = attributesToSave.stream()
.map(KvEntry::getKey)
.collect(Collectors.toSet());
json.keySet().removeIf(key -> !keysToSave.contains(key));
return json;
}
private ListenableFuture<Void> processAttributeDeleteMsg(TenantId tenantId, EntityId entityId, AttributeDeleteMsg attributeDeleteMsg, String entityType) {
String scope = attributeDeleteMsg.getScope(); String scope = attributeDeleteMsg.getScope();
List<String> attributeKeys = attributeDeleteMsg.getAttributeNamesList(); List<String> attributeKeys = attributeDeleteMsg.getAttributeNamesList();
ListenableFuture<List<String>> removeAllFuture = edgeCtx.getAttributesService().removeAll(tenantId, entityId, AttributeScope.valueOf(scope), attributeKeys); ListenableFuture<List<String>> removeAllFuture = edgeCtx.getAttributesService().removeAll(tenantId, entityId, AttributeScope.valueOf(scope), attributeKeys);
@ -401,21 +432,19 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor {
entityDataMsgConstructor.constructEntityDataMsg(tenantId, entityId, actionType, JsonParser.parseString(bodyJackson)); entityDataMsgConstructor.constructEntityDataMsg(tenantId, entityId, actionType, JsonParser.parseString(bodyJackson));
} }
private List<AttributeKvEntry> filterAttributesByTs(TenantId tenantId, EntityId entityId, AttributeScope scope, List<AttributeKvEntry> attributes, JsonObject jsonObject) throws Exception { private ListenableFuture<List<AttributeKvEntry>> filterAttributesByTs(TenantId tenantId, EntityId entityId, AttributeScope scope,
List<AttributeKvEntry> attributes) {
List<String> keys = attributes.stream().map(KvEntry::getKey).toList(); List<String> keys = attributes.stream().map(KvEntry::getKey).toList();
Map<String, Long> existingAttributesTs = edgeCtx.getAttributesService().find(tenantId, entityId, scope, keys).get() ListenableFuture<List<AttributeKvEntry>> future = edgeCtx.getAttributesService().find(tenantId, entityId, scope, keys);
.stream().collect(Collectors.toMap(KvEntry::getKey, AttributeKvEntry::getLastUpdateTs)); return Futures.transform(future, input -> {
return attributes.stream() Map<String, Long> existingAttributesTs = input.stream().collect(Collectors.toMap(KvEntry::getKey, AttributeKvEntry::getLastUpdateTs));
.filter(attribute -> { return attributes.stream()
String key = attribute.getKey(); .filter(attribute -> {
long incomingTs = attribute.getLastUpdateTs(); String key = attribute.getKey();
if (incomingTs > existingAttributesTs.getOrDefault(key, 0L)) { long incomingTs = attribute.getLastUpdateTs();
return true; return incomingTs > existingAttributesTs.getOrDefault(key, 0L);
} else { }).toList();
jsonObject.remove(key); }, dbCallbackExecutorService);
return false;
}
}).toList();
} }
} }