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

reportActivity: add error to callback

Browse Source
This commit is contained in:
nick 2024-10-17 16:46:16 +03:00
parent 104ad0deb9
commit 12c35d903a
1 changed files with 14 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
View File

@ -1101,8 +1101,16 @@ public class DefaultTransportService extends TransportActivityManager implements
} }
private void sendToCore(TenantId tenantId, EntityId entityId, ToCoreMsg msg, UUID routingKey, TransportServiceCallback<Void> callback) { private void sendToCore(TenantId tenantId, EntityId entityId, ToCoreMsg msg, UUID routingKey, TransportServiceCallback<Void> callback) {
TopicPartitionInfo tpi;
try { try {
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId); tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId);
} catch (TenantNotFoundException e) {
log.trace("Failed to send message to core. Tenant with ID [{}] not found in the database. Message delivery aborted.", tenantId, e);
tpi = TopicPartitionInfo.builder().topic(e.getMessage()).build();
if (callback != null) {
callback.onError(e);
}
}
if (log.isTraceEnabled()) { if (log.isTraceEnabled()) {
log.trace("[{}][{}] Pushing to topic {} message {}", tenantId, entityId, tpi.getFullTopicName(), msg); log.trace("[{}][{}] Pushing to topic {} message {}", tenantId, entityId, tpi.getFullTopicName(), msg);
} }
@ -1111,9 +1119,6 @@ public class DefaultTransportService extends TransportActivityManager implements
tbCoreProducerStats.incrementTotal(); tbCoreProducerStats.incrementTotal();
StatsCallback wrappedCallback = new StatsCallback(transportTbQueueCallback, tbCoreProducerStats); StatsCallback wrappedCallback = new StatsCallback(transportTbQueueCallback, tbCoreProducerStats);
tbCoreMsgProducer.send(tpi, new TbProtoQueueMsg<>(routingKey, msg), wrappedCallback); tbCoreMsgProducer.send(tpi, new TbProtoQueueMsg<>(routingKey, msg), wrappedCallback);
} catch (TenantNotFoundException e) {
log.trace("Failed to send message to core. Tenant with ID [{}] not found in the database. Message delivery aborted.", tenantId, e);
}
} }
private void sendToRuleEngine(TenantId tenantId, DeviceId deviceId, CustomerId customerId, TransportProtos.SessionInfoProto sessionInfo, JsonObject json, private void sendToRuleEngine(TenantId tenantId, DeviceId deviceId, CustomerId customerId, TransportProtos.SessionInfoProto sessionInfo, JsonObject json,