[WIP] Reimplemented activity management

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

@@ -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<Key, State extends ActivityState> {
+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<Key> callback);
+@Value
+public class IntegrationActivityKey {
+
+    @NonNull TenantId tenantId;
+    @NonNull DeviceId deviceId;
 
 }

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/AbstractActivityManager.java

@@ -1,22 +1,22 @@
 /**
  * ThingsBoard, Inc. ("COMPANY") CONFIDENTIAL
- * <p>
+ *
  * Copyright © 2016-2023 ThingsBoard, Inc. All Rights Reserved.
- * <p>
+ *
  * 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.
- * <p>
+ *
  * Dissemination of this information or reproduction of this material is strictly forbidden
  * unless prior written permission is obtained from COMPANY.
- * <p>
+ *
  * 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.
- * <p>
+ *
  * 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<Key, Metadata> implements ActivityManager<Key, ActivityState<Metadata>> {
+public abstract class AbstractActivityManager<Key, Metadata> implements ActivityManager<Key> {
 
     private final ConcurrentMap<Key, ActivityState<Metadata>> states = new ConcurrentHashMap<>();
-    private String name;
+    protected String name;
     private long reportingPeriodMillis;
-    private ActivityStateReporter<Key, ActivityState<Metadata>> reporter;
     private ScheduledExecutorService scheduler;
     private boolean initialized;
 
     @Override
-    public synchronized void init(String name, long reportingPeriodMillis, ActivityStateReporter<Key, ActivityState<Metadata>> 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<Metadata> createNewState(Key key);
+
+    protected abstract ActivityStrategy getStrategy();
+
+    protected abstract ActivityState<Metadata> updateState(Key key, ActivityState<Metadata> state);
+
+    protected abstract boolean hasExpired(Key key, ActivityState<Metadata> state);
+
+    protected abstract void onStateExpire(Key key, Metadata metadata);
+
+    protected abstract void reportActivity(Key key, Metadata metadata, long timeToReport, ActivityReportCallback<Key> callback);
+
     @Override
-    public void onActivity(Key key, Supplier<ActivityState<Metadata>> 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<Key, Metadata> 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<ActivityState<Metadata>> newStateSupplier) {
-
-    }
-
-    protected abstract ActivityStrategy getStrategy();
 
-    private void doOnActivity(Key key, Supplier<ActivityState<Metadata>> 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<Key, Metadata> implements Activity
         }
     }
 
+    protected long getLastRecordedTime(Key key) {
+        ActivityState<Metadata> state = states.get(key);
+        return state == null ? 0L : state.getLastRecordedTime();
+    }
+
     private void onReportingPeriodEnd() {
         log.debug("[{}] Going to end reporting period.", name);
         for (Map.Entry<Key, ActivityState<Metadata>> 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<Key, Metadata> 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;
         });
     }
 

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityManager.java

@@ -1,22 +1,22 @@
 /**
  * ThingsBoard, Inc. ("COMPANY") CONFIDENTIAL
- * <p>
+ *
  * Copyright © 2016-2023 ThingsBoard, Inc. All Rights Reserved.
- * <p>
+ *
  * 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.
- * <p>
+ *
  * Dissemination of this information or reproduction of this material is strictly forbidden
  * unless prior written permission is obtained from COMPANY.
- * <p>
+ *
  * 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.
- * <p>
+ *
  * 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<Key> {
 
-public interface ActivityManager<Key, Metadata> {
+    void init(String name, long reportingPeriodMillis);
 
-    void init(String name, long reportingPeriodMillis, ActivityStateReporter<Key, ActivityState<Metadata>> reporter);
-
-    void onActivity(Key key, Supplier<ActivityState<Metadata>> newStateSupplier);
+    void onActivity(Key key);
 
     void destroy();
 

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();
 
 }

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 + ".");
+        }
+    }
+
+}

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/strategy/ActivityStrategyType.java

@@ -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;
-    }
-
-}

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;
     }
 
 }

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;
     }
 
 }

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;
     }
 

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;
     }
 
 }

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<UUID, TransportProtos.SessionInfoProto> 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<UUID, TransportActivityState> activityManager;
 
     protected TbQueueRequestTemplate<TbProtoQueueMsg<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> transportApiRequestTemplate;
     protected TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> ruleEngineMsgProducer;
@@ -235,13 +238,9 @@ public class DefaultTransportService implements TransportService {
         this.entityLimitsCache = entityLimitsCache;
     }
 
-    @Autowired
-    public void setActivityManager(ActivityManager<UUID, TransportActivityState> 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<UUID, TransportActivityState> 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<TransportProtos.SessionInfoProto> createNewState(UUID sessionId) {
+        SessionMetaData session = sessions.get(sessionId);
+        if (session == null) {
+            return null;
+        }
+        ActivityState<TransportProtos.SessionInfoProto> state = new ActivityState<>();
+        state.setMetadata(session.getSessionInfo());
+        return state;
+    }
+
+    @Override
+    protected ActivityStrategy getStrategy() {
+        return ActivityStrategyFactory.createStrategy(reportingStrategyName);
+    }
+
+    @Override
+    protected ActivityState<TransportProtos.SessionInfoProto> updateState(UUID sessionId, ActivityState<TransportProtos.SessionInfoProto> 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<TransportProtos.SessionInfoProto> 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<UUID> 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);
+            }
         });
     }
 

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/FirstAndLastEventTransportActivityManager.java

@@ -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<UUID, TransportActivityState> {
-
-    private final ConcurrentMap<UUID, ActivityStateWrapper> 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<TransportActivityState> newStateSupplier) {
-        long newLastRecordedTime = System.currentTimeMillis();
-        SettableFuture<Pair<UUID, Long>> 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<UUID, Long> 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<UUID> statesToRemove = new HashSet<>();
-        for (Map.Entry<UUID, ActivityStateWrapper> 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;
-        });
-    }
-
-}

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/OnlyFirstEventTransportActivityManager.java

@@ -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<UUID, TransportActivityState> {
-
-    private final ConcurrentMap<UUID, ActivityStateWrapper> 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<TransportActivityState> newStateSupplier) {
-        long newLastRecordedTime = System.currentTimeMillis();
-        SettableFuture<Pair<UUID, Long>> 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<UUID, Long> 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<UUID> statesToRemove = new HashSet<>();
-        for (Map.Entry<UUID, ActivityStateWrapper> 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;
-        });
-    }
-
-}

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/OnlyLastEventTransportActivityManager.java

@@ -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<UUID, TransportActivityState> {
-
-    private final ConcurrentMap<UUID, TransportActivityState> states = new ConcurrentHashMap<>();
-
-    @Value("${transport.sessions.inactivity_timeout}")
-    private long sessionInactivityTimeout;
-
-    @Autowired
-    private TransportService transportService;
-
-    @Override
-    protected void doOnActivity(UUID sessionId, Supplier<TransportActivityState> 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<UUID> statesToRemove = new HashSet<>();
-        for (Map.Entry<UUID, TransportActivityState> 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;
-        });
-    }
-
-}