From 7431233f5d519b814e13c6282a8105d47dfddc3f Mon Sep 17 00:00:00 2001 From: Dmytro Skarzhynets Date: Fri, 15 Dec 2023 09:02:06 +0200 Subject: [PATCH] [WIP] Reimplemented activity management --- .../integration/IntegrationActivityKey.java | 14 +- .../activity/AbstractActivityManager.java | 120 ++++++----- .../transport/activity/ActivityManager.java | 18 +- .../activity/strategy/ActivityStrategy.java | 36 +++- .../strategy/ActivityStrategyFactory.java | 58 +++++ .../strategy/ActivityStrategyType.java | 20 -- .../strategy/AllEventsActivityStrategy.java | 40 +++- .../FirstAndLastEventActivityStrategy.java | 49 ++++- .../strategy/FirstEventActivityStrategy.java | 50 ++++- .../strategy/LastEventActivityStrategy.java | 38 +++- .../service/DefaultTransportService.java | 135 ++++++++---- ...tAndLastEventTransportActivityManager.java | 199 ----------------- ...nlyFirstEventTransportActivityManager.java | 200 ------------------ ...OnlyLastEventTransportActivityManager.java | 149 ------------- 14 files changed, 420 insertions(+), 706 deletions(-) rename common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityStateReporter.java => application/src/main/java/org/thingsboard/server/service/integration/IntegrationActivityKey.java (83%) create mode 100644 common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/ActivityStrategyFactory.java delete mode 100644 common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/ActivityStrategyType.java delete mode 100644 common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/FirstAndLastEventTransportActivityManager.java delete mode 100644 common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/OnlyFirstEventTransportActivityManager.java delete mode 100644 common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/OnlyLastEventTransportActivityManager.java diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityStateReporter.java b/application/src/main/java/org/thingsboard/server/service/integration/IntegrationActivityKey.java similarity index 83% rename from common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityStateReporter.java rename to application/src/main/java/org/thingsboard/server/service/integration/IntegrationActivityKey.java index 1092c569d2..d830d382cb 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityStateReporter.java +++ b/application/src/main/java/org/thingsboard/server/service/integration/IntegrationActivityKey.java @@ -28,11 +28,17 @@ * DOES NOT CONVEY OR IMPLY ANY RIGHTS TO REPRODUCE, DISCLOSE OR DISTRIBUTE ITS CONTENTS, * OR TO MANUFACTURE, USE, OR SELL ANYTHING THAT IT MAY DESCRIBE, IN WHOLE OR IN PART. */ -package org.thingsboard.server.common.transport.activity; +package org.thingsboard.server.service.integration; -@FunctionalInterface -public interface ActivityStateReporter { +import lombok.NonNull; +import lombok.Value; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.TenantId; - void report(Key key, long timeToReport, State activityState, ActivityReportCallback callback); +@Value +public class IntegrationActivityKey { + + @NonNull TenantId tenantId; + @NonNull DeviceId deviceId; } 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 065a0106f3..1c5da90bbd 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 @@ -1,22 +1,22 @@ /** * ThingsBoard, Inc. ("COMPANY") CONFIDENTIAL - *

+ * * Copyright © 2016-2023 ThingsBoard, Inc. All Rights Reserved. - *

+ * * NOTICE: All information contained herein is, and remains * the property of ThingsBoard, Inc. and its suppliers, * if any. The intellectual and technical concepts contained * herein are proprietary to ThingsBoard, Inc. * and its suppliers and may be covered by U.S. and Foreign Patents, * patents in process, and are protected by trade secret or copyright law. - *

+ * * Dissemination of this information or reproduction of this material is strictly forbidden * unless prior written permission is obtained from COMPANY. - *

+ * * Access to the source code contained herein is hereby forbidden to anyone except current COMPANY employees, * managers or contractors who have executed Confidentiality and Non-disclosure agreements * explicitly covering such access. - *

+ * * The copyright notice above does not evidence any actual or intended publication * or disclosure of this source code, which includes * information that is confidential and/or proprietary, and is a trade secret, of COMPANY. @@ -36,7 +36,6 @@ import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.transport.activity.strategy.ActivityStrategy; import java.util.Map; -import java.util.Objects; import java.util.Random; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -44,38 +43,47 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.function.Supplier; @Slf4j -public abstract class AbstractActivityManager implements ActivityManager> { +public abstract class AbstractActivityManager implements ActivityManager { private final ConcurrentMap> states = new ConcurrentHashMap<>(); - private String name; + protected String name; private long reportingPeriodMillis; - private ActivityStateReporter> reporter; private ScheduledExecutorService scheduler; private boolean initialized; @Override - public synchronized void init(String name, long reportingPeriodMillis, ActivityStateReporter> reporter) { + public synchronized void init(String name, long reportingPeriodMillis) { if (!initialized) { this.name = StringUtils.notBlankOrDefault(name, "activity-manager"); - log.info("[{}] initializing.", this.name); - this.reporter = Objects.requireNonNull(reporter, "Failed to initialize activity manager: provided activity reporter is null."); + log.info("Activity manager with name [{}] is initializing.", this.name); 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; - scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName(this.name)); + scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName(this.name)); // TODO: use scheduler component scheduler.scheduleAtFixedRate(this::onReportingPeriodEnd, new Random().nextInt((int) reportingPeriodMillis), reportingPeriodMillis, TimeUnit.MILLISECONDS); initialized = true; - log.info("[{}] initialized.", this.name); + log.info("Activity manager with name [{}] is initialized.", this.name); } } + protected abstract ActivityState createNewState(Key key); + + protected abstract ActivityStrategy getStrategy(); + + protected abstract ActivityState updateState(Key key, ActivityState state); + + protected abstract boolean hasExpired(Key key, ActivityState state); + + protected abstract void onStateExpire(Key key, Metadata metadata); + + protected abstract void reportActivity(Key key, Metadata metadata, long timeToReport, ActivityReportCallback callback); + @Override - public void onActivity(Key key, Supplier> newStateSupplier) { + public void onActivity(Key key) { if (!initialized) { log.error("[{}] Failed to process activity event: activity manager is not initialized.", name); return; @@ -84,39 +92,35 @@ public abstract class AbstractActivityManager implements Activity log.error("[{}] Failed to process activity event: provided activity key is null.", name); return; } - if (newStateSupplier == null) { - log.error("[{}] Failed to process activity event: provided new activity state supplier is null.", name); - return; - } log.debug("[{}] Received activity event for key: [{}]", name, key); - doOnActivity(key, newStateSupplier); - } - - private boolean validate(Key key, Supplier> newStateSupplier) { - - } - - protected abstract ActivityStrategy getStrategy(); - private void doOnActivity(Key key, Supplier> newStateSupplier) { long newLastRecordedTime = System.currentTimeMillis(); - var shouldReport = new AtomicBoolean(false); var activityState = states.compute(key, (__, state) -> { if (state == null) { - state = newStateSupplier.get(); + var newState = createNewState(key); + if (newState == null) { + return null; + } + state = newState; state.setStrategy(getStrategy()); } if (state.getLastRecordedTime() < newLastRecordedTime) { state.setLastRecordedTime(newLastRecordedTime); } - shouldReport.set(state.getStrategy().onActivity(state)); + shouldReport.set(state.getStrategy().onActivity()); return state; }); - if (shouldReport.get()) { + if (activityState == null) { + return; + } + + long lastRecordedTime = activityState.getLastRecordedTime(); + long lastReportedTime = activityState.getLastReportedTime(); + if (shouldReport.get() && lastReportedTime < lastRecordedTime) { log.debug("[{}] Going to report first activity event for key: [{}].", name, key); - reporter.report(key, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() { + reportActivity(key, activityState.getMetadata(), lastRecordedTime, new ActivityReportCallback<>() { @Override public void onSuccess(Key key, long reportedTime) { updateLastReportedTime(key, reportedTime); @@ -130,33 +134,53 @@ public abstract class AbstractActivityManager implements Activity } } + protected long getLastRecordedTime(Key key) { + ActivityState state = states.get(key); + return state == null ? 0L : state.getLastRecordedTime(); + } + private void onReportingPeriodEnd() { log.debug("[{}] Going to end reporting period.", name); for (Map.Entry> entry : states.entrySet()) { var key = entry.getKey(); - var state = entry.getValue(); - long lastRecordedTime = state.getLastRecordedTime(); + var currentState = entry.getValue(); - boolean hasExpired = false; // TODO: implement state expiration + long lastRecordedTime = currentState.getLastRecordedTime(); + long lastReportedTime = currentState.getLastReportedTime(); + var metadata = currentState.getMetadata(); + + boolean hasExpired; boolean shouldReport; - if (hasExpired) { + + var updatedState = updateState(key, currentState); + if (updatedState != null) { + lastRecordedTime = updatedState.getLastRecordedTime(); + lastReportedTime = updatedState.getLastReportedTime(); + metadata = updatedState.getMetadata(); + hasExpired = hasExpired(key, updatedState); + shouldReport = updatedState.getStrategy().onReportingPeriodEnd(); + } else { states.remove(key); + hasExpired = false; shouldReport = true; - } else { - shouldReport = state.getStrategy().onReportingPeriodEnd(state); } - if (shouldReport) { + if (hasExpired) { + states.remove(key); + onStateExpire(key, metadata); + } + + if (shouldReport && lastReportedTime < lastRecordedTime) { log.debug("[{}] Going to report last activity event for key: [{}].", name, key); - reporter.report(key, lastRecordedTime, state, new ActivityReportCallback<>() { + reportActivity(key, metadata, currentState.getLastRecordedTime(), new ActivityReportCallback<>() { @Override - public void onSuccess(Key key, long newLastReportedTime) { - updateLastReportedTime(key, newLastReportedTime); + public void onSuccess(Key key, long reportedTime) { + updateLastReportedTime(key, reportedTime); } @Override public void onFailure(Key key, Throwable t) { - log.debug("[{}] Failed to report last activity event in a period for key: [{}].", name, key, t); + log.debug("[{}] Failed to report last activity event for key: [{}].", name, key, t); } }); } @@ -164,9 +188,9 @@ public abstract class AbstractActivityManager implements Activity } private void updateLastReportedTime(Key key, long newLastReportedTime) { - states.computeIfPresent(key, (__, activityState) -> { - activityState.setLastReportedTime(Math.max(activityState.getLastReportedTime(), newLastReportedTime)); - return activityState; + states.computeIfPresent(key, (__, state) -> { + state.setLastReportedTime(Math.max(state.getLastReportedTime(), newLastReportedTime)); + return state; }); } 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 53228cb1a2..5e659bef1e 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 @@ -1,22 +1,22 @@ /** * ThingsBoard, Inc. ("COMPANY") CONFIDENTIAL - *

+ * * Copyright © 2016-2023 ThingsBoard, Inc. All Rights Reserved. - *

+ * * NOTICE: All information contained herein is, and remains * the property of ThingsBoard, Inc. and its suppliers, * if any. The intellectual and technical concepts contained * herein are proprietary to ThingsBoard, Inc. * and its suppliers and may be covered by U.S. and Foreign Patents, * patents in process, and are protected by trade secret or copyright law. - *

+ * * Dissemination of this information or reproduction of this material is strictly forbidden * unless prior written permission is obtained from COMPANY. - *

+ * * Access to the source code contained herein is hereby forbidden to anyone except current COMPANY employees, * managers or contractors who have executed Confidentiality and Non-disclosure agreements * explicitly covering such access. - *

+ * * The copyright notice above does not evidence any actual or intended publication * or disclosure of this source code, which includes * information that is confidential and/or proprietary, and is a trade secret, of COMPANY. @@ -30,13 +30,11 @@ */ package org.thingsboard.server.common.transport.activity; -import java.util.function.Supplier; +public interface ActivityManager { -public interface ActivityManager { + void init(String name, long reportingPeriodMillis); - void init(String name, long reportingPeriodMillis, ActivityStateReporter> reporter); - - void onActivity(Key key, Supplier> newStateSupplier); + void onActivity(Key key); void destroy(); diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/ActivityStrategy.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/ActivityStrategy.java index 58c8c6d165..1bb8dfdcaf 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/ActivityStrategy.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/ActivityStrategy.java @@ -1,11 +1,39 @@ +/** + * ThingsBoard, Inc. ("COMPANY") CONFIDENTIAL + * + * Copyright © 2016-2023 ThingsBoard, Inc. All Rights Reserved. + * + * NOTICE: All information contained herein is, and remains + * the property of ThingsBoard, Inc. and its suppliers, + * if any. The intellectual and technical concepts contained + * herein are proprietary to ThingsBoard, Inc. + * and its suppliers and may be covered by U.S. and Foreign Patents, + * patents in process, and are protected by trade secret or copyright law. + * + * Dissemination of this information or reproduction of this material is strictly forbidden + * unless prior written permission is obtained from COMPANY. + * + * Access to the source code contained herein is hereby forbidden to anyone except current COMPANY employees, + * managers or contractors who have executed Confidentiality and Non-disclosure agreements + * explicitly covering such access. + * + * The copyright notice above does not evidence any actual or intended publication + * or disclosure of this source code, which includes + * information that is confidential and/or proprietary, and is a trade secret, of COMPANY. + * ANY REPRODUCTION, MODIFICATION, DISTRIBUTION, PUBLIC PERFORMANCE, + * OR PUBLIC DISPLAY OF OR THROUGH USE OF THIS SOURCE CODE WITHOUT + * THE EXPRESS WRITTEN CONSENT OF COMPANY IS STRICTLY PROHIBITED, + * AND IN VIOLATION OF APPLICABLE LAWS AND INTERNATIONAL TREATIES. + * THE RECEIPT OR POSSESSION OF THIS SOURCE CODE AND/OR RELATED INFORMATION + * DOES NOT CONVEY OR IMPLY ANY RIGHTS TO REPRODUCE, DISCLOSE OR DISTRIBUTE ITS CONTENTS, + * OR TO MANUFACTURE, USE, OR SELL ANYTHING THAT IT MAY DESCRIBE, IN WHOLE OR IN PART. + */ package org.thingsboard.server.common.transport.activity.strategy; -import org.thingsboard.server.common.transport.activity.ActivityState; - public interface ActivityStrategy { - boolean onActivity(ActivityState state); + boolean onActivity(); - boolean onReportingPeriodEnd(ActivityState state); + boolean onReportingPeriodEnd(); } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/ActivityStrategyFactory.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/ActivityStrategyFactory.java new file mode 100644 index 0000000000..f4619e2afd --- /dev/null +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/ActivityStrategyFactory.java @@ -0,0 +1,58 @@ +/** + * ThingsBoard, Inc. ("COMPANY") CONFIDENTIAL + * + * Copyright © 2016-2023 ThingsBoard, Inc. All Rights Reserved. + * + * NOTICE: All information contained herein is, and remains + * the property of ThingsBoard, Inc. and its suppliers, + * if any. The intellectual and technical concepts contained + * herein are proprietary to ThingsBoard, Inc. + * and its suppliers and may be covered by U.S. and Foreign Patents, + * patents in process, and are protected by trade secret or copyright law. + * + * Dissemination of this information or reproduction of this material is strictly forbidden + * unless prior written permission is obtained from COMPANY. + * + * Access to the source code contained herein is hereby forbidden to anyone except current COMPANY employees, + * managers or contractors who have executed Confidentiality and Non-disclosure agreements + * explicitly covering such access. + * + * The copyright notice above does not evidence any actual or intended publication + * or disclosure of this source code, which includes + * information that is confidential and/or proprietary, and is a trade secret, of COMPANY. + * ANY REPRODUCTION, MODIFICATION, DISTRIBUTION, PUBLIC PERFORMANCE, + * OR PUBLIC DISPLAY OF OR THROUGH USE OF THIS SOURCE CODE WITHOUT + * THE EXPRESS WRITTEN CONSENT OF COMPANY IS STRICTLY PROHIBITED, + * AND IN VIOLATION OF APPLICABLE LAWS AND INTERNATIONAL TREATIES. + * THE RECEIPT OR POSSESSION OF THIS SOURCE CODE AND/OR RELATED INFORMATION + * DOES NOT CONVEY OR IMPLY ANY RIGHTS TO REPRODUCE, DISCLOSE OR DISTRIBUTE ITS CONTENTS, + * OR TO MANUFACTURE, USE, OR SELL ANYTHING THAT IT MAY DESCRIBE, IN WHOLE OR IN PART. + */ +package org.thingsboard.server.common.transport.activity.strategy; + +public final class ActivityStrategyFactory { + + private static final String ALL_EVENTS_STRATEGY_NAME = "ALL"; + private static final String FIRST_EVENT_STRATEGY_NAME = "FIRST"; + private static final String LAST_EVENT_STRATEGY_NAME = "LAST"; + private static final String FIRST_AND_LAST_STRATEGY_NAME = "FIRST_AND_LAST"; + + private ActivityStrategyFactory() { + } + + public static ActivityStrategy createStrategy(String strategyName) { + switch (strategyName) { + case ALL_EVENTS_STRATEGY_NAME: + return new AllEventsActivityStrategy(); + case FIRST_EVENT_STRATEGY_NAME: + return new FirstEventActivityStrategy(); + case LAST_EVENT_STRATEGY_NAME: + return new LastEventActivityStrategy(); + case FIRST_AND_LAST_STRATEGY_NAME: + return new FirstAndLastEventActivityStrategy(); + default: + throw new IllegalArgumentException("Unknown activity strategy name: [" + strategyName + "."); + } + } + +} diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/ActivityStrategyType.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/ActivityStrategyType.java deleted file mode 100644 index b351d4a59a..0000000000 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/ActivityStrategyType.java +++ /dev/null @@ -1,20 +0,0 @@ -package org.thingsboard.server.common.transport.activity.strategy; - -public enum ActivityStrategyType { - - FIRST(new FirstEventActivityStrategy()), - LAST(new LastEventActivityStrategy()), - FIRST_AND_LAST(new FirstAndLastEventActivityStrategy()), - ALL(new AllEventsActivityStrategy()); - - private final ActivityStrategy strategy; - - ActivityStrategyType(ActivityStrategy strategy) { - this.strategy = strategy; - } - - public ActivityStrategy getStrategy() { - return strategy; - } - -} diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/AllEventsActivityStrategy.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/AllEventsActivityStrategy.java index add491c3fa..27797c5315 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/AllEventsActivityStrategy.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/AllEventsActivityStrategy.java @@ -1,17 +1,45 @@ +/** + * ThingsBoard, Inc. ("COMPANY") CONFIDENTIAL + * + * Copyright © 2016-2023 ThingsBoard, Inc. All Rights Reserved. + * + * NOTICE: All information contained herein is, and remains + * the property of ThingsBoard, Inc. and its suppliers, + * if any. The intellectual and technical concepts contained + * herein are proprietary to ThingsBoard, Inc. + * and its suppliers and may be covered by U.S. and Foreign Patents, + * patents in process, and are protected by trade secret or copyright law. + * + * Dissemination of this information or reproduction of this material is strictly forbidden + * unless prior written permission is obtained from COMPANY. + * + * Access to the source code contained herein is hereby forbidden to anyone except current COMPANY employees, + * managers or contractors who have executed Confidentiality and Non-disclosure agreements + * explicitly covering such access. + * + * The copyright notice above does not evidence any actual or intended publication + * or disclosure of this source code, which includes + * information that is confidential and/or proprietary, and is a trade secret, of COMPANY. + * ANY REPRODUCTION, MODIFICATION, DISTRIBUTION, PUBLIC PERFORMANCE, + * OR PUBLIC DISPLAY OF OR THROUGH USE OF THIS SOURCE CODE WITHOUT + * THE EXPRESS WRITTEN CONSENT OF COMPANY IS STRICTLY PROHIBITED, + * AND IN VIOLATION OF APPLICABLE LAWS AND INTERNATIONAL TREATIES. + * THE RECEIPT OR POSSESSION OF THIS SOURCE CODE AND/OR RELATED INFORMATION + * DOES NOT CONVEY OR IMPLY ANY RIGHTS TO REPRODUCE, DISCLOSE OR DISTRIBUTE ITS CONTENTS, + * OR TO MANUFACTURE, USE, OR SELL ANYTHING THAT IT MAY DESCRIBE, IN WHOLE OR IN PART. + */ package org.thingsboard.server.common.transport.activity.strategy; -import org.thingsboard.server.common.transport.activity.ActivityState; - public class AllEventsActivityStrategy implements ActivityStrategy { @Override - public boolean onActivity(ActivityState state) { - return state.getLastReportedTime() < state.getLastReportedTime(); + public boolean onActivity() { + return true; } @Override - public boolean onReportingPeriodEnd(ActivityState state) { - return state.getLastReportedTime() < state.getLastReportedTime(); + public boolean onReportingPeriodEnd() { + return true; } } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/FirstAndLastEventActivityStrategy.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/FirstAndLastEventActivityStrategy.java index ecfbaff005..c2605cdeaf 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/FirstAndLastEventActivityStrategy.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/FirstAndLastEventActivityStrategy.java @@ -1,25 +1,52 @@ +/** + * ThingsBoard, Inc. ("COMPANY") CONFIDENTIAL + * + * Copyright © 2016-2023 ThingsBoard, Inc. All Rights Reserved. + * + * NOTICE: All information contained herein is, and remains + * the property of ThingsBoard, Inc. and its suppliers, + * if any. The intellectual and technical concepts contained + * herein are proprietary to ThingsBoard, Inc. + * and its suppliers and may be covered by U.S. and Foreign Patents, + * patents in process, and are protected by trade secret or copyright law. + * + * Dissemination of this information or reproduction of this material is strictly forbidden + * unless prior written permission is obtained from COMPANY. + * + * Access to the source code contained herein is hereby forbidden to anyone except current COMPANY employees, + * managers or contractors who have executed Confidentiality and Non-disclosure agreements + * explicitly covering such access. + * + * The copyright notice above does not evidence any actual or intended publication + * or disclosure of this source code, which includes + * information that is confidential and/or proprietary, and is a trade secret, of COMPANY. + * ANY REPRODUCTION, MODIFICATION, DISTRIBUTION, PUBLIC PERFORMANCE, + * OR PUBLIC DISPLAY OF OR THROUGH USE OF THIS SOURCE CODE WITHOUT + * THE EXPRESS WRITTEN CONSENT OF COMPANY IS STRICTLY PROHIBITED, + * AND IN VIOLATION OF APPLICABLE LAWS AND INTERNATIONAL TREATIES. + * THE RECEIPT OR POSSESSION OF THIS SOURCE CODE AND/OR RELATED INFORMATION + * DOES NOT CONVEY OR IMPLY ANY RIGHTS TO REPRODUCE, DISCLOSE OR DISTRIBUTE ITS CONTENTS, + * OR TO MANUFACTURE, USE, OR SELL ANYTHING THAT IT MAY DESCRIBE, IN WHOLE OR IN PART. + */ package org.thingsboard.server.common.transport.activity.strategy; -import org.thingsboard.server.common.transport.activity.ActivityState; - public class FirstAndLastEventActivityStrategy implements ActivityStrategy { - private volatile long firstEventTs; + private boolean firstEventReceived; @Override - public boolean onActivity(ActivityState state) { - long lastRecordedTime = state.getLastRecordedTime(); - if (firstEventTs == 0L) { - firstEventTs = lastRecordedTime; - return state.getLastReportedTime() < firstEventTs; + public synchronized boolean onActivity() { + if (!firstEventReceived) { + firstEventReceived = true; + return true; } return false; } @Override - public boolean onReportingPeriodEnd(ActivityState state) { - firstEventTs = 0L; - return state.getLastReportedTime() < state.getLastRecordedTime(); + public synchronized boolean onReportingPeriodEnd() { + firstEventReceived = false; + return true; } } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/FirstEventActivityStrategy.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/FirstEventActivityStrategy.java index 2b787c3251..d837591e04 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/FirstEventActivityStrategy.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/FirstEventActivityStrategy.java @@ -1,24 +1,54 @@ +/** + * ThingsBoard, Inc. ("COMPANY") CONFIDENTIAL + * + * Copyright © 2016-2023 ThingsBoard, Inc. All Rights Reserved. + * + * NOTICE: All information contained herein is, and remains + * the property of ThingsBoard, Inc. and its suppliers, + * if any. The intellectual and technical concepts contained + * herein are proprietary to ThingsBoard, Inc. + * and its suppliers and may be covered by U.S. and Foreign Patents, + * patents in process, and are protected by trade secret or copyright law. + * + * Dissemination of this information or reproduction of this material is strictly forbidden + * unless prior written permission is obtained from COMPANY. + * + * Access to the source code contained herein is hereby forbidden to anyone except current COMPANY employees, + * managers or contractors who have executed Confidentiality and Non-disclosure agreements + * explicitly covering such access. + * + * The copyright notice above does not evidence any actual or intended publication + * or disclosure of this source code, which includes + * information that is confidential and/or proprietary, and is a trade secret, of COMPANY. + * ANY REPRODUCTION, MODIFICATION, DISTRIBUTION, PUBLIC PERFORMANCE, + * OR PUBLIC DISPLAY OF OR THROUGH USE OF THIS SOURCE CODE WITHOUT + * THE EXPRESS WRITTEN CONSENT OF COMPANY IS STRICTLY PROHIBITED, + * AND IN VIOLATION OF APPLICABLE LAWS AND INTERNATIONAL TREATIES. + * THE RECEIPT OR POSSESSION OF THIS SOURCE CODE AND/OR RELATED INFORMATION + * DOES NOT CONVEY OR IMPLY ANY RIGHTS TO REPRODUCE, DISCLOSE OR DISTRIBUTE ITS CONTENTS, + * OR TO MANUFACTURE, USE, OR SELL ANYTHING THAT IT MAY DESCRIBE, IN WHOLE OR IN PART. + */ package org.thingsboard.server.common.transport.activity.strategy; -import org.thingsboard.server.common.transport.activity.ActivityState; - public class FirstEventActivityStrategy implements ActivityStrategy { - private volatile long firstEventTs; + private boolean firstEventReceived; @Override - public boolean onActivity(ActivityState state) { - long lastRecordedTime = state.getLastRecordedTime(); - if (firstEventTs == 0L) { - firstEventTs = lastRecordedTime; - return state.getLastReportedTime() < firstEventTs; + public synchronized boolean onActivity() { + if (!firstEventReceived) { + firstEventReceived = true; + return true; } return false; } @Override - public boolean onReportingPeriodEnd(ActivityState state) { - firstEventTs = 0L; + public synchronized boolean onReportingPeriodEnd() { + if (!firstEventReceived) { + return true; + } + firstEventReceived = false; return false; } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/LastEventActivityStrategy.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/LastEventActivityStrategy.java index 57260ed5ca..e22263096e 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/LastEventActivityStrategy.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/LastEventActivityStrategy.java @@ -1,17 +1,45 @@ +/** + * ThingsBoard, Inc. ("COMPANY") CONFIDENTIAL + * + * Copyright © 2016-2023 ThingsBoard, Inc. All Rights Reserved. + * + * NOTICE: All information contained herein is, and remains + * the property of ThingsBoard, Inc. and its suppliers, + * if any. The intellectual and technical concepts contained + * herein are proprietary to ThingsBoard, Inc. + * and its suppliers and may be covered by U.S. and Foreign Patents, + * patents in process, and are protected by trade secret or copyright law. + * + * Dissemination of this information or reproduction of this material is strictly forbidden + * unless prior written permission is obtained from COMPANY. + * + * Access to the source code contained herein is hereby forbidden to anyone except current COMPANY employees, + * managers or contractors who have executed Confidentiality and Non-disclosure agreements + * explicitly covering such access. + * + * The copyright notice above does not evidence any actual or intended publication + * or disclosure of this source code, which includes + * information that is confidential and/or proprietary, and is a trade secret, of COMPANY. + * ANY REPRODUCTION, MODIFICATION, DISTRIBUTION, PUBLIC PERFORMANCE, + * OR PUBLIC DISPLAY OF OR THROUGH USE OF THIS SOURCE CODE WITHOUT + * THE EXPRESS WRITTEN CONSENT OF COMPANY IS STRICTLY PROHIBITED, + * AND IN VIOLATION OF APPLICABLE LAWS AND INTERNATIONAL TREATIES. + * THE RECEIPT OR POSSESSION OF THIS SOURCE CODE AND/OR RELATED INFORMATION + * DOES NOT CONVEY OR IMPLY ANY RIGHTS TO REPRODUCE, DISCLOSE OR DISTRIBUTE ITS CONTENTS, + * OR TO MANUFACTURE, USE, OR SELL ANYTHING THAT IT MAY DESCRIBE, IN WHOLE OR IN PART. + */ package org.thingsboard.server.common.transport.activity.strategy; -import org.thingsboard.server.common.transport.activity.ActivityState; - public class LastEventActivityStrategy implements ActivityStrategy { @Override - public boolean onActivity(ActivityState state) { + public boolean onActivity() { return false; } @Override - public boolean onReportingPeriodEnd(ActivityState state) { - return state.getLastReportedTime() < state.getLastRecordedTime(); + public boolean onReportingPeriodEnd() { + return true; } } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index a86bdda312..5646e35cec 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java @@ -23,7 +23,6 @@ import com.google.gson.JsonObject; import com.google.protobuf.ByteString; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.annotation.Lazy; @@ -72,8 +71,11 @@ import org.thingsboard.server.common.transport.TransportResourceCache; import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportServiceCallback; import org.thingsboard.server.common.transport.TransportTenantProfileCache; -import org.thingsboard.server.common.transport.activity.ActivityManager; -import org.thingsboard.server.common.transport.activity.ActivityStateReporter; +import org.thingsboard.server.common.transport.activity.AbstractActivityManager; +import org.thingsboard.server.common.transport.activity.ActivityReportCallback; +import org.thingsboard.server.common.transport.activity.ActivityState; +import org.thingsboard.server.common.transport.activity.strategy.ActivityStrategy; +import org.thingsboard.server.common.transport.activity.strategy.ActivityStrategyFactory; import org.thingsboard.server.common.transport.auth.GetOrCreateDeviceFromGatewayResponse; import org.thingsboard.server.common.transport.auth.TransportDeviceInfo; import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse; @@ -132,7 +134,7 @@ import java.util.stream.Collectors; @Slf4j @Service @TbTransportComponent -public class DefaultTransportService implements TransportService { +public class DefaultTransportService extends AbstractActivityManager implements TransportService { public static final String OVERWRITE_ACTIVITY_TIME = "overwriteActivityTime"; public static final String SESSION_EXPIRED_MESSAGE = "Session has expired due to last activity time!"; @@ -156,6 +158,8 @@ public class DefaultTransportService implements TransportService { private long sessionInactivityTimeout; @Value("${transport.sessions.report_timeout}") private long sessionReportTimeout; + @Value("${transport.activity.reporting_strategy:LAST}") + private String reportingStrategyName; @Value("${transport.client_side_rpc.timeout:60000}") private long clientSideRpcTimeout; @Value("${queue.transport.poll_interval}") @@ -187,7 +191,6 @@ public class DefaultTransportService implements TransportService { private final TransportResourceCache transportResourceCache; private final NotificationRuleProcessor notificationRuleProcessor; private final EntityLimitsCache entityLimitsCache; - private ActivityManager activityManager; protected TbQueueRequestTemplate, TbProtoQueueMsg> transportApiRequestTemplate; protected TbQueueProducer> ruleEngineMsgProducer; @@ -235,13 +238,9 @@ public class DefaultTransportService implements TransportService { this.entityLimitsCache = entityLimitsCache; } - @Autowired - public void setActivityManager(ActivityManager activityManager) { - this.activityManager = activityManager; - } - @PostConstruct public void init() { + super.init(ACTIVITY_MANAGER_NAME, sessionReportTimeout); this.ruleEngineProducerStats = statsFactory.createMessagesStats(StatsType.RULE_ENGINE.getName() + ".producer"); this.tbCoreProducerStats = statsFactory.createMessagesStats(StatsType.CORE.getName() + ".producer"); this.transportApiStats = statsFactory.createMessagesStats(StatsType.TRANSPORT.getName() + ".producer"); @@ -256,31 +255,8 @@ public class DefaultTransportService implements TransportService { transportNotificationsConsumer.subscribe(Collections.singleton(tpi)); transportApiRequestTemplate.init(); mainConsumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("transport-consumer")); - activityManager.init(ACTIVITY_MANAGER_NAME, sessionReportTimeout, activityReporter); } - private final ActivityStateReporter activityReporter = (sessionId, timeToReport, state, reportCallback) -> { - log.debug("[{}] Reporting activity state for session with id: [{}]. Time to report: [{}].", ACTIVITY_MANAGER_NAME, sessionId, timeToReport); - SessionMetaData sessionMetaData = sessions.get(sessionId); - TransportProtos.SubscriptionInfoProto subscriptionInfo = TransportProtos.SubscriptionInfoProto.newBuilder() - .setAttributeSubscription(sessionMetaData != null && sessionMetaData.isSubscribedToAttributes()) - .setRpcSubscription(sessionMetaData != null && sessionMetaData.isSubscribedToRPC()) - .setLastActivityTime(timeToReport) - .build(); - process(state.getSessionInfoProto(), subscriptionInfo, new TransportServiceCallback<>() { - @Override - public void onSuccess(Void msgAcknowledged) { - reportCallback.onSuccess(sessionId, timeToReport); - - } - - @Override - public void onError(Throwable e) { - reportCallback.onFailure(sessionId, e); - } - }); - }; - @AfterStartUp(order = AfterStartUp.TRANSPORT_SERVICE) public void start() { mainConsumerExecutor.execute(() -> { @@ -319,6 +295,7 @@ public class DefaultTransportService implements TransportService { @PreDestroy public void destroy() { stopped = true; + super.destroy(); if (transportNotificationsConsumer != null) { transportNotificationsConsumer.unsubscribe(); @@ -332,9 +309,6 @@ public class DefaultTransportService implements TransportService { if (transportApiRequestTemplate != null) { transportApiRequestTemplate.stop(); } - if (activityManager != null) { - activityManager.destroy(); - } } @Override @@ -814,10 +788,91 @@ public class DefaultTransportService implements TransportService { } private void reportActivityInternal(TransportProtos.SessionInfoProto sessionInfo) { - activityManager.onActivity(toSessionId(sessionInfo), () -> { - var activityState = new TransportActivityState(); - activityState.setSessionInfoProto(sessionInfo); - return activityState; + onActivity(toSessionId(sessionInfo)); + } + + @Override + protected ActivityState createNewState(UUID sessionId) { + SessionMetaData session = sessions.get(sessionId); + if (session == null) { + return null; + } + ActivityState state = new ActivityState<>(); + state.setMetadata(session.getSessionInfo()); + return state; + } + + @Override + protected ActivityStrategy getStrategy() { + return ActivityStrategyFactory.createStrategy(reportingStrategyName); + } + + @Override + protected ActivityState updateState(UUID sessionId, ActivityState state) { + SessionMetaData session = sessions.get(sessionId); + if (session == null) { + return null; + } + + state.setMetadata(session.getSessionInfo()); + var sessionInfo = state.getMetadata(); + + if (sessionInfo.getGwSessionIdMSB() == 0L || sessionInfo.getCustomerIdLSB() == 0L) { + return state; + } + + var gwSessionId = new UUID(sessionInfo.getGwSessionIdMSB(), sessionInfo.getGwSessionIdLSB()); + SessionMetaData gwSession = sessions.get(gwSessionId); + if (gwSession == null || !gwSession.isOverwriteActivityTime()) { + return state; + } + + long lastRecordedTime = state.getLastRecordedTime(); + long gwLastRecordedTime = getLastRecordedTime(gwSessionId); + log.debug("[{}] Session with id: [{}] has gateway session with id: [{}] with overwrite activity time enabled. " + + "Updating last activity time. Session last recorded time: [{}], gateway session last recorded time: [{}].", + name, sessionId, gwSessionId, lastRecordedTime, gwLastRecordedTime); + state.setLastRecordedTime(Math.max(lastRecordedTime, gwLastRecordedTime)); + return state; + } + + @Override + protected boolean hasExpired(UUID uuid, ActivityState state) { + return (System.currentTimeMillis() - sessionInactivityTimeout) < state.getLastReportedTime(); + } + + @Override + protected void onStateExpire(UUID sessionId, TransportProtos.SessionInfoProto sessionInfo) { + log.debug("[{}] Session with id: [{}] has expired due to last activity time.", name, sessionId); + SessionMetaData expiredSession = sessions.remove(sessionId); + if (expiredSession != null) { + deregisterSession(sessionInfo); + process(sessionInfo, SESSION_EVENT_MSG_CLOSED, null); + expiredSession.getListener().onRemoteSessionCloseCommand(sessionId, SESSION_EXPIRED_NOTIFICATION_PROTO); + } + } + + @Override + protected void reportActivity(UUID sessionId, TransportProtos.SessionInfoProto currentSessionInfo, long timeToReport, ActivityReportCallback callback) { + log.debug("[{}] Reporting activity state for session with id: [{}]. Time to report: [{}].", name, sessionId, timeToReport); + SessionMetaData session = sessions.get(sessionId); + TransportProtos.SubscriptionInfoProto subscriptionInfo = TransportProtos.SubscriptionInfoProto.newBuilder() + .setAttributeSubscription(session != null && session.isSubscribedToAttributes()) + .setRpcSubscription(session != null && session.isSubscribedToRPC()) + .setLastActivityTime(timeToReport) + .build(); + TransportProtos.SessionInfoProto sessionInfo = session != null ? session.getSessionInfo() : currentSessionInfo; + process(sessionInfo, subscriptionInfo, new TransportServiceCallback<>() { + @Override + public void onSuccess(Void msgAcknowledged) { + callback.onSuccess(sessionId, timeToReport); + + } + + @Override + public void onError(Throwable e) { + callback.onFailure(sessionId, e); + } }); } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/FirstAndLastEventTransportActivityManager.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/FirstAndLastEventTransportActivityManager.java deleted file mode 100644 index 9c582b2587..0000000000 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/FirstAndLastEventTransportActivityManager.java +++ /dev/null @@ -1,199 +0,0 @@ -/** - * ThingsBoard, Inc. ("COMPANY") CONFIDENTIAL - * - * Copyright © 2016-2023 ThingsBoard, Inc. All Rights Reserved. - * - * NOTICE: All information contained herein is, and remains - * the property of ThingsBoard, Inc. and its suppliers, - * if any. The intellectual and technical concepts contained - * herein are proprietary to ThingsBoard, Inc. - * and its suppliers and may be covered by U.S. and Foreign Patents, - * patents in process, and are protected by trade secret or copyright law. - * - * Dissemination of this information or reproduction of this material is strictly forbidden - * unless prior written permission is obtained from COMPANY. - * - * Access to the source code contained herein is hereby forbidden to anyone except current COMPANY employees, - * managers or contractors who have executed Confidentiality and Non-disclosure agreements - * explicitly covering such access. - * - * The copyright notice above does not evidence any actual or intended publication - * or disclosure of this source code, which includes - * information that is confidential and/or proprietary, and is a trade secret, of COMPANY. - * ANY REPRODUCTION, MODIFICATION, DISTRIBUTION, PUBLIC PERFORMANCE, - * OR PUBLIC DISPLAY OF OR THROUGH USE OF THIS SOURCE CODE WITHOUT - * THE EXPRESS WRITTEN CONSENT OF COMPANY IS STRICTLY PROHIBITED, - * AND IN VIOLATION OF APPLICABLE LAWS AND INTERNATIONAL TREATIES. - * THE RECEIPT OR POSSESSION OF THIS SOURCE CODE AND/OR RELATED INFORMATION - * DOES NOT CONVEY OR IMPLY ANY RIGHTS TO REPRODUCE, DISCLOSE OR DISTRIBUTE ITS CONTENTS, - * OR TO MANUFACTURE, USE, OR SELL ANYTHING THAT IT MAY DESCRIBE, IN WHOLE OR IN PART. - */ -package org.thingsboard.server.common.transport.service.activity; - -import com.google.common.util.concurrent.FutureCallback; -import com.google.common.util.concurrent.Futures; -import com.google.common.util.concurrent.MoreExecutors; -import com.google.common.util.concurrent.SettableFuture; -import lombok.Data; -import lombok.extern.slf4j.Slf4j; -import org.checkerframework.checker.nullness.qual.NonNull; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.data.util.Pair; -import org.springframework.stereotype.Component; -import org.thingsboard.server.common.transport.TransportService; -import org.thingsboard.server.common.transport.activity.AbstractActivityManager; -import org.thingsboard.server.common.transport.activity.ActivityReportCallback; -import org.thingsboard.server.common.transport.service.SessionMetaData; -import org.thingsboard.server.common.transport.service.TransportActivityState; -import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.queue.util.TbTransportComponent; - -import java.util.HashSet; -import java.util.Map; -import java.util.Set; -import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.function.Supplier; - -import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EVENT_MSG_CLOSED; -import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EXPIRED_NOTIFICATION_PROTO; - -@Slf4j -@Component -@TbTransportComponent -@ConditionalOnProperty(prefix = "transport.activity", value = "reporting_strategy", havingValue = "first-and-last") -public class FirstAndLastEventTransportActivityManager extends AbstractActivityManager { - - private final ConcurrentMap states = new ConcurrentHashMap<>(); - - @Data - private static class ActivityStateWrapper { - - volatile TransportActivityState state; - volatile boolean alreadyBeenReported; - - } - - @Value("${transport.sessions.inactivity_timeout}") - private long sessionInactivityTimeout; - - @Autowired - private TransportService transportService; - - @Override - protected void doOnActivity(UUID sessionId, Supplier newStateSupplier) { - long newLastRecordedTime = System.currentTimeMillis(); - SettableFuture> reportCompletedFuture = SettableFuture.create(); - states.compute(sessionId, (key, activityStateWrapper) -> { - if (activityStateWrapper == null) { - activityStateWrapper = new ActivityStateWrapper(); - activityStateWrapper.setState(newStateSupplier.get()); - } - var activityState = activityStateWrapper.getState(); - if (activityState.getLastRecordedTime() < newLastRecordedTime) { - activityState.setLastRecordedTime(newLastRecordedTime); - } - if (activityStateWrapper.isAlreadyBeenReported()) { - return activityStateWrapper; - } - if (activityState.getLastReportedTime() < activityState.getLastRecordedTime()) { - log.debug("[{}] Going to report first activity event for session with id: [{}]", name, sessionId); - reporter.report(key, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() { - @Override - public void onSuccess(UUID key, long reportedTime) { - reportCompletedFuture.set(Pair.of(key, reportedTime)); - } - - @Override - public void onFailure(UUID key, Throwable t) { - reportCompletedFuture.setException(t); - } - }); - } - activityStateWrapper.setAlreadyBeenReported(true); - return activityStateWrapper; - }); - Futures.addCallback(reportCompletedFuture, new FutureCallback<>() { - @Override - public void onSuccess(Pair reportResult) { - updateLastReportedTime(reportResult.getFirst(), reportResult.getSecond()); - } - - @Override - public void onFailure(@NonNull Throwable t) { - log.debug("[{}] Failed to report first activity event for session with id: [{}]", name, sessionId); - } - }, MoreExecutors.directExecutor()); - } - - @Override - protected void doOnReportingPeriodEnd() { - Set statesToRemove = new HashSet<>(); - for (Map.Entry entry : states.entrySet()) { - var sessionId = entry.getKey(); - var activityStateWrapper = entry.getValue(); - var activityState = activityStateWrapper.getState(); - - SessionMetaData sessionMetaData = transportService.getSession(sessionId); - if (sessionMetaData != null) { - activityState.setSessionInfoProto(sessionMetaData.getSessionInfo()); - } else { - log.debug("[{}] Session with id: [{}] is not present. Marking it's activity state for removal.", name, sessionId); - statesToRemove.add(sessionId); - } - - long lastActivityTime = activityState.getLastRecordedTime(); - TransportProtos.SessionInfoProto sessionInfo = activityState.getSessionInfoProto(); - - if (sessionInfo.getGwSessionIdMSB() != 0 && sessionInfo.getGwSessionIdLSB() != 0) { - var gwSessionId = new UUID(sessionInfo.getGwSessionIdMSB(), sessionInfo.getGwSessionIdLSB()); - SessionMetaData gwSessionMetaData = transportService.getSession(gwSessionId); - if (gwSessionMetaData != null && gwSessionMetaData.isOverwriteActivityTime()) { - ActivityStateWrapper gwActivityStateWrapper = states.get(gwSessionId); - if (gwActivityStateWrapper != null) { - log.debug("[{}] Session with id: [{}] has gateway session with id: [{}] with overwrite activity time enabled. Updating last activity time.", name, sessionId, gwSessionId); - lastActivityTime = Math.max(gwActivityStateWrapper.getState().getLastRecordedTime(), lastActivityTime); - } - } - } - - long expirationTime = System.currentTimeMillis() - sessionInactivityTimeout; - boolean hasExpired = sessionMetaData != null && lastActivityTime < expirationTime; - if (hasExpired) { - log.debug("[{}] Session with id: [{}] has expired due to last activity time: [{}]. Marking it's activity state for removal.", name, sessionId, lastActivityTime); - transportService.deregisterSession(sessionInfo); - statesToRemove.add(sessionId); - transportService.process(sessionInfo, SESSION_EVENT_MSG_CLOSED, null); - sessionMetaData.getListener().onRemoteSessionCloseCommand(sessionId, SESSION_EXPIRED_NOTIFICATION_PROTO); - } - if (activityState.getLastReportedTime() < lastActivityTime) { - log.debug("[{}] Going to report last activity event for session with id: [{}].", name, sessionId); - reporter.report(sessionId, lastActivityTime, activityState, new ActivityReportCallback<>() { - @Override - public void onSuccess(UUID key, long reportedTime) { - updateLastReportedTime(key, reportedTime); - } - - @Override - public void onFailure(UUID key, Throwable t) { - log.debug("[{}] Failed to report last activity event for session with id: [{}].", name, sessionId); - } - }); - } - activityStateWrapper.setAlreadyBeenReported(false); - } - statesToRemove.forEach(states::remove); - } - - private void updateLastReportedTime(UUID key, long newLastReportedTime) { - states.computeIfPresent(key, (__, activityStateWrapper) -> { - var activityState = activityStateWrapper.getState(); - activityState.setLastReportedTime(Math.max(activityState.getLastReportedTime(), newLastReportedTime)); - return activityStateWrapper; - }); - } - -} diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/OnlyFirstEventTransportActivityManager.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/OnlyFirstEventTransportActivityManager.java deleted file mode 100644 index 19a6314450..0000000000 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/OnlyFirstEventTransportActivityManager.java +++ /dev/null @@ -1,200 +0,0 @@ -/** - * ThingsBoard, Inc. ("COMPANY") CONFIDENTIAL - * - * Copyright © 2016-2023 ThingsBoard, Inc. All Rights Reserved. - * - * NOTICE: All information contained herein is, and remains - * the property of ThingsBoard, Inc. and its suppliers, - * if any. The intellectual and technical concepts contained - * herein are proprietary to ThingsBoard, Inc. - * and its suppliers and may be covered by U.S. and Foreign Patents, - * patents in process, and are protected by trade secret or copyright law. - * - * Dissemination of this information or reproduction of this material is strictly forbidden - * unless prior written permission is obtained from COMPANY. - * - * Access to the source code contained herein is hereby forbidden to anyone except current COMPANY employees, - * managers or contractors who have executed Confidentiality and Non-disclosure agreements - * explicitly covering such access. - * - * The copyright notice above does not evidence any actual or intended publication - * or disclosure of this source code, which includes - * information that is confidential and/or proprietary, and is a trade secret, of COMPANY. - * ANY REPRODUCTION, MODIFICATION, DISTRIBUTION, PUBLIC PERFORMANCE, - * OR PUBLIC DISPLAY OF OR THROUGH USE OF THIS SOURCE CODE WITHOUT - * THE EXPRESS WRITTEN CONSENT OF COMPANY IS STRICTLY PROHIBITED, - * AND IN VIOLATION OF APPLICABLE LAWS AND INTERNATIONAL TREATIES. - * THE RECEIPT OR POSSESSION OF THIS SOURCE CODE AND/OR RELATED INFORMATION - * DOES NOT CONVEY OR IMPLY ANY RIGHTS TO REPRODUCE, DISCLOSE OR DISTRIBUTE ITS CONTENTS, - * OR TO MANUFACTURE, USE, OR SELL ANYTHING THAT IT MAY DESCRIBE, IN WHOLE OR IN PART. - */ -package org.thingsboard.server.common.transport.service.activity; - -import com.google.common.util.concurrent.FutureCallback; -import com.google.common.util.concurrent.Futures; -import com.google.common.util.concurrent.MoreExecutors; -import com.google.common.util.concurrent.SettableFuture; -import lombok.Data; -import lombok.extern.slf4j.Slf4j; -import org.checkerframework.checker.nullness.qual.NonNull; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.data.util.Pair; -import org.springframework.stereotype.Component; -import org.thingsboard.server.common.transport.TransportService; -import org.thingsboard.server.common.transport.activity.AbstractActivityManager; -import org.thingsboard.server.common.transport.activity.ActivityReportCallback; -import org.thingsboard.server.common.transport.service.SessionMetaData; -import org.thingsboard.server.common.transport.service.TransportActivityState; -import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.queue.util.TbTransportComponent; - -import java.util.HashSet; -import java.util.Map; -import java.util.Set; -import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.function.Supplier; - -import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EVENT_MSG_CLOSED; -import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EXPIRED_NOTIFICATION_PROTO; - -@Slf4j -@Component -@TbTransportComponent -@ConditionalOnProperty(prefix = "transport.activity", value = "reporting_strategy", havingValue = "first") -public class OnlyFirstEventTransportActivityManager extends AbstractActivityManager { - - private final ConcurrentMap states = new ConcurrentHashMap<>(); - - @Data - private static class ActivityStateWrapper { - - volatile TransportActivityState state; - volatile boolean alreadyBeenReported; - - } - - @Value("${transport.sessions.inactivity_timeout}") - private long sessionInactivityTimeout; - - @Autowired - private TransportService transportService; - - @Override - protected void doOnActivity(UUID sessionId, Supplier newStateSupplier) { - long newLastRecordedTime = System.currentTimeMillis(); - SettableFuture> reportCompletedFuture = SettableFuture.create(); - states.compute(sessionId, (key, activityStateWrapper) -> { - if (activityStateWrapper == null) { - activityStateWrapper = new ActivityStateWrapper(); - activityStateWrapper.setState(newStateSupplier.get()); - } - var activityState = activityStateWrapper.getState(); - if (activityState.getLastRecordedTime() < newLastRecordedTime) { - activityState.setLastRecordedTime(newLastRecordedTime); - } - if (activityStateWrapper.isAlreadyBeenReported()) { - return activityStateWrapper; - } - if (activityState.getLastReportedTime() < activityState.getLastRecordedTime()) { - log.debug("[{}] Going to report first activity event for session with id: [{}]", name, sessionId); - reporter.report(key, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() { - @Override - public void onSuccess(UUID key, long reportedTime) { - reportCompletedFuture.set(Pair.of(key, reportedTime)); - } - - @Override - public void onFailure(UUID key, Throwable t) { - reportCompletedFuture.setException(t); - } - }); - } - activityStateWrapper.setAlreadyBeenReported(true); - return activityStateWrapper; - }); - Futures.addCallback(reportCompletedFuture, new FutureCallback<>() { - @Override - public void onSuccess(Pair reportResult) { - updateLastReportedTime(reportResult.getFirst(), reportResult.getSecond()); - } - - @Override - public void onFailure(@NonNull Throwable t) { - log.debug("[{}] Failed to report first activity event for session with id: [{}]", name, sessionId); - } - }, MoreExecutors.directExecutor()); - } - - @Override - protected void doOnReportingPeriodEnd() { - Set statesToRemove = new HashSet<>(); - for (Map.Entry entry : states.entrySet()) { - var sessionId = entry.getKey(); - var activityStateWrapper = entry.getValue(); - var activityState = activityStateWrapper.getState(); - - SessionMetaData sessionMetaData = transportService.getSession(sessionId); - if (sessionMetaData != null) { - activityState.setSessionInfoProto(sessionMetaData.getSessionInfo()); - } else { - log.debug("[{}] Session with id: [{}] is not present. Marking it's activity state for removal.", name, sessionId); - statesToRemove.add(sessionId); - } - - long lastActivityTime = activityState.getLastRecordedTime(); - TransportProtos.SessionInfoProto sessionInfo = activityState.getSessionInfoProto(); - - if (sessionInfo.getGwSessionIdMSB() != 0 && sessionInfo.getGwSessionIdLSB() != 0) { - var gwSessionId = new UUID(sessionInfo.getGwSessionIdMSB(), sessionInfo.getGwSessionIdLSB()); - SessionMetaData gwSessionMetaData = transportService.getSession(gwSessionId); - if (gwSessionMetaData != null && gwSessionMetaData.isOverwriteActivityTime()) { - ActivityStateWrapper gwActivityStateWrapper = states.get(gwSessionId); - if (gwActivityStateWrapper != null) { - log.debug("[{}] Session with id: [{}] has gateway session with id: [{}] with overwrite activity time enabled. Updating last activity time.", name, sessionId, gwSessionId); - lastActivityTime = Math.max(gwActivityStateWrapper.getState().getLastRecordedTime(), lastActivityTime); - } - } - } - - long expirationTime = System.currentTimeMillis() - sessionInactivityTimeout; - boolean hasExpired = sessionMetaData != null && lastActivityTime < expirationTime; - if (hasExpired) { - log.debug("[{}] Session with id: [{}] has expired due to last activity time: [{}]. Marking it's activity state for removal.", name, sessionId, lastActivityTime); - transportService.deregisterSession(sessionInfo); - statesToRemove.add(sessionId); - transportService.process(sessionInfo, SESSION_EVENT_MSG_CLOSED, null); - sessionMetaData.getListener().onRemoteSessionCloseCommand(sessionId, SESSION_EXPIRED_NOTIFICATION_PROTO); - } - boolean shouldReportLeftoverEvents = sessionMetaData == null || hasExpired; - if (shouldReportLeftoverEvents && activityState.getLastReportedTime() < lastActivityTime) { - log.debug("[{}] Going to report leftover activity event for session with id: [{}].", name, sessionId); - reporter.report(sessionId, lastActivityTime, activityState, new ActivityReportCallback<>() { - @Override - public void onSuccess(UUID key, long reportedTime) { - updateLastReportedTime(key, reportedTime); - } - - @Override - public void onFailure(UUID key, Throwable t) { - log.debug("[{}] Failed to report leftover activity event for session with id: [{}].", name, sessionId); - } - }); - } - activityStateWrapper.setAlreadyBeenReported(false); - } - statesToRemove.forEach(states::remove); - } - - private void updateLastReportedTime(UUID key, long newLastReportedTime) { - states.computeIfPresent(key, (__, activityStateWrapper) -> { - var activityState = activityStateWrapper.getState(); - activityState.setLastReportedTime(Math.max(activityState.getLastReportedTime(), newLastReportedTime)); - return activityStateWrapper; - }); - } - -} diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/OnlyLastEventTransportActivityManager.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/OnlyLastEventTransportActivityManager.java deleted file mode 100644 index 977930677f..0000000000 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/OnlyLastEventTransportActivityManager.java +++ /dev/null @@ -1,149 +0,0 @@ -/** - * ThingsBoard, Inc. ("COMPANY") CONFIDENTIAL - * - * Copyright © 2016-2023 ThingsBoard, Inc. All Rights Reserved. - * - * NOTICE: All information contained herein is, and remains - * the property of ThingsBoard, Inc. and its suppliers, - * if any. The intellectual and technical concepts contained - * herein are proprietary to ThingsBoard, Inc. - * and its suppliers and may be covered by U.S. and Foreign Patents, - * patents in process, and are protected by trade secret or copyright law. - * - * Dissemination of this information or reproduction of this material is strictly forbidden - * unless prior written permission is obtained from COMPANY. - * - * Access to the source code contained herein is hereby forbidden to anyone except current COMPANY employees, - * managers or contractors who have executed Confidentiality and Non-disclosure agreements - * explicitly covering such access. - * - * The copyright notice above does not evidence any actual or intended publication - * or disclosure of this source code, which includes - * information that is confidential and/or proprietary, and is a trade secret, of COMPANY. - * ANY REPRODUCTION, MODIFICATION, DISTRIBUTION, PUBLIC PERFORMANCE, - * OR PUBLIC DISPLAY OF OR THROUGH USE OF THIS SOURCE CODE WITHOUT - * THE EXPRESS WRITTEN CONSENT OF COMPANY IS STRICTLY PROHIBITED, - * AND IN VIOLATION OF APPLICABLE LAWS AND INTERNATIONAL TREATIES. - * THE RECEIPT OR POSSESSION OF THIS SOURCE CODE AND/OR RELATED INFORMATION - * DOES NOT CONVEY OR IMPLY ANY RIGHTS TO REPRODUCE, DISCLOSE OR DISTRIBUTE ITS CONTENTS, - * OR TO MANUFACTURE, USE, OR SELL ANYTHING THAT IT MAY DESCRIBE, IN WHOLE OR IN PART. - */ -package org.thingsboard.server.common.transport.service.activity; - -import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.stereotype.Component; -import org.thingsboard.server.common.transport.TransportService; -import org.thingsboard.server.common.transport.activity.AbstractActivityManager; -import org.thingsboard.server.common.transport.activity.ActivityReportCallback; -import org.thingsboard.server.common.transport.service.SessionMetaData; -import org.thingsboard.server.common.transport.service.TransportActivityState; -import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.queue.util.TbTransportComponent; - -import java.util.HashSet; -import java.util.Map; -import java.util.Set; -import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.function.Supplier; - -import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EVENT_MSG_CLOSED; -import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EXPIRED_NOTIFICATION_PROTO; - -@Slf4j -@Component -@TbTransportComponent -@ConditionalOnProperty(prefix = "transport.activity", value = "reporting_strategy", havingValue = "last") -public class OnlyLastEventTransportActivityManager extends AbstractActivityManager { - - private final ConcurrentMap states = new ConcurrentHashMap<>(); - - @Value("${transport.sessions.inactivity_timeout}") - private long sessionInactivityTimeout; - - @Autowired - private TransportService transportService; - - @Override - protected void doOnActivity(UUID sessionId, Supplier newStateSupplier) { - long newLastRecordedTime = System.currentTimeMillis(); - states.compute(sessionId, (__, activityState) -> { - if (activityState == null) { - activityState = newStateSupplier.get(); - } - if (activityState.getLastRecordedTime() < newLastRecordedTime) { - activityState.setLastRecordedTime(newLastRecordedTime); - } - return activityState; - }); - } - - @Override - protected void doOnReportingPeriodEnd() { - Set statesToRemove = new HashSet<>(); - for (Map.Entry entry : states.entrySet()) { - var sessionId = entry.getKey(); - var activityState = entry.getValue(); - - SessionMetaData sessionMetaData = transportService.getSession(sessionId); - if (sessionMetaData != null) { - activityState.setSessionInfoProto(sessionMetaData.getSessionInfo()); - } else { - log.debug("[{}] Session with id: [{}] is not present. Marking it's activity state for removal.", name, sessionId); - statesToRemove.add(sessionId); - } - - long lastActivityTime = activityState.getLastRecordedTime(); - TransportProtos.SessionInfoProto sessionInfo = activityState.getSessionInfoProto(); - - if (sessionInfo.getGwSessionIdMSB() != 0 && sessionInfo.getGwSessionIdLSB() != 0) { - var gwSessionId = new UUID(sessionInfo.getGwSessionIdMSB(), sessionInfo.getGwSessionIdLSB()); - SessionMetaData gwSessionMetaData = transportService.getSession(gwSessionId); - if (gwSessionMetaData != null && gwSessionMetaData.isOverwriteActivityTime()) { - TransportActivityState gwActivityState = states.get(gwSessionId); - if (gwActivityState != null) { - log.debug("[{}] Session with id: [{}] has gateway session with id: [{}] with overwrite activity time enabled. Updating last activity time.", name, sessionId, gwSessionId); - lastActivityTime = Math.max(gwActivityState.getLastRecordedTime(), lastActivityTime); - } - } - } - - long expirationTime = System.currentTimeMillis() - sessionInactivityTimeout; - boolean hasExpired = sessionMetaData != null && lastActivityTime < expirationTime; - if (hasExpired) { - log.debug("[{}] Session with id: [{}] has expired due to last activity time: [{}]. Marking it's activity state for removal.", name, sessionId, lastActivityTime); - transportService.deregisterSession(sessionInfo); - statesToRemove.add(sessionId); - transportService.process(sessionInfo, SESSION_EVENT_MSG_CLOSED, null); - sessionMetaData.getListener().onRemoteSessionCloseCommand(sessionId, SESSION_EXPIRED_NOTIFICATION_PROTO); - } - if (activityState.getLastReportedTime() < lastActivityTime) { - log.debug("[{}] Going to report last activity event for session with id: [{}].", name, sessionId); - reporter.report(sessionId, lastActivityTime, activityState, new ActivityReportCallback<>() { - @Override - public void onSuccess(UUID key, long reportedTime) { - updateLastReportedTime(key, reportedTime); - } - - @Override - public void onFailure(UUID key, Throwable t) { - log.debug("[{}] Failed to report last activity event for session with id: [{}].", name, sessionId); - } - }); - } - } - statesToRemove.forEach(states::remove); - } - - private void updateLastReportedTime(UUID key, long newLastReportedTime) { - states.computeIfPresent(key, (__, activityState) -> { - activityState.setLastReportedTime(Math.max(activityState.getLastReportedTime(), newLastReportedTime)); - return activityState; - }); - } - -}