tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

[WIP] Refactor abstract and integration activity managers

pull/9980/head
Dmytro Skarzhynets 3 years ago
parent
2ca1f71998
commit
c3556eba7d
5 changed files with 72 additions and 83 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 31
      application/src/main/java/org/thingsboard/server/service/integration/activity/FirstAndLastIntegrationActivityManager.java
  2. 59
      application/src/main/java/org/thingsboard/server/service/integration/activity/FirstOnlyIntegrationActivityManager.java
  3. 15
      application/src/main/java/org/thingsboard/server/service/integration/activity/LastOnlyIntegrationActivityManager.java
  4. 42
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/AbstractActivityManager.java
  5. 8
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/activity/ActivityManager.java

31
application/src/main/java/org/thingsboard/server/service/integration/activity/FirstAndLastIntegrationActivityManager.java
View File

@ -70,19 +70,16 @@ public class FirstAndLastIntegrationActivityManager extends AbstractActivityMana
SettableFuture<Pair<IntegrationActivityKey, Long>> reportCompletedFuture = SettableFuture.create();
states.compute(activityKey, (key, activityStateWrapper) -> {
if (activityStateWrapper == null) {
ActivityState activityState = newStateSupplier.get();
activityState.setLastRecordedTime(newLastRecordedTime);
activityState.setLastReportedTime(0L);
activityStateWrapper = new ActivityStateWrapper();
activityStateWrapper.setState(activityState);
activityStateWrapper.setAlreadyBeenReported(false);
} else {
activityStateWrapper.getState().setLastRecordedTime(newLastRecordedTime);
activityStateWrapper.setState(newStateSupplier.get());
}
var activityState = activityStateWrapper.getState();
if (activityState.getLastRecordedTime() < newLastRecordedTime) {
activityState.setLastRecordedTime(newLastRecordedTime);
}
if (activityStateWrapper.isAlreadyBeenReported()) {
return activityStateWrapper;
}
var activityState = activityStateWrapper.getState();
if (activityState.getLastReportedTime() < activityState.getLastRecordedTime()) {
reporter.report(key, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() {
@Override
@ -107,36 +104,36 @@ public class FirstAndLastIntegrationActivityManager extends AbstractActivityMana
@Override
public void onFailure(@NonNull Throwable t) {
log.debug("[{}] Failed to report first activity event in a period for device with id: [{}]", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId());
log.debug("[{}] Failed to report first activity event in a period for device with id: [{}].", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId());
}
}, MoreExecutors.directExecutor());
}
@Override
protected void onReportingPeriodEnd() {
long expirationTime = System.currentTimeMillis() - reportingPeriodMillis;
for (Map.Entry<IntegrationActivityKey, ActivityStateWrapper> entry : states.entrySet()) {
var activityKey = entry.getKey();
var activityStateWrapper = entry.getValue();
var activityState = activityStateWrapper.getState();
if (activityState.getLastRecordedTime() < expirationTime) {
long lastRecordedTime = activityState.getLastRecordedTime();
// if there were no activities during the reporting period, we should remove the entry to prevent memory leaks
if (!activityStateWrapper.isAlreadyBeenReported()) {
states.remove(activityKey);
}
if (activityState.getLastReportedTime() < activityState.getLastRecordedTime()) {
reporter.report(activityKey, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() {
if (activityState.getLastReportedTime() < lastRecordedTime) {
reporter.report(activityKey, lastRecordedTime, activityState, new ActivityReportCallback<>() {
@Override
public void onSuccess(IntegrationActivityKey key, long reportedTime) {
updateLastReportedTime(key, reportedTime);
public void onSuccess(IntegrationActivityKey key, long newLastReportedTime) {
updateLastReportedTime(key, newLastReportedTime);
}
@Override
public void onFailure(IntegrationActivityKey key, Throwable t) {
log.debug("[{}] Failed to report last activity event in a period for device with id: [{}]", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId());
log.debug("[{}] Failed to report last activity event in a period for device with id: [{}].", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId());
}
});
}
activityStateWrapper.setAlreadyBeenReported(false);
activityState.getReportsCount().set(0);
}
}

59
application/src/main/java/org/thingsboard/server/service/integration/activity/FirstOnlyIntegrationActivityManager.java
View File

@ -70,19 +70,16 @@ public class FirstOnlyIntegrationActivityManager extends AbstractActivityManager
SettableFuture<Pair<IntegrationActivityKey, Long>> reportCompletedFuture = SettableFuture.create();
states.compute(activityKey, (key, activityStateWrapper) -> {
if (activityStateWrapper == null) {
ActivityState activityState = newStateSupplier.get();
activityState.setLastRecordedTime(newLastRecordedTime);
activityState.setLastReportedTime(0L);
activityStateWrapper = new ActivityStateWrapper();
activityStateWrapper.setState(activityState);
activityStateWrapper.setAlreadyBeenReported(false);
} else {
activityStateWrapper.getState().setLastRecordedTime(newLastRecordedTime);
activityStateWrapper.setState(newStateSupplier.get());
}
var activityState = activityStateWrapper.getState();
if (activityState.getLastRecordedTime() < newLastRecordedTime) {
activityState.setLastRecordedTime(newLastRecordedTime);
}
if (activityStateWrapper.isAlreadyBeenReported()) {
return activityStateWrapper;
}
var activityState = activityStateWrapper.getState();
if (activityState.getLastReportedTime() < activityState.getLastRecordedTime()) {
reporter.report(key, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() {
@Override
@ -107,40 +104,46 @@ public class FirstOnlyIntegrationActivityManager extends AbstractActivityManager
@Override
public void onFailure(@NonNull Throwable t) {
log.debug("[{}] Failed to report first activity event in a period for device with id: [{}]", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId());
log.debug("[{}] Failed to report first activity event in a period for device with id: [{}].", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId());
}
}, MoreExecutors.directExecutor());
}
private void updateLastReportedTime(IntegrationActivityKey key, long newLastReportedTime) {
states.computeIfPresent(key, (__, activityStateWrapper) -> {
var activityState = activityStateWrapper.getState();
activityState.setLastReportedTime(Math.max(activityState.getLastReportedTime(), newLastReportedTime));
return activityStateWrapper;
});
}
@Override
protected void onReportingPeriodEnd() {
for (Map.Entry<IntegrationActivityKey, ActivityStateWrapper> entry : states.entrySet()) {
var activityKey = entry.getKey();
var activityStateWrapper = entry.getValue();
var activityState = activityStateWrapper.getState();
if (!activityStateWrapper.isAlreadyBeenReported() && activityState.getLastReportedTime() < activityState.getLastRecordedTime()) {
reporter.report(activityKey, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() {
@Override
public void onSuccess(IntegrationActivityKey key, long reportedTime) {
states.remove(key);
}
long lastRecordedTime = activityState.getLastRecordedTime();
// if there were no activities during the reporting period, we should remove the entry to prevent memory leaks
if (!activityStateWrapper.isAlreadyBeenReported()) {
states.remove(activityKey);
// report leftover events
if (activityState.getLastReportedTime() < lastRecordedTime) {
reporter.report(activityKey, lastRecordedTime, activityState, new ActivityReportCallback<>() {
@Override
public void onSuccess(IntegrationActivityKey key, long reportedTime) {
updateLastReportedTime(key, reportedTime); // just in case the same key was added again
}
@Override
public void onFailure(IntegrationActivityKey key, Throwable t) {
log.debug("[{}] Failed to report last activity event in a period for device with id: [{}]", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId());
}
});
@Override
public void onFailure(IntegrationActivityKey key, Throwable t) {
log.debug("[{}] Failed to report last activity event in a period for device with id: [{}].", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId());
}
});
}
}
activityStateWrapper.setAlreadyBeenReported(false);
}
}
private void updateLastReportedTime(IntegrationActivityKey key, long newLastReportedTime) {
states.computeIfPresent(key, (__, activityStateWrapper) -> {
var activityState = activityStateWrapper.getState();
activityState.setLastReportedTime(Math.max(activityState.getLastReportedTime(), newLastReportedTime));
return activityStateWrapper;
});
}
}

15
application/src/main/java/org/thingsboard/server/service/integration/activity/LastOnlyIntegrationActivityManager.java
View File

@ -55,9 +55,8 @@ public class LastOnlyIntegrationActivityManager extends AbstractActivityManager<
states.compute(activityKey, (__, activityState) -> {
if (activityState == null) {
activityState = newStateSupplier.get();
activityState.setLastRecordedTime(newLastRecordedTime);
activityState.setLastReportedTime(0L);
} else {
}
if (activityState.getLastRecordedTime() < newLastRecordedTime) {
activityState.setLastRecordedTime(newLastRecordedTime);
}
return activityState;
@ -70,11 +69,13 @@ public class LastOnlyIntegrationActivityManager extends AbstractActivityManager<
for (Map.Entry<IntegrationActivityKey, ActivityState> entry : states.entrySet()) {
var activityKey = entry.getKey();
var activityState = entry.getValue();
if (activityState.getLastRecordedTime() < expirationTime) {
long lastRecordedTime = activityState.getLastRecordedTime();
// if there were no activities during the reporting period, we should remove the entry to prevent memory leaks
if (lastRecordedTime < expirationTime) {
states.remove(activityKey);
}
if (activityState.getLastReportedTime() < activityState.getLastRecordedTime()) {
reporter.report(activityKey, activityState.getLastRecordedTime(), activityState, new ActivityReportCallback<>() {
if (activityState.getLastReportedTime() < lastRecordedTime) {
reporter.report(activityKey, lastRecordedTime, activityState, new ActivityReportCallback<>() {
@Override
public void onSuccess(IntegrationActivityKey key, long reportedTime) {
updateLastReportedTime(key, reportedTime);
@ -82,7 +83,7 @@ public class LastOnlyIntegrationActivityManager extends AbstractActivityManager<
@Override
public void onFailure(IntegrationActivityKey key, Throwable t) {
log.debug("[{}] Failed to report last activity event in a period for device with id: [{}]", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId());
log.debug("[{}] Failed to report last activity event in a period for device with id: [{}].", activityKey.getTenantId().getId(), activityKey.getDeviceId().getId());
}
});
}

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

@ -30,7 +30,9 @@
*/
package org.thingsboard.server.common.transport.activity;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.common.util.ThingsBoardThreadFactory;
import org.thingsboard.server.common.data.StringUtils;
import java.util.Objects;
import java.util.Random;
@ -39,52 +41,44 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
@Slf4j
public abstract class AbstractActivityManager<Key, State extends ActivityState> implements ActivityManager<Key, State> {
private String name;
protected long reportingPeriodMillis;
protected ActivityStateReporter<Key, State> reporter;
private ScheduledExecutorService scheduler;
protected boolean initialized;
private boolean initialized;
@Override
public synchronized void init() {
Objects.requireNonNull(reporter, "Failed to initialize activity manager: provided activity reporter is null.");
if (reportingPeriodMillis < 1000) {
throw new IllegalArgumentException("Failed to initialize activity manager: provided reporting period duration is less that 1 second.");
public synchronized void init(String name, long reportingPeriodMillis, ActivityStateReporter<Key, State> reporter) {
this.name = StringUtils.notBlankOrDefault(name, "activity-manager");
this.reporter = Objects.requireNonNull(reporter, "Failed to initialize activity manager: provided activity reporter is null.");
if (reportingPeriodMillis <= 0) {
reportingPeriodMillis = 3000;
log.error("[{}] Negative or zero reporting period millisecond was provided. Going to use reporting period value of 3 seconds.", this.name);
}
this.reportingPeriodMillis = reportingPeriodMillis;
if (!initialized) {
scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName(name == null ? "activity-manager" : name));
scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName(this.name));
scheduler.scheduleAtFixedRate(this::onReportingPeriodEnd, new Random().nextInt((int) reportingPeriodMillis), reportingPeriodMillis, TimeUnit.MILLISECONDS);
initialized = true;
}
}
@Override
public void setName(String name) {
this.name = name;
}
@Override
public void setReportingPeriod(long reportingPeriodMillis) {
this.reportingPeriodMillis = reportingPeriodMillis;
}
@Override
public void setActivityReporter(ActivityStateReporter<Key, State> activityReporter) {
reporter = activityReporter;
}
@Override
public void onActivity(Key key, Supplier<State> newStateSupplier) {
if (!initialized) {
throw new IllegalStateException(name + " is not initialized.");
log.error("[{}] Failed to process activity event: activity manager is not initialized.", name);
return;
}
if (key == null) {
throw new IllegalArgumentException("Activity key can't be null.");
log.error("[{}] Failed to process activity event: provided activity key is null.", name);
return;
}
if (newStateSupplier == null) {
throw new IllegalArgumentException("New activity state supplier can't be null.");
log.error("[{}] Failed to process activity event: provided new activity state supplier is null.", name);
return;
}
doOnActivity(key, newStateSupplier);
}

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

@ -34,13 +34,7 @@ import java.util.function.Supplier;
public interface ActivityManager<Key, State extends ActivityState> {
void setName(String name);
void setReportingPeriod(long reportingPeriodMillis);
void setActivityReporter(ActivityStateReporter<Key, State> activityReporter);
void init();
void init(String name, long reportingPeriodMillis, ActivityStateReporter<Key, State> reporter);
void onActivity(Key key, Supplier<State> newStateSupplier);

Write Preview
Loading…
Cancel
Save