Fixes after manual testing
This commit is contained in:
parent
7431233f5d
commit
315202d9df
@ -63,7 +63,6 @@ import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509Ce
|
|||||||
import org.thingsboard.server.gen.transport.TransportProtos.ValidateOrCreateDeviceX509CertRequestMsg;
|
import org.thingsboard.server.gen.transport.TransportProtos.ValidateOrCreateDeviceX509CertRequestMsg;
|
||||||
|
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.UUID;
|
|
||||||
import java.util.concurrent.ExecutorService;
|
import java.util.concurrent.ExecutorService;
|
||||||
import java.util.concurrent.atomic.AtomicInteger;
|
import java.util.concurrent.atomic.AtomicInteger;
|
||||||
|
|
||||||
@ -162,7 +161,5 @@ public interface TransportService {
|
|||||||
|
|
||||||
boolean hasSession(SessionInfoProto sessionInfo);
|
boolean hasSession(SessionInfoProto sessionInfo);
|
||||||
|
|
||||||
SessionMetaData getSession(UUID sessionId);
|
|
||||||
|
|
||||||
void createGaugeStats(String openConnections, AtomicInteger connectionsCounter);
|
void createGaugeStats(String openConnections, AtomicInteger connectionsCounter);
|
||||||
}
|
}
|
||||||
|
|||||||
@ -31,16 +31,15 @@
|
|||||||
package org.thingsboard.server.common.transport.activity;
|
package org.thingsboard.server.common.transport.activity;
|
||||||
|
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
import org.thingsboard.common.util.ThingsBoardThreadFactory;
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
import org.thingsboard.server.common.data.StringUtils;
|
import org.thingsboard.server.common.data.StringUtils;
|
||||||
import org.thingsboard.server.common.transport.activity.strategy.ActivityStrategy;
|
import org.thingsboard.server.common.transport.activity.strategy.ActivityStrategy;
|
||||||
|
import org.thingsboard.server.queue.scheduler.SchedulerComponent;
|
||||||
|
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
import java.util.Random;
|
import java.util.Random;
|
||||||
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.ConcurrentHashMap;
|
||||||
import java.util.concurrent.ConcurrentMap;
|
import java.util.concurrent.ConcurrentMap;
|
||||||
import java.util.concurrent.Executors;
|
|
||||||
import java.util.concurrent.ScheduledExecutorService;
|
|
||||||
import java.util.concurrent.TimeUnit;
|
import java.util.concurrent.TimeUnit;
|
||||||
import java.util.concurrent.atomic.AtomicBoolean;
|
import java.util.concurrent.atomic.AtomicBoolean;
|
||||||
|
|
||||||
@ -48,9 +47,11 @@ import java.util.concurrent.atomic.AtomicBoolean;
|
|||||||
public abstract class AbstractActivityManager<Key, Metadata> implements ActivityManager<Key> {
|
public abstract class AbstractActivityManager<Key, Metadata> implements ActivityManager<Key> {
|
||||||
|
|
||||||
private final ConcurrentMap<Key, ActivityState<Metadata>> states = new ConcurrentHashMap<>();
|
private final ConcurrentMap<Key, ActivityState<Metadata>> states = new ConcurrentHashMap<>();
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
protected SchedulerComponent scheduler;
|
||||||
|
|
||||||
protected String name;
|
protected String name;
|
||||||
private long reportingPeriodMillis;
|
|
||||||
private ScheduledExecutorService scheduler;
|
|
||||||
private boolean initialized;
|
private boolean initialized;
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
@ -62,8 +63,6 @@ public abstract class AbstractActivityManager<Key, Metadata> implements Activity
|
|||||||
reportingPeriodMillis = 3000;
|
reportingPeriodMillis = 3000;
|
||||||
log.error("[{}] Negative or zero reporting period millisecond was provided. Going to use reporting period value of 3 seconds.", this.name);
|
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)); // TODO: use scheduler component
|
|
||||||
scheduler.scheduleAtFixedRate(this::onReportingPeriodEnd, new Random().nextInt((int) reportingPeriodMillis), reportingPeriodMillis, TimeUnit.MILLISECONDS);
|
scheduler.scheduleAtFixedRate(this::onReportingPeriodEnd, new Random().nextInt((int) reportingPeriodMillis), reportingPeriodMillis, TimeUnit.MILLISECONDS);
|
||||||
initialized = true;
|
initialized = true;
|
||||||
log.info("Activity manager with name [{}] is initialized.", this.name);
|
log.info("Activity manager with name [{}] is initialized.", this.name);
|
||||||
@ -172,7 +171,7 @@ public abstract class AbstractActivityManager<Key, Metadata> implements Activity
|
|||||||
|
|
||||||
if (shouldReport && lastReportedTime < lastRecordedTime) {
|
if (shouldReport && lastReportedTime < lastRecordedTime) {
|
||||||
log.debug("[{}] Going to report last activity event for key: [{}].", name, key);
|
log.debug("[{}] Going to report last activity event for key: [{}].", name, key);
|
||||||
reportActivity(key, metadata, currentState.getLastRecordedTime(), new ActivityReportCallback<>() {
|
reportActivity(key, metadata, lastRecordedTime, new ActivityReportCallback<>() {
|
||||||
@Override
|
@Override
|
||||||
public void onSuccess(Key key, long reportedTime) {
|
public void onSuccess(Key key, long reportedTime) {
|
||||||
updateLastReportedTime(key, reportedTime);
|
updateLastReportedTime(key, reportedTime);
|
||||||
@ -194,22 +193,4 @@ public abstract class AbstractActivityManager<Key, Metadata> implements Activity
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
@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();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
@ -36,6 +36,4 @@ public interface ActivityManager<Key> {
|
|||||||
|
|
||||||
void onActivity(Key key);
|
void onActivity(Key key);
|
||||||
|
|
||||||
void destroy();
|
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
@ -186,7 +186,6 @@ public class DefaultTransportService extends AbstractActivityManager<UUID, Trans
|
|||||||
|
|
||||||
private final TransportRateLimitService rateLimitService;
|
private final TransportRateLimitService rateLimitService;
|
||||||
private final DataDecodingEncodingService dataDecodingEncodingService;
|
private final DataDecodingEncodingService dataDecodingEncodingService;
|
||||||
private final SchedulerComponent scheduler;
|
|
||||||
private final ApplicationEventPublisher eventPublisher;
|
private final ApplicationEventPublisher eventPublisher;
|
||||||
private final TransportResourceCache transportResourceCache;
|
private final TransportResourceCache transportResourceCache;
|
||||||
private final NotificationRuleProcessor notificationRuleProcessor;
|
private final NotificationRuleProcessor notificationRuleProcessor;
|
||||||
@ -295,7 +294,6 @@ public class DefaultTransportService extends AbstractActivityManager<UUID, Trans
|
|||||||
@PreDestroy
|
@PreDestroy
|
||||||
public void destroy() {
|
public void destroy() {
|
||||||
stopped = true;
|
stopped = true;
|
||||||
super.destroy();
|
|
||||||
|
|
||||||
if (transportNotificationsConsumer != null) {
|
if (transportNotificationsConsumer != null) {
|
||||||
transportNotificationsConsumer.unsubscribe();
|
transportNotificationsConsumer.unsubscribe();
|
||||||
@ -838,7 +836,7 @@ public class DefaultTransportService extends AbstractActivityManager<UUID, Trans
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected boolean hasExpired(UUID uuid, ActivityState<TransportProtos.SessionInfoProto> state) {
|
protected boolean hasExpired(UUID uuid, ActivityState<TransportProtos.SessionInfoProto> state) {
|
||||||
return (System.currentTimeMillis() - sessionInactivityTimeout) < state.getLastReportedTime();
|
return (System.currentTimeMillis() - sessionInactivityTimeout) > state.getLastRecordedTime();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
@ -1351,11 +1349,6 @@ public class DefaultTransportService extends AbstractActivityManager<UUID, Trans
|
|||||||
return sessions.containsKey(toSessionId(sessionInfo));
|
return sessions.containsKey(toSessionId(sessionInfo));
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
|
||||||
public SessionMetaData getSession(UUID sessionId) {
|
|
||||||
return sessions.get(sessionId);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void createGaugeStats(String statsName, AtomicInteger number) {
|
public void createGaugeStats(String statsName, AtomicInteger number) {
|
||||||
statsFactory.createGauge(StatsType.TRANSPORT + "." + statsName, number);
|
statsFactory.createGauge(StatsType.TRANSPORT + "." + statsName, number);
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user