From b2b33199fea5fcf6940a2aca642931f332942376 Mon Sep 17 00:00:00 2001 From: Dmytro Skarzhynets Date: Thu, 6 Feb 2025 15:55:41 +0200 Subject: [PATCH] Corrections after code review --- .../DefaultTelemetrySubscriptionService.java | 96 +++++++++---------- ...faultTelemetrySubscriptionServiceTest.java | 2 +- 2 files changed, 48 insertions(+), 50 deletions(-) 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 43e4b66c40..5b5054ae38 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 @@ -150,7 +150,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer if (strategy.sendWsUpdate()) { addWsCallback(saveFuture, success -> onTimeSeriesUpdate(tenantId, entityId, request.getEntries())); } - if (strategy.saveLatest()) { + if (strategy.saveLatest() && entityId.getEntityType().isOneOf(EntityType.DEVICE, EntityType.ASSET)) { addMainCallback(saveFuture, __ -> copyLatestToEntityViews(tenantId, entityId, request.getEntries())); } return saveFuture; @@ -207,58 +207,56 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer } private void copyLatestToEntityViews(TenantId tenantId, EntityId entityId, List ts) { - if (entityId.getEntityType().isOneOf(EntityType.DEVICE, EntityType.ASSET)) { - Futures.addCallback(tbEntityViewService.findEntityViewsByTenantIdAndEntityIdAsync(tenantId, entityId), - new FutureCallback<>() { - @Override - public void onSuccess(@Nullable List result) { - if (result != null && !result.isEmpty()) { - Map> tsMap = new HashMap<>(); - for (TsKvEntry entry : ts) { - tsMap.computeIfAbsent(entry.getKey(), s -> new ArrayList<>()).add(entry); - } - for (EntityView entityView : result) { - List keys = entityView.getKeys() != null && entityView.getKeys().getTimeseries() != null ? - entityView.getKeys().getTimeseries() : new ArrayList<>(tsMap.keySet()); - List entityViewLatest = new ArrayList<>(); - long startTs = entityView.getStartTimeMs(); - long endTs = entityView.getEndTimeMs() == 0 ? Long.MAX_VALUE : entityView.getEndTimeMs(); - for (String key : keys) { - List entries = tsMap.get(key); - if (entries != null) { - Optional tsKvEntry = entries.stream() - .filter(entry -> entry.getTs() > startTs && entry.getTs() <= endTs) - .max(Comparator.comparingLong(TsKvEntry::getTs)); - tsKvEntry.ifPresent(entityViewLatest::add); - } - } - if (!entityViewLatest.isEmpty()) { - saveTimeseries(TimeseriesSaveRequest.builder() - .tenantId(tenantId) - .entityId(entityView.getId()) - .entries(entityViewLatest) - .strategy(TimeseriesSaveRequest.Strategy.LATEST_AND_WS) - .callback(new FutureCallback<>() { - @Override - public void onSuccess(@Nullable Void tmp) {} - - @Override - public void onFailure(Throwable t) { - log.error("[{}][{}] Failed to save entity view latest timeseries: {}", tenantId, entityView.getId(), entityViewLatest, t); - } - }) - .build()); + Futures.addCallback(tbEntityViewService.findEntityViewsByTenantIdAndEntityIdAsync(tenantId, entityId), + new FutureCallback<>() { + @Override + public void onSuccess(@Nullable List result) { + if (result != null && !result.isEmpty()) { + Map> tsMap = new HashMap<>(); + for (TsKvEntry entry : ts) { + tsMap.computeIfAbsent(entry.getKey(), s -> new ArrayList<>()).add(entry); + } + for (EntityView entityView : result) { + List keys = entityView.getKeys() != null && entityView.getKeys().getTimeseries() != null ? + entityView.getKeys().getTimeseries() : new ArrayList<>(tsMap.keySet()); + List entityViewLatest = new ArrayList<>(); + long startTs = entityView.getStartTimeMs(); + long endTs = entityView.getEndTimeMs() == 0 ? Long.MAX_VALUE : entityView.getEndTimeMs(); + for (String key : keys) { + List entries = tsMap.get(key); + if (entries != null) { + Optional tsKvEntry = entries.stream() + .filter(entry -> entry.getTs() > startTs && entry.getTs() <= endTs) + .max(Comparator.comparingLong(TsKvEntry::getTs)); + tsKvEntry.ifPresent(entityViewLatest::add); } } + if (!entityViewLatest.isEmpty()) { + saveTimeseries(TimeseriesSaveRequest.builder() + .tenantId(tenantId) + .entityId(entityView.getId()) + .entries(entityViewLatest) + .strategy(TimeseriesSaveRequest.Strategy.LATEST_AND_WS) + .callback(new FutureCallback<>() { + @Override + public void onSuccess(@Nullable Void tmp) {} + + @Override + public void onFailure(Throwable t) { + log.error("[{}][{}] Failed to save entity view latest timeseries: {}", tenantId, entityView.getId(), entityViewLatest, t); + } + }) + .build()); + } } } + } - @Override - public void onFailure(Throwable t) { - log.error("Error while finding entity views by tenantId and entityId", t); - } - }, MoreExecutors.directExecutor()); - } + @Override + public void onFailure(Throwable t) { + log.error("Error while finding entity views by tenantId and entityId", t); + } + }, MoreExecutors.directExecutor()); } private void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes, boolean notifyDevice) { @@ -313,7 +311,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer } private void addMainCallback(ListenableFuture saveFuture, Consumer onSuccess) { - DonAsynchron.withCallback(saveFuture, onSuccess, null, tsCallBackExecutor); + addMainCallback(saveFuture, onSuccess, null); } private void addMainCallback(ListenableFuture saveFuture, Consumer onSuccess, Consumer onFailure) { diff --git a/application/src/test/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionServiceTest.java b/application/src/test/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionServiceTest.java index 4ded639c76..60a16c16a4 100644 --- a/application/src/test/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionServiceTest.java @@ -345,7 +345,7 @@ class DefaultTelemetrySubscriptionServiceTest { .callback(emptyCallback) .build(); - given(tsService.save(tenantId, entityId, sampleTelemetry, sampleTtl)).willReturn(immediateFailedFuture(new RuntimeException("failed to save latest on main entity"))); + given(tsService.save(tenantId, entityId, sampleTelemetry, sampleTtl)).willReturn(immediateFailedFuture(new RuntimeException("failed to save data on main entity"))); // WHEN telemetryService.saveTimeseries(request);