diff --git a/application/src/main/java/org/thingsboard/server/service/integration/activity/FirstAndLastIntegrationActivityManager.java b/application/src/main/java/org/thingsboard/server/service/integration/activity/FirstAndLastIntegrationActivityManager.java index 12be2101c4..4a38af1a4d 100644 --- a/application/src/main/java/org/thingsboard/server/service/integration/activity/FirstAndLastIntegrationActivityManager.java +++ b/application/src/main/java/org/thingsboard/server/service/integration/activity/FirstAndLastIntegrationActivityManager.java @@ -37,80 +37,54 @@ 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.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.data.util.Pair; -import org.thingsboard.common.util.ThingsBoardThreadFactory; -import org.thingsboard.server.common.transport.activity.ActivityManager; +import org.springframework.stereotype.Component; +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.ActivityStateReportCallback; -import org.thingsboard.server.common.transport.activity.ActivityStateReporter; -import java.util.HashSet; import java.util.Map; -import java.util.Objects; -import java.util.Random; -import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; import java.util.function.Supplier; @Slf4j -public class FirstAndLastIntegrationActivityManager implements ActivityManager { +@Component +@ConditionalOnProperty(prefix = "integrations.activity", value = "reporting_strategy", havingValue = "first-and-last") +public class FirstAndLastIntegrationActivityManager extends AbstractActivityManager { - private final ConcurrentMap activityStates = new ConcurrentHashMap<>(); + private final ConcurrentMap states = new ConcurrentHashMap<>(); @Data private static class ActivityStateWrapper { - volatile ActivityState activityState; + volatile ActivityState state; volatile boolean alreadyBeenReported; } - private ScheduledExecutorService scheduler; - private final ActivityStateReporter reporter; - private final long reportingPeriodDurationMillis; - private final String name; - private boolean initialized; - - public FirstAndLastIntegrationActivityManager(ActivityStateReporter reporter, long reportingPeriodDurationMillis, String name) { - this.reporter = Objects.requireNonNull(reporter, "Failed to create activity manager: provided reporter is null."); - this.reportingPeriodDurationMillis = reportingPeriodDurationMillis; // TODO: add min/max duration validation - this.name = name == null ? "activity-state-manager" : name; - } - - @Override - public synchronized void init() { - if (!initialized) { - scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName(name)); - scheduler.scheduleAtFixedRate(this::onReportingPeriodEnd, new Random().nextInt((int) reportingPeriodDurationMillis), reportingPeriodDurationMillis, TimeUnit.MILLISECONDS); - initialized = true; - } - } - @Override - public void onActivity(IntegrationActivityKey activityKey, Supplier newStateSupplier) { + protected void doOnActivity(IntegrationActivityKey activityKey, Supplier newStateSupplier) { long newLastRecordedTime = System.currentTimeMillis(); SettableFuture> reportCompletedFuture = SettableFuture.create(); - activityStates.compute(activityKey, (key, activityStateWrapper) -> { + states.compute(activityKey, (key, activityStateWrapper) -> { if (activityStateWrapper == null) { ActivityState activityState = newStateSupplier.get(); activityState.setLastRecordedTime(newLastRecordedTime); activityState.setLastReportedTime(0L); activityStateWrapper = new ActivityStateWrapper(); - activityStateWrapper.setActivityState(activityState); + activityStateWrapper.setState(activityState); activityStateWrapper.setAlreadyBeenReported(false); } else { - activityStateWrapper.getActivityState().setLastRecordedTime(newLastRecordedTime); + activityStateWrapper.getState().setLastRecordedTime(newLastRecordedTime); } if (activityStateWrapper.isAlreadyBeenReported()) { return activityStateWrapper; } - var activityState = activityStateWrapper.getActivityState(); + var activityState = activityStateWrapper.getState(); if (activityState.getLastReportedTime() < activityState.getLastRecordedTime()) { - reporter.report(key, activityState, new ActivityStateReportCallback<>() { + reporter.report(key, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() { @Override public void onSuccess(IntegrationActivityKey key, long reportedTime) { reportCompletedFuture.set(Pair.of(key, reportedTime)); @@ -133,24 +107,23 @@ public class FirstAndLastIntegrationActivityManager implements ActivityManager statesToRemove = new HashSet<>(); - Set> entries = activityStates.entrySet(); - for (Map.Entry entry : entries) { + @Override + protected void onReportingPeriodEnd() { + long expirationTime = System.currentTimeMillis() - reportingPeriodMillis; + for (Map.Entry entry : states.entrySet()) { var activityKey = entry.getKey(); var activityStateWrapper = entry.getValue(); - var activityState = activityStateWrapper.getActivityState(); + var activityState = activityStateWrapper.getState(); if (activityState.getLastRecordedTime() < expirationTime) { - statesToRemove.add(activityKey); + states.remove(activityKey); } if (activityState.getLastReportedTime() < activityState.getLastRecordedTime()) { - reporter.report(activityKey, activityState, new ActivityStateReportCallback<>() { + reporter.report(activityKey, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() { @Override public void onSuccess(IntegrationActivityKey key, long reportedTime) { updateLastReportedTime(key, reportedTime); @@ -158,39 +131,21 @@ public class FirstAndLastIntegrationActivityManager implements ActivityManager { - var activityState = activityStateWrapper.getActivityState(); + states.computeIfPresent(key, (__, activityStateWrapper) -> { + var activityState = activityStateWrapper.getState(); activityState.setLastReportedTime(Math.max(activityState.getLastReportedTime(), newLastReportedTime)); return activityStateWrapper; }); } - @Override - public synchronized void destroy() { - if (initialized) { - initialized = false; - if (scheduler != null) { - scheduler.shutdown(); - try { - if (scheduler.awaitTermination(10L, TimeUnit.SECONDS)) { - scheduler.shutdownNow(); - } - } catch (InterruptedException e) { - scheduler.shutdownNow(); - Thread.currentThread().interrupt(); - } - } - } - } - } diff --git a/application/src/main/java/org/thingsboard/server/service/integration/activity/FirstOnlyIntegrationActivityManager.java b/application/src/main/java/org/thingsboard/server/service/integration/activity/FirstOnlyIntegrationActivityManager.java index 7a0bccd48c..b9965098df 100644 --- a/application/src/main/java/org/thingsboard/server/service/integration/activity/FirstOnlyIntegrationActivityManager.java +++ b/application/src/main/java/org/thingsboard/server/service/integration/activity/FirstOnlyIntegrationActivityManager.java @@ -37,80 +37,54 @@ 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.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.data.util.Pair; -import org.thingsboard.common.util.ThingsBoardThreadFactory; -import org.thingsboard.server.common.transport.activity.ActivityManager; +import org.springframework.stereotype.Component; +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.ActivityStateReportCallback; -import org.thingsboard.server.common.transport.activity.ActivityStateReporter; -import java.util.HashMap; import java.util.Map; -import java.util.Objects; -import java.util.Random; -import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; import java.util.function.Supplier; @Slf4j -public class FirstOnlyIntegrationActivityManager implements ActivityManager { +@Component +@ConditionalOnProperty(prefix = "integrations.activity", value = "reporting_strategy", havingValue = "first") +public class FirstOnlyIntegrationActivityManager extends AbstractActivityManager { - private final ConcurrentMap activityStates = new ConcurrentHashMap<>(); + private final ConcurrentMap states = new ConcurrentHashMap<>(); @Data private static class ActivityStateWrapper { - volatile ActivityState activityState; + volatile ActivityState state; volatile boolean alreadyBeenReported; } - private ScheduledExecutorService scheduler; - private final ActivityStateReporter reporter; - private final long reportingPeriodDurationMillis; - private final String name; - private boolean initialized; - - public FirstOnlyIntegrationActivityManager(ActivityStateReporter reporter, long reportingPeriodDurationMillis, String name) { - this.reporter = Objects.requireNonNull(reporter, "Failed to create activity manager: provided reporter is null."); - this.reportingPeriodDurationMillis = reportingPeriodDurationMillis; // TODO: add min/max duration validation - this.name = name == null ? "activity-state-manager" : name; - } - - @Override - public synchronized void init() { - if (!initialized) { - scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName(name)); - scheduler.scheduleAtFixedRate(this::onReportingPeriodEnd, new Random().nextInt((int) reportingPeriodDurationMillis), reportingPeriodDurationMillis, TimeUnit.MILLISECONDS); - initialized = true; - } - } - @Override - public void onActivity(IntegrationActivityKey activityKey, Supplier newStateSupplier) { + protected void doOnActivity(IntegrationActivityKey activityKey, Supplier newStateSupplier) { long newLastRecordedTime = System.currentTimeMillis(); SettableFuture> reportCompletedFuture = SettableFuture.create(); - activityStates.compute(activityKey, (key, activityStateWrapper) -> { + states.compute(activityKey, (key, activityStateWrapper) -> { if (activityStateWrapper == null) { ActivityState activityState = newStateSupplier.get(); activityState.setLastRecordedTime(newLastRecordedTime); activityState.setLastReportedTime(0L); activityStateWrapper = new ActivityStateWrapper(); - activityStateWrapper.setActivityState(activityState); + activityStateWrapper.setState(activityState); activityStateWrapper.setAlreadyBeenReported(false); } else { - activityStateWrapper.getActivityState().setLastRecordedTime(newLastRecordedTime); + activityStateWrapper.getState().setLastRecordedTime(newLastRecordedTime); } if (activityStateWrapper.isAlreadyBeenReported()) { return activityStateWrapper; } - var activityState = activityStateWrapper.getActivityState(); + var activityState = activityStateWrapper.getState(); if (activityState.getLastReportedTime() < activityState.getLastRecordedTime()) { - reporter.report(key, activityState, new ActivityStateReportCallback<>() { + reporter.report(key, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() { @Override public void onSuccess(IntegrationActivityKey key, long reportedTime) { reportCompletedFuture.set(Pair.of(key, reportedTime)); @@ -133,59 +107,39 @@ public class FirstOnlyIntegrationActivityManager implements ActivityManager { - var activityState = activityStateWrapper.getActivityState(); + states.computeIfPresent(key, (__, activityStateWrapper) -> { + var activityState = activityStateWrapper.getState(); activityState.setLastReportedTime(Math.max(activityState.getLastReportedTime(), newLastReportedTime)); return activityStateWrapper; }); } - private void onReportingPeriodEnd() { - Map statesToRemoveAndReport = new HashMap<>(); - Set> entries = activityStates.entrySet(); - for (Map.Entry entry : entries) { + @Override + protected void onReportingPeriodEnd() { + for (Map.Entry entry : states.entrySet()) { var activityKey = entry.getKey(); var activityStateWrapper = entry.getValue(); - var activityState = activityStateWrapper.getActivityState(); + var activityState = activityStateWrapper.getState(); if (!activityStateWrapper.isAlreadyBeenReported() && activityState.getLastReportedTime() < activityState.getLastRecordedTime()) { - statesToRemoveAndReport.put(activityKey, activityState); - } - activityStateWrapper.setAlreadyBeenReported(false); - } - statesToRemoveAndReport.forEach((key, state) -> reporter.report(key, state, new ActivityStateReportCallback<>() { - @Override - public void onSuccess(IntegrationActivityKey key, long reportedTime) { - activityStates.remove(key); - } - - @Override - public void onFailure(IntegrationActivityKey key, Throwable t) { - // TODO: log - } - })); - } + reporter.report(activityKey, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() { + @Override + public void onSuccess(IntegrationActivityKey key, long reportedTime) { + states.remove(key); + } - @Override - public synchronized void destroy() { - if (initialized) { - initialized = false; - if (scheduler != null) { - scheduler.shutdown(); - try { - if (scheduler.awaitTermination(10L, TimeUnit.SECONDS)) { - scheduler.shutdownNow(); + @Override + public void onFailure(IntegrationActivityKey key, Throwable t) { + log.debug("[{}] Failed to report last activity event in a period for device with id: [{}]", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId()); } - } catch (InterruptedException e) { - scheduler.shutdownNow(); - Thread.currentThread().interrupt(); - } + }); } + activityStateWrapper.setAlreadyBeenReported(false); } } diff --git a/application/src/main/java/org/thingsboard/server/service/integration/activity/LastOnlyIntegrationActivityManager.java b/application/src/main/java/org/thingsboard/server/service/integration/activity/LastOnlyIntegrationActivityManager.java index 0dd7aa6fc1..bffdc49054 100644 --- a/application/src/main/java/org/thingsboard/server/service/integration/activity/LastOnlyIntegrationActivityManager.java +++ b/application/src/main/java/org/thingsboard/server/service/integration/activity/LastOnlyIntegrationActivityManager.java @@ -31,54 +31,28 @@ package org.thingsboard.server.service.integration.activity; import lombok.extern.slf4j.Slf4j; -import org.thingsboard.common.util.ThingsBoardThreadFactory; -import org.thingsboard.server.common.transport.activity.ActivityManager; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.stereotype.Component; +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.ActivityStateReportCallback; -import org.thingsboard.server.common.transport.activity.ActivityStateReporter; -import java.util.HashSet; import java.util.Map; -import java.util.Objects; -import java.util.Random; -import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; import java.util.function.Supplier; @Slf4j -public class LastOnlyIntegrationActivityManager implements ActivityManager { +@Component +@ConditionalOnProperty(prefix = "integrations.activity", value = "reporting_strategy", havingValue = "last") +public class LastOnlyIntegrationActivityManager extends AbstractActivityManager { - private final ConcurrentMap activityStates = new ConcurrentHashMap<>(); - - private ScheduledExecutorService scheduler; - private final ActivityStateReporter reporter; - private final long reportingPeriodDurationMillis; - private final String name; - private boolean initialized; - - public LastOnlyIntegrationActivityManager(ActivityStateReporter reporter, long reportingPeriodDurationMillis, String name) { - this.reporter = Objects.requireNonNull(reporter, "Failed to create activity manager: provided reporter is null."); - this.reportingPeriodDurationMillis = reportingPeriodDurationMillis; // TODO: add min/max duration validation - this.name = name == null ? "activity-state-manager" : name; - } - - @Override - public synchronized void init() { - if (!initialized) { - scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName(name)); - scheduler.scheduleAtFixedRate(this::onReportingPeriodEnd, new Random().nextInt((int) reportingPeriodDurationMillis), reportingPeriodDurationMillis, TimeUnit.MILLISECONDS); - initialized = true; - } - } + private final ConcurrentMap states = new ConcurrentHashMap<>(); @Override - public void onActivity(IntegrationActivityKey activityKey, Supplier newStateSupplier) { + protected void doOnActivity(IntegrationActivityKey activityKey, Supplier newStateSupplier) { long newLastRecordedTime = System.currentTimeMillis(); - activityStates.compute(activityKey, (key, activityState) -> { + states.compute(activityKey, (__, activityState) -> { if (activityState == null) { activityState = newStateSupplier.get(); activityState.setLastRecordedTime(newLastRecordedTime); @@ -90,18 +64,17 @@ public class LastOnlyIntegrationActivityManager implements ActivityManager statesToRemove = new HashSet<>(); - Set> entries = activityStates.entrySet(); - for (Map.Entry entry : entries) { + @Override + public void onReportingPeriodEnd() { + long expirationTime = System.currentTimeMillis() - reportingPeriodMillis; + for (Map.Entry entry : states.entrySet()) { var activityKey = entry.getKey(); var activityState = entry.getValue(); if (activityState.getLastRecordedTime() < expirationTime) { - statesToRemove.add(activityKey); + states.remove(activityKey); } if (activityState.getLastReportedTime() < activityState.getLastRecordedTime()) { - reporter.report(activityKey, activityState, new ActivityStateReportCallback<>() { + reporter.report(activityKey, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() { @Override public void onSuccess(IntegrationActivityKey key, long reportedTime) { updateLastReportedTime(key, reportedTime); @@ -109,37 +82,18 @@ public class LastOnlyIntegrationActivityManager implements ActivityManager { + states.computeIfPresent(key, (__, activityState) -> { activityState.setLastReportedTime(Math.max(activityState.getLastReportedTime(), newLastReportedTime)); return activityState; }); } - @Override - public synchronized void destroy() { - if (initialized) { - initialized = false; - if (scheduler != null) { - scheduler.shutdown(); - try { - if (scheduler.awaitTermination(10L, TimeUnit.SECONDS)) { - scheduler.shutdownNow(); - } - } catch (InterruptedException e) { - scheduler.shutdownNow(); - Thread.currentThread().interrupt(); - } - } - } - } - } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index e746dc9dba..73f0622cd1 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -866,6 +866,13 @@ transport: inactivity_timeout: "${TB_TRANSPORT_SESSIONS_INACTIVITY_TIMEOUT:300000}" # Interval of periodic check for expired sessions and report of the changes to session last activity time report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:3000}" + activity: + # This property specifies the strategy for reporting activity events within each reporting period. + # The accepted values are 'first', 'last', and 'first and last'. + # - 'first': Only the first activity event in each reporting period is reported. + # - 'last': Only the last activity event in the reporting period is reported. + # - 'first-and-last': Both the first and last activity events in the reporting period are reported. + reporting_strategy: "${TB_TRANSPORT_ACTIVITY_REPORTING_STRATEGY:first-and-last}" json: # Cast String data types to Numeric if possible when processing Telemetry/Attributes JSON type_cast_enabled: "${JSON_TYPE_CAST_ENABLED:true}" @@ -1695,4 +1702,4 @@ management: web: exposure: # Expose metrics endpoint (use value 'prometheus' to enable prometheus metrics). - include: '${METRICS_ENDPOINTS_EXPOSE:info}' \ No newline at end of file + include: '${METRICS_ENDPOINTS_EXPOSE:info}' diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java index f51b399a5b..067527efec 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java @@ -63,6 +63,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509Ce import org.thingsboard.server.gen.transport.TransportProtos.ValidateOrCreateDeviceX509CertRequestMsg; import java.util.List; +import java.util.UUID; import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicInteger; @@ -161,5 +162,7 @@ public interface TransportService { boolean hasSession(SessionInfoProto sessionInfo); + SessionMetaData getSession(UUID sessionId); + void createGaugeStats(String openConnections, AtomicInteger connectionsCounter); } 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 new file mode 100644 index 0000000000..0c7e00816a --- /dev/null +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/AbstractActivityManager.java @@ -0,0 +1,114 @@ +/** + * 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; + +import org.thingsboard.common.util.ThingsBoardThreadFactory; + +import java.util.Objects; +import java.util.Random; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; + +public abstract class AbstractActivityManager implements ActivityManager { + + private String name; + protected long reportingPeriodMillis; + protected ActivityStateReporter reporter; + private ScheduledExecutorService scheduler; + protected boolean initialized; + + @Override + public synchronized void init() { + Objects.requireNonNull(reporter, "Failed to initialize activity manager: provided activity reporter is null."); + if (reportingPeriodMillis < 1000) { + throw new IllegalArgumentException("Failed to initialize activity manager: provided reporting period duration is less that 1 second."); + } + if (!initialized) { + scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName(name == null ? "activity-manager" : name)); + scheduler.scheduleAtFixedRate(this::onReportingPeriodEnd, new Random().nextInt((int) reportingPeriodMillis), reportingPeriodMillis, TimeUnit.MILLISECONDS); + initialized = true; + } + } + + @Override + public void setName(String name) { + this.name = name; + } + + @Override + public void setReportingPeriod(long reportingPeriodMillis) { + this.reportingPeriodMillis = reportingPeriodMillis; + } + + @Override + public void setActivityReporter(ActivityStateReporter activityReporter) { + reporter = activityReporter; + } + + @Override + public void onActivity(Key key, Supplier newStateSupplier) { + if (!initialized) { + throw new IllegalStateException(name + " is not initialized."); + } + if (key == null) { + throw new IllegalArgumentException("Activity key can't be null."); + } + if (newStateSupplier == null) { + throw new IllegalArgumentException("New activity state supplier can't be null."); + } + doOnActivity(key, newStateSupplier); + } + + protected abstract void doOnActivity(Key key, Supplier newStateSupplier); + + protected abstract void onReportingPeriodEnd(); + + @Override + public synchronized void destroy() { + if (initialized) { + initialized = false; + if (scheduler != null) { + scheduler.shutdown(); + try { + if (scheduler.awaitTermination(10L, TimeUnit.SECONDS)) { + scheduler.shutdownNow(); + } + } catch (InterruptedException e) { + scheduler.shutdownNow(); + Thread.currentThread().interrupt(); + } + } + } + } + +} 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 d93756a922..ba09ccf773 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityManager.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityManager.java @@ -34,6 +34,12 @@ import java.util.function.Supplier; public interface ActivityManager { + void setName(String name); + + void setReportingPeriod(long reportingPeriodMillis); + + void setActivityReporter(ActivityStateReporter activityReporter); + void init(); void onActivity(Key key, Supplier newStateSupplier); diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityStateReportCallback.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityReportCallback.java similarity index 97% rename from common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityStateReportCallback.java rename to common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityReportCallback.java index 4ff752d61d..9d009d2707 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityStateReportCallback.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityReportCallback.java @@ -30,7 +30,7 @@ */ package org.thingsboard.server.common.transport.activity; -public interface ActivityStateReportCallback { +public interface ActivityReportCallback { void onSuccess(Key key, long reportedTime); diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityStateReporter.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityStateReporter.java index 620646f894..1092c569d2 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityStateReporter.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityStateReporter.java @@ -30,12 +30,9 @@ */ package org.thingsboard.server.common.transport.activity; -import java.util.Map; - +@FunctionalInterface public interface ActivityStateReporter { - void report(Key key, State activityState, ActivityStateReportCallback reportCallback); - - void report(Map activityStates, ActivityStateReportCallback reportCallback); + void report(Key key, long timeToReport, State activityState, ActivityReportCallback callback); } 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 31ab2df09e..6453cbd24c 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 @@ -73,7 +73,6 @@ 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.ActivityStateReportCallback; import org.thingsboard.server.common.transport.activity.ActivityStateReporter; import org.thingsboard.server.common.transport.auth.GetOrCreateDeviceFromGatewayResponse; import org.thingsboard.server.common.transport.auth.TransportDeviceInfo; @@ -81,7 +80,6 @@ import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsRes import org.thingsboard.server.common.transport.limits.EntityLimitKey; import org.thingsboard.server.common.transport.limits.EntityLimitsCache; import org.thingsboard.server.common.transport.limits.TransportRateLimitService; -import org.thingsboard.server.common.transport.service.activity.LastOnlyTransportActivityManager; import org.thingsboard.server.common.transport.util.JsonUtils; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ProvisionDeviceRequestMsg; @@ -187,8 +185,8 @@ public class DefaultTransportService implements TransportService { private final ApplicationEventPublisher eventPublisher; private final TransportResourceCache transportResourceCache; private final NotificationRuleProcessor notificationRuleProcessor; - private final EntityLimitsCache entityLimitsCache; + private ActivityManager activityManager; protected TbQueueRequestTemplate, TbProtoQueueMsg> transportApiRequestTemplate; protected TbQueueProducer> ruleEngineMsgProducer; @@ -203,7 +201,6 @@ public class DefaultTransportService implements TransportService { private ExecutorService mainConsumerExecutor; public final ConcurrentMap sessions = new ConcurrentHashMap<>(); - private ActivityManager activityManager; private final Map toServerRpcPendingMap = new ConcurrentHashMap<>(); private volatile boolean stopped = false; @@ -237,6 +234,11 @@ public class DefaultTransportService implements TransportService { this.entityLimitsCache = entityLimitsCache; } + @Autowired + public void setActivityManager(ActivityManager activityManager) { + this.activityManager = activityManager; + } + @PostConstruct public void init() { this.ruleEngineProducerStats = statsFactory.createMessagesStats(StatsType.RULE_ENGINE.getName() + ".producer"); @@ -253,104 +255,32 @@ public class DefaultTransportService implements TransportService { transportNotificationsConsumer.subscribe(Collections.singleton(tpi)); transportApiRequestTemplate.init(); mainConsumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("transport-consumer")); - activityManager = new LastOnlyTransportActivityManager(reporter, sessionReportTimeout, "transport-activity-state-manager"); + + activityManager.setName("transport-activity-manager"); + activityManager.setReportingPeriod(sessionReportTimeout); + activityManager.setActivityReporter(activityReporter); activityManager.init(); } - // TODO: why sessions and session activities are managed in a separate maps? maybe it is better to manage in a single map - // (eg. we can check if we had sent last activity event when deregistering session) - - // TODO: currently if activity event is received between precise session expiration time and reportLastEventAndStartNewPeriod() call - // we will "resurrect" this session meaning that it will be considered alive but in fact it did not perform any activity for more than timeout allows - - // TODO: I optimistically set alreadyBeenReported status to true to avoid reporting several activity events if they arrive in rapid succession - // this can cause lost activity updates if first event that got reported failed to enqueue in Kafka and next reporting period is already started - // (maybe having a queue of "first" events will help with this, queue will be cleared once event was successfully queued or later time was reported by event in next period) - // setting alreadyBeenReported status in callbacks ensures that events are reported but can cause several "first" events to be reported - private final ActivityStateReporter reporter = new ActivityStateReporter<>() { - @Override - public void report(UUID sessionId, TransportActivityState activityState, ActivityStateReportCallback reportCallback) { - SessionMetaData sessionMetaData = sessions.get(sessionId); - reportActivityStateToCore(activityState.getSessionInfoProto(), sessionMetaData, activityState.getLastRecordedTime(), new TransportServiceCallback<>() { - @Override - public void onSuccess(Void msg) { - reportCallback.onSuccess(sessionId, activityState.getLastRecordedTime()); - - } - - @Override - public void onError(Throwable e) { - reportCallback.onFailure(sessionId, e); - - } - }); - } - - @Override - public void report(Map activityStates, ActivityStateReportCallback reportCallback) { - long expirationTime = System.currentTimeMillis() - sessionInactivityTimeout; - for (Map.Entry entry : activityStates.entrySet()) { - var sessionId = entry.getKey(); - var activityState = entry.getValue(); - - SessionMetaData sessionMetaData = sessions.get(sessionId); - if (sessionMetaData != null) { - activityState.setSessionInfoProto(sessionMetaData.getSessionInfo()); - } else { - log.info("Removing activity state due to session deregistration."); - activityStates.remove(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 = sessions.get(gwSessionId); - if (gwSessionMetaData != null && gwSessionMetaData.isOverwriteActivityTime()) { - TransportActivityState gwActivityState = activityStates.get(gwSessionId); - if (gwActivityState != null) { - lastActivityTime = Math.max(gwActivityState.getLastRecordedTime(), lastActivityTime); - } - } - } - - if (sessionMetaData != null && lastActivityTime < expirationTime) { - if (log.isDebugEnabled()) { - log.debug("[{}] Session has expired due to last activity time: {}!", sessionId, lastActivityTime); - } - sessions.remove(sessionId); - log.info("Removing activity state due to session expiration."); - activityStates.remove(sessionId); - process(sessionInfo, SESSION_EVENT_MSG_CLOSED, null); - sessionMetaData.getListener().onRemoteSessionCloseCommand(sessionId, SESSION_EXPIRED_NOTIFICATION_PROTO); - } else if (activityState.getLastReportedTime() < lastActivityTime) { - long finalLastActivityTime = lastActivityTime; - reportActivityStateToCore(sessionInfo, sessionMetaData, lastActivityTime, new TransportServiceCallback<>() { - @Override - public void onSuccess(Void msgAcknowledged) { - reportCallback.onSuccess(sessionId, finalLastActivityTime); - } + private final ActivityStateReporter activityReporter = (sessionId, timeToReport, state, reportCallback) -> { + 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); - } - }); - } } - } - private void reportActivityStateToCore( - TransportProtos.SessionInfoProto sessionInfo, SessionMetaData sessionMetaData, long lastActivityTime, TransportServiceCallback callback - ) { - TransportProtos.SubscriptionInfoProto subscriptionInfo = TransportProtos.SubscriptionInfoProto.newBuilder() - .setAttributeSubscription(sessionMetaData != null && sessionMetaData.isSubscribedToAttributes()) - .setRpcSubscription(sessionMetaData != null && sessionMetaData.isSubscribedToRPC()) - .setLastActivityTime(lastActivityTime) - .build(); - process(sessionInfo, subscriptionInfo, callback); - } + @Override + public void onError(Throwable e) { + reportCallback.onFailure(sessionId, e); + } + }); }; @AfterStartUp(order = AfterStartUp.TRANSPORT_SERVICE) @@ -658,7 +588,6 @@ public class DefaultTransportService implements TransportService { @Override public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.SessionEventMsg msg, TransportServiceCallback callback) { - log.info("Received session event: [{}]", msg.getEvent()); if (checkLimits(sessionInfo, msg, callback)) { reportActivityInternal(sessionInfo); sendToDeviceActor(sessionInfo, TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo) @@ -949,7 +878,6 @@ public class DefaultTransportService implements TransportService { log.debug("Stopping scheduler to avoid resending response if request has been ack."); currentSession.getScheduledFuture().cancel(false); } - log.info("Deregistering session."); sessions.remove(toSessionId(sessionInfo)); } @@ -1370,6 +1298,11 @@ public class DefaultTransportService implements TransportService { return sessions.containsKey(toSessionId(sessionInfo)); } + @Override + public SessionMetaData getSession(UUID sessionId) { + return sessions.get(sessionId); + } + @Override public void createGaugeStats(String statsName, AtomicInteger number) { statsFactory.createGauge(StatsType.TRANSPORT + "." + statsName, number); diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/FirstAndLastTransportActivityManager.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/FirstAndLastTransportActivityManager.java new file mode 100644 index 0000000000..e6bffa4dfd --- /dev/null +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/FirstAndLastTransportActivityManager.java @@ -0,0 +1,203 @@ +/** + * 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.TransportServiceCallback; +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 java.util.Map; +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 +@ConditionalOnProperty(prefix = "transport.activity", value = "reporting_strategy", havingValue = "first-and-last") +public class FirstAndLastTransportActivityManager 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) { + TransportActivityState activityState = newStateSupplier.get(); + activityState.setLastRecordedTime(newLastRecordedTime); + activityState.setLastReportedTime(0L); + activityStateWrapper = new ActivityStateWrapper(); + activityStateWrapper.setState(activityState); + activityStateWrapper.setAlreadyBeenReported(false); + } else { + activityStateWrapper.getState().setLastRecordedTime(newLastRecordedTime); + } + if (activityStateWrapper.isAlreadyBeenReported()) { + return activityStateWrapper; + } + var activityState = activityStateWrapper.getState(); + if (activityState.getLastReportedTime() < activityState.getLastRecordedTime()) { + reporter.report(key, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() { + @Override + 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 in a period for session.", sessionId); + } + }, MoreExecutors.directExecutor()); + } + + @Override + protected void onReportingPeriodEnd() { + long expirationTime = System.currentTimeMillis() - sessionInactivityTimeout; + 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 { + states.remove(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) { + lastActivityTime = Math.max(gwActivityStateWrapper.getState().getLastRecordedTime(), lastActivityTime); + } + } + } + + boolean hasExpired = sessionMetaData != null && lastActivityTime < expirationTime; + if (hasExpired) { + if (log.isDebugEnabled()) { + log.debug("[{}] Session has expired due to last activity time: [{}]!", sessionId, lastActivityTime); + } + transportService.deregisterSession(sessionInfo); + transportService.process(sessionInfo, SESSION_EVENT_MSG_CLOSED, new TransportServiceCallback<>() { + @Override + public void onSuccess(Void msgAcknowledged) { + states.remove(sessionId); + } + + @Override + public void onError(Throwable e) { + states.remove(sessionId); + } + }); + sessionMetaData.getListener().onRemoteSessionCloseCommand(sessionId, SESSION_EXPIRED_NOTIFICATION_PROTO); + } else if (activityState.getLastReportedTime() < lastActivityTime) { + 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 in a period for session.", sessionId); + } + }); + } + activityStateWrapper.setAlreadyBeenReported(false); + } + } + + 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/FirstOnlyTransportActivityManager.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/FirstOnlyTransportActivityManager.java new file mode 100644 index 0000000000..8cefbb0b46 --- /dev/null +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/FirstOnlyTransportActivityManager.java @@ -0,0 +1,204 @@ +/** + * 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.TransportServiceCallback; +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 java.util.Map; +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 +@ConditionalOnProperty(prefix = "transport.activity", value = "reporting_strategy", havingValue = "first") +public class FirstOnlyTransportActivityManager 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) { + TransportActivityState activityState = newStateSupplier.get(); + activityState.setLastRecordedTime(newLastRecordedTime); + activityState.setLastReportedTime(0L); + activityStateWrapper = new ActivityStateWrapper(); + activityStateWrapper.setState(activityState); + activityStateWrapper.setAlreadyBeenReported(false); + } else { + activityStateWrapper.getState().setLastRecordedTime(newLastRecordedTime); + } + if (activityStateWrapper.isAlreadyBeenReported()) { + return activityStateWrapper; + } + var activityState = activityStateWrapper.getState(); + if (activityState.getLastReportedTime() < activityState.getLastRecordedTime()) { + reporter.report(key, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() { + @Override + 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 in a period for session.", sessionId); + } + }, MoreExecutors.directExecutor()); + } + + @Override + protected void onReportingPeriodEnd() { + long expirationTime = System.currentTimeMillis() - sessionInactivityTimeout; + 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 { + states.remove(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) { + lastActivityTime = Math.max(gwActivityStateWrapper.getState().getLastRecordedTime(), lastActivityTime); + } + } + } + + boolean hasExpired = sessionMetaData != null && lastActivityTime < expirationTime; + if (hasExpired) { + if (log.isDebugEnabled()) { + log.debug("[{}] Session has expired due to last activity time: {}!", sessionId, lastActivityTime); + } + transportService.deregisterSession(sessionInfo); + transportService.process(sessionInfo, SESSION_EVENT_MSG_CLOSED, new TransportServiceCallback<>() { + @Override + public void onSuccess(Void msgAcknowledged) { + states.remove(sessionId); + } + + @Override + public void onError(Throwable e) { + states.remove(sessionId); + } + }); + sessionMetaData.getListener().onRemoteSessionCloseCommand(sessionId, SESSION_EXPIRED_NOTIFICATION_PROTO); + } + if ((sessionMetaData == null || hasExpired) && activityState.getLastReportedTime() < lastActivityTime) { + 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 in a period for session.", sessionId); + } + }); + } + activityStateWrapper.setAlreadyBeenReported(false); + } + } + + 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/LastOnlyTransportActivityManager.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/LastOnlyTransportActivityManager.java index 41f0207f72..bc723e0e4a 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/LastOnlyTransportActivityManager.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/activity/LastOnlyTransportActivityManager.java @@ -31,54 +31,45 @@ package org.thingsboard.server.common.transport.service.activity; import lombok.extern.slf4j.Slf4j; -import org.thingsboard.common.util.ThingsBoardThreadFactory; -import org.thingsboard.server.common.transport.activity.ActivityManager; -import org.thingsboard.server.common.transport.activity.ActivityStateReportCallback; -import org.thingsboard.server.common.transport.activity.ActivityStateReporter; +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.TransportServiceCallback; +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 java.util.Objects; -import java.util.Random; +import java.util.Map; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; import java.util.function.Supplier; -@Slf4j -public class LastOnlyTransportActivityManager implements ActivityManager { +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; - private final ConcurrentMap activityStates = new ConcurrentHashMap<>(); +@Slf4j +@Component +@ConditionalOnProperty(prefix = "transport.activity", value = "reporting_strategy", havingValue = "last") +public class LastOnlyTransportActivityManager extends AbstractActivityManager { - private ScheduledExecutorService scheduler; - private final ActivityStateReporter reporter; - private final long reportingPeriodDurationMillis; - private final String name; - private boolean initialized; + private final ConcurrentMap states = new ConcurrentHashMap<>(); - public LastOnlyTransportActivityManager(ActivityStateReporter reporter, long reportingPeriodDurationMillis, String name) { - this.reporter = Objects.requireNonNull(reporter, "Failed to create activity manager: provided reporter is null."); - this.reportingPeriodDurationMillis = reportingPeriodDurationMillis; // TODO: add min/max duration validation - this.name = name == null ? "activity-state-manager" : name; - } + @Value("${transport.sessions.inactivity_timeout}") + private long sessionInactivityTimeout; - @Override - public synchronized void init() { - if (!initialized) { - scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName(name)); - scheduler.scheduleAtFixedRate(this::onReportingPeriodEnd, new Random().nextInt((int) reportingPeriodDurationMillis), reportingPeriodDurationMillis, TimeUnit.MILLISECONDS); - initialized = true; - } - } + @Autowired + private TransportService transportService; @Override - public void onActivity(UUID activityKey, Supplier newStateSupplier) { + protected void doOnActivity(UUID activityKey, Supplier newStateSupplier) { long newLastRecordedTime = System.currentTimeMillis(); - activityStates.compute(activityKey, (key, activityState) -> { + states.compute(activityKey, (__, activityState) -> { if (activityState == null) { - log.info("Creating new activity state."); activityState = newStateSupplier.get(); activityState.setLastRecordedTime(newLastRecordedTime); activityState.setLastReportedTime(0L); @@ -89,43 +80,73 @@ public class LastOnlyTransportActivityManager implements ActivityManager() { - @Override - public void onSuccess(UUID key, long reportedTime) { - updateLastReportedTime(key, reportedTime); + @Override + protected void onReportingPeriodEnd() { + long expirationTime = System.currentTimeMillis() - sessionInactivityTimeout; + 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 { + states.remove(sessionId); } - @Override - public void onFailure(UUID uuid, Throwable t) { - // TODO: log + 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) { + lastActivityTime = Math.max(gwActivityState.getLastRecordedTime(), lastActivityTime); + } + } } - }); + + boolean hasExpired = sessionMetaData != null && lastActivityTime < expirationTime; + if (hasExpired) { + if (log.isDebugEnabled()) { + log.debug("[{}] Session has expired due to last activity time: [{}]!", sessionId, lastActivityTime); + } + transportService.deregisterSession(sessionInfo); + transportService.process(sessionInfo, SESSION_EVENT_MSG_CLOSED, new TransportServiceCallback<>() { + @Override + public void onSuccess(Void msgAcknowledged) { + states.remove(sessionId); + } + + @Override + public void onError(Throwable e) { + states.remove(sessionId); + } + }); + sessionMetaData.getListener().onRemoteSessionCloseCommand(sessionId, SESSION_EXPIRED_NOTIFICATION_PROTO); + } else if (activityState.getLastReportedTime() < lastActivityTime) { + 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 in a period for session.", sessionId); + } + }); + } + } } private void updateLastReportedTime(UUID key, long newLastReportedTime) { - activityStates.computeIfPresent(key, (__, activityState) -> { + states.computeIfPresent(key, (__, activityState) -> { activityState.setLastReportedTime(Math.max(activityState.getLastReportedTime(), newLastReportedTime)); return activityState; }); } - @Override - public synchronized void destroy() { - if (initialized) { - initialized = false; - if (scheduler != null) { - scheduler.shutdown(); - try { - if (scheduler.awaitTermination(10L, TimeUnit.SECONDS)) { - scheduler.shutdownNow(); - } - } catch (InterruptedException e) { - scheduler.shutdownNow(); - Thread.currentThread().interrupt(); - } - } - } - } - }