From c3556eba7d5c2560bde18944f57ab7b2bb14e8cf Mon Sep 17 00:00:00 2001 From: Dmytro Skarzhynets Date: Fri, 8 Dec 2023 09:55:37 +0200 Subject: [PATCH] [WIP] Refactor abstract and integration activity managers --- ...irstAndLastIntegrationActivityManager.java | 31 +++++----- .../FirstOnlyIntegrationActivityManager.java | 59 ++++++++++--------- .../LastOnlyIntegrationActivityManager.java | 15 ++--- .../activity/AbstractActivityManager.java | 42 ++++++------- .../transport/activity/ActivityManager.java | 8 +-- 5 files changed, 72 insertions(+), 83 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/integration/activity/FirstAndLastIntegrationActivityManager.java b/application/src/main/java/org/thingsboard/server/service/integration/activity/FirstAndLastIntegrationActivityManager.java index 4a38af1a4d..e0dcc97ac0 100644 --- a/application/src/main/java/org/thingsboard/server/service/integration/activity/FirstAndLastIntegrationActivityManager.java +++ b/application/src/main/java/org/thingsboard/server/service/integration/activity/FirstAndLastIntegrationActivityManager.java @@ -70,19 +70,16 @@ public class FirstAndLastIntegrationActivityManager extends AbstractActivityMana SettableFuture> reportCompletedFuture = SettableFuture.create(); states.compute(activityKey, (key, activityStateWrapper) -> { if (activityStateWrapper == null) { - ActivityState activityState = newStateSupplier.get(); - activityState.setLastRecordedTime(newLastRecordedTime); - activityState.setLastReportedTime(0L); activityStateWrapper = new ActivityStateWrapper(); - activityStateWrapper.setState(activityState); - activityStateWrapper.setAlreadyBeenReported(false); - } else { - activityStateWrapper.getState().setLastRecordedTime(newLastRecordedTime); + activityStateWrapper.setState(newStateSupplier.get()); + } + var activityState = activityStateWrapper.getState(); + if (activityState.getLastRecordedTime() < newLastRecordedTime) { + activityState.setLastRecordedTime(newLastRecordedTime); } if (activityStateWrapper.isAlreadyBeenReported()) { return activityStateWrapper; } - var activityState = activityStateWrapper.getState(); if (activityState.getLastReportedTime() < activityState.getLastRecordedTime()) { reporter.report(key, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() { @Override @@ -107,36 +104,36 @@ public class FirstAndLastIntegrationActivityManager extends AbstractActivityMana @Override public void onFailure(@NonNull Throwable t) { - log.debug("[{}] Failed to report first activity event in a period for device with id: [{}]", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId()); + log.debug("[{}] Failed to report first activity event in a period for device with id: [{}].", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId()); } }, MoreExecutors.directExecutor()); } @Override protected void onReportingPeriodEnd() { - long expirationTime = System.currentTimeMillis() - reportingPeriodMillis; for (Map.Entry entry : states.entrySet()) { var activityKey = entry.getKey(); var activityStateWrapper = entry.getValue(); var activityState = activityStateWrapper.getState(); - if (activityState.getLastRecordedTime() < expirationTime) { + long lastRecordedTime = activityState.getLastRecordedTime(); + // if there were no activities during the reporting period, we should remove the entry to prevent memory leaks + if (!activityStateWrapper.isAlreadyBeenReported()) { states.remove(activityKey); } - if (activityState.getLastReportedTime() < activityState.getLastRecordedTime()) { - reporter.report(activityKey, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() { + if (activityState.getLastReportedTime() < lastRecordedTime) { + reporter.report(activityKey, lastRecordedTime, activityState, new ActivityReportCallback<>() { @Override - public void onSuccess(IntegrationActivityKey key, long reportedTime) { - updateLastReportedTime(key, reportedTime); + public void onSuccess(IntegrationActivityKey key, long newLastReportedTime) { + updateLastReportedTime(key, newLastReportedTime); } @Override public void onFailure(IntegrationActivityKey key, Throwable t) { - log.debug("[{}] Failed to report last activity event in a period for device with id: [{}]", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId()); + log.debug("[{}] Failed to report last activity event in a period for device with id: [{}].", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId()); } }); } activityStateWrapper.setAlreadyBeenReported(false); - activityState.getReportsCount().set(0); } } diff --git a/application/src/main/java/org/thingsboard/server/service/integration/activity/FirstOnlyIntegrationActivityManager.java b/application/src/main/java/org/thingsboard/server/service/integration/activity/FirstOnlyIntegrationActivityManager.java index b9965098df..38f14f538a 100644 --- a/application/src/main/java/org/thingsboard/server/service/integration/activity/FirstOnlyIntegrationActivityManager.java +++ b/application/src/main/java/org/thingsboard/server/service/integration/activity/FirstOnlyIntegrationActivityManager.java @@ -70,19 +70,16 @@ public class FirstOnlyIntegrationActivityManager extends AbstractActivityManager SettableFuture> reportCompletedFuture = SettableFuture.create(); states.compute(activityKey, (key, activityStateWrapper) -> { if (activityStateWrapper == null) { - ActivityState activityState = newStateSupplier.get(); - activityState.setLastRecordedTime(newLastRecordedTime); - activityState.setLastReportedTime(0L); activityStateWrapper = new ActivityStateWrapper(); - activityStateWrapper.setState(activityState); - activityStateWrapper.setAlreadyBeenReported(false); - } else { - activityStateWrapper.getState().setLastRecordedTime(newLastRecordedTime); + activityStateWrapper.setState(newStateSupplier.get()); + } + var activityState = activityStateWrapper.getState(); + if (activityState.getLastRecordedTime() < newLastRecordedTime) { + activityState.setLastRecordedTime(newLastRecordedTime); } if (activityStateWrapper.isAlreadyBeenReported()) { return activityStateWrapper; } - var activityState = activityStateWrapper.getState(); if (activityState.getLastReportedTime() < activityState.getLastRecordedTime()) { reporter.report(key, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() { @Override @@ -107,40 +104,46 @@ public class FirstOnlyIntegrationActivityManager extends AbstractActivityManager @Override public void onFailure(@NonNull Throwable t) { - log.debug("[{}] Failed to report first activity event in a period for device with id: [{}]", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId()); + log.debug("[{}] Failed to report first activity event in a period for device with id: [{}].", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId()); } }, MoreExecutors.directExecutor()); } - private void updateLastReportedTime(IntegrationActivityKey key, long newLastReportedTime) { - states.computeIfPresent(key, (__, activityStateWrapper) -> { - var activityState = activityStateWrapper.getState(); - activityState.setLastReportedTime(Math.max(activityState.getLastReportedTime(), newLastReportedTime)); - return activityStateWrapper; - }); - } - @Override protected void onReportingPeriodEnd() { for (Map.Entry entry : states.entrySet()) { var activityKey = entry.getKey(); var activityStateWrapper = entry.getValue(); var activityState = activityStateWrapper.getState(); - if (!activityStateWrapper.isAlreadyBeenReported() && activityState.getLastReportedTime() < activityState.getLastRecordedTime()) { - reporter.report(activityKey, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() { - @Override - public void onSuccess(IntegrationActivityKey key, long reportedTime) { - states.remove(key); - } + long lastRecordedTime = activityState.getLastRecordedTime(); + // if there were no activities during the reporting period, we should remove the entry to prevent memory leaks + if (!activityStateWrapper.isAlreadyBeenReported()) { + states.remove(activityKey); + // report leftover events + if (activityState.getLastReportedTime() < lastRecordedTime) { + reporter.report(activityKey, lastRecordedTime, activityState, new ActivityReportCallback<>() { + @Override + public void onSuccess(IntegrationActivityKey key, long reportedTime) { + updateLastReportedTime(key, reportedTime); // just in case the same key was added again + } - @Override - public void onFailure(IntegrationActivityKey key, Throwable t) { - log.debug("[{}] Failed to report last activity event in a period for device with id: [{}]", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId()); - } - }); + @Override + public void onFailure(IntegrationActivityKey key, Throwable t) { + log.debug("[{}] Failed to report last activity event in a period for device with id: [{}].", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId()); + } + }); + } } activityStateWrapper.setAlreadyBeenReported(false); } } + private void updateLastReportedTime(IntegrationActivityKey key, long newLastReportedTime) { + states.computeIfPresent(key, (__, activityStateWrapper) -> { + var activityState = activityStateWrapper.getState(); + activityState.setLastReportedTime(Math.max(activityState.getLastReportedTime(), newLastReportedTime)); + return activityStateWrapper; + }); + } + } diff --git a/application/src/main/java/org/thingsboard/server/service/integration/activity/LastOnlyIntegrationActivityManager.java b/application/src/main/java/org/thingsboard/server/service/integration/activity/LastOnlyIntegrationActivityManager.java index bffdc49054..ef6fa4f8b6 100644 --- a/application/src/main/java/org/thingsboard/server/service/integration/activity/LastOnlyIntegrationActivityManager.java +++ b/application/src/main/java/org/thingsboard/server/service/integration/activity/LastOnlyIntegrationActivityManager.java @@ -55,9 +55,8 @@ public class LastOnlyIntegrationActivityManager extends AbstractActivityManager< states.compute(activityKey, (__, activityState) -> { if (activityState == null) { activityState = newStateSupplier.get(); - activityState.setLastRecordedTime(newLastRecordedTime); - activityState.setLastReportedTime(0L); - } else { + } + if (activityState.getLastRecordedTime() < newLastRecordedTime) { activityState.setLastRecordedTime(newLastRecordedTime); } return activityState; @@ -70,11 +69,13 @@ public class LastOnlyIntegrationActivityManager extends AbstractActivityManager< for (Map.Entry entry : states.entrySet()) { var activityKey = entry.getKey(); var activityState = entry.getValue(); - if (activityState.getLastRecordedTime() < expirationTime) { + long lastRecordedTime = activityState.getLastRecordedTime(); + // if there were no activities during the reporting period, we should remove the entry to prevent memory leaks + if (lastRecordedTime < expirationTime) { states.remove(activityKey); } - if (activityState.getLastReportedTime() < activityState.getLastRecordedTime()) { - reporter.report(activityKey, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() { + if (activityState.getLastReportedTime() < lastRecordedTime) { + reporter.report(activityKey, lastRecordedTime, activityState, new ActivityReportCallback<>() { @Override public void onSuccess(IntegrationActivityKey key, long reportedTime) { updateLastReportedTime(key, reportedTime); @@ -82,7 +83,7 @@ public class LastOnlyIntegrationActivityManager extends AbstractActivityManager< @Override public void onFailure(IntegrationActivityKey key, Throwable t) { - log.debug("[{}] Failed to report last activity event in a period for device with id: [{}]", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId()); + log.debug("[{}] Failed to report last activity event in a period for device with id: [{}].", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId()); } }); } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/AbstractActivityManager.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/AbstractActivityManager.java index 0c7e00816a..4dc84037a8 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/AbstractActivityManager.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/AbstractActivityManager.java @@ -30,7 +30,9 @@ */ package org.thingsboard.server.common.transport.activity; +import lombok.extern.slf4j.Slf4j; import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.server.common.data.StringUtils; import java.util.Objects; import java.util.Random; @@ -39,52 +41,44 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.function.Supplier; +@Slf4j public abstract class AbstractActivityManager implements ActivityManager { private String name; protected long reportingPeriodMillis; protected ActivityStateReporter reporter; private ScheduledExecutorService scheduler; - protected boolean initialized; + private boolean initialized; @Override - public synchronized void init() { - Objects.requireNonNull(reporter, "Failed to initialize activity manager: provided activity reporter is null."); - if (reportingPeriodMillis < 1000) { - throw new IllegalArgumentException("Failed to initialize activity manager: provided reporting period duration is less that 1 second."); + public synchronized void init(String name, long reportingPeriodMillis, ActivityStateReporter reporter) { + this.name = StringUtils.notBlankOrDefault(name, "activity-manager"); + this.reporter = Objects.requireNonNull(reporter, "Failed to initialize activity manager: provided activity reporter is null."); + if (reportingPeriodMillis <= 0) { + reportingPeriodMillis = 3000; + log.error("[{}] Negative or zero reporting period millisecond was provided. Going to use reporting period value of 3 seconds.", this.name); } + this.reportingPeriodMillis = reportingPeriodMillis; if (!initialized) { - scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName(name == null ? "activity-manager" : name)); + scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName(this.name)); scheduler.scheduleAtFixedRate(this::onReportingPeriodEnd, new Random().nextInt((int) reportingPeriodMillis), reportingPeriodMillis, TimeUnit.MILLISECONDS); initialized = true; } } - @Override - public void setName(String name) { - this.name = name; - } - - @Override - public void setReportingPeriod(long reportingPeriodMillis) { - this.reportingPeriodMillis = reportingPeriodMillis; - } - - @Override - public void setActivityReporter(ActivityStateReporter activityReporter) { - reporter = activityReporter; - } - @Override public void onActivity(Key key, Supplier newStateSupplier) { if (!initialized) { - throw new IllegalStateException(name + " is not initialized."); + log.error("[{}] Failed to process activity event: activity manager is not initialized.", name); + return; } if (key == null) { - throw new IllegalArgumentException("Activity key can't be null."); + log.error("[{}] Failed to process activity event: provided activity key is null.", name); + return; } if (newStateSupplier == null) { - throw new IllegalArgumentException("New activity state supplier can't be null."); + log.error("[{}] Failed to process activity event: provided new activity state supplier is null.", name); + return; } doOnActivity(key, newStateSupplier); } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityManager.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityManager.java index ba09ccf773..8f006003ef 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityManager.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityManager.java @@ -34,13 +34,7 @@ import java.util.function.Supplier; public interface ActivityManager { - void setName(String name); - - void setReportingPeriod(long reportingPeriodMillis); - - void setActivityReporter(ActivityStateReporter activityReporter); - - void init(); + void init(String name, long reportingPeriodMillis, ActivityStateReporter reporter); void onActivity(Key key, Supplier newStateSupplier);