From 506151185fc34c4abc3f2c006bb95ca3288f4d9d Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Tue, 2 Mar 2021 16:42:59 +0200 Subject: [PATCH 1/4] added ping for WS --- .../controller/plugin/TbWebSocketHandler.java | 46 ++++++++++++++++++- .../src/main/resources/thingsboard.yml | 1 + 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java index ade9797be0..7c6a2f8c9d 100644 --- a/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java +++ b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java @@ -41,16 +41,25 @@ import org.thingsboard.server.service.telemetry.TelemetryWebSocketMsgEndpoint; import org.thingsboard.server.service.telemetry.TelemetryWebSocketService; import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef; -import javax.websocket.*; +import javax.annotation.PreDestroy; +import javax.websocket.RemoteEndpoint; +import javax.websocket.SendHandler; +import javax.websocket.SendResult; +import javax.websocket.Session; import java.io.IOException; import java.net.URI; +import java.nio.ByteBuffer; import java.security.InvalidParameterException; import java.util.Queue; import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; @Service @TbCoreComponent @@ -79,6 +88,9 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr @Value("${server.ws.limits.max_updates_per_session:}") private String perSessionUpdatesConfiguration; + @Value("${server.ws.ping_timeout:30000}") + private long pingTimeout; + private ConcurrentMap blacklistedSessions = new ConcurrentHashMap<>(); private ConcurrentMap perSessionUpdateLimits = new ConcurrentHashMap<>(); @@ -87,6 +99,8 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr private ConcurrentMap> regularUserSessionsMap = new ConcurrentHashMap<>(); private ConcurrentMap> publicUserSessionsMap = new ConcurrentHashMap<>(); + private ScheduledExecutorService pingExecutor = Executors.newSingleThreadScheduledExecutor(); + @Override public void handleTextMessage(WebSocketSession session, TextMessage message) { try { @@ -120,6 +134,9 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr return; } internalSessionMap.put(internalSessionId, new SessionMetaData(session, sessionRef, maxMsgQueuePerSession)); + + pingExecutor.scheduleWithFixedDelay(() -> internalSessionMap.get(internalSessionId).sendPing(), 10000, 10000, TimeUnit.MILLISECONDS); + externalSessionMap.put(externalSessionId, internalSessionId); processInWebSocketService(sessionRef, SessionEvent.onEstablished()); log.info("[{}][{}][{}] Session is opened", sessionRef.getSecurityCtx().getTenantId(), externalSessionId, session.getId()); @@ -189,6 +206,8 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr private volatile boolean isSending = false; private final Queue msgQueue; + private final AtomicLong lastActivityTime; + SessionMetaData(WebSocketSession session, TelemetryWebSocketSessionRef sessionRef, int maxMsgQueuePerSession) { super(); this.session = session; @@ -196,12 +215,30 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr this.asyncRemote = nativeSession.getAsyncRemote(); this.sessionRef = sessionRef; this.msgQueue = new LinkedBlockingQueue<>(maxMsgQueuePerSession); + this.lastActivityTime = new AtomicLong(System.currentTimeMillis()); + } + + synchronized void sendPing() { + try { + if (System.currentTimeMillis() - lastActivityTime.get() >= pingTimeout) { + this.asyncRemote.sendPing(ByteBuffer.wrap(new byte[]{})); + lastActivityTime.set(System.currentTimeMillis()); + } + } catch (Exception e) { + log.trace("[{}] Failed to send ping msg", session.getId(), e); + try { + close(this.sessionRef, CloseStatus.SESSION_NOT_RELIABLE); + } catch (IOException ioe) { + log.trace("[{}] Session transport error", session.getId(), ioe); + } + } } synchronized void sendMsg(String msg) { if (isSending) { try { msgQueue.add(msg); + lastActivityTime.set(System.currentTimeMillis()); } catch (RuntimeException e) { if (log.isTraceEnabled()) { log.trace("[{}][{}] Session closed due to queue error", sessionRef.getSecurityCtx().getTenantId(), session.getId(), e); @@ -393,4 +430,11 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr } } + @PreDestroy + private void destroy() { + if (pingExecutor != null) { + pingExecutor.shutdownNow(); + } + } + } \ No newline at end of file diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 7097acd4dd..f750000750 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -34,6 +34,7 @@ server: log_controller_error_stack_trace: "${HTTP_LOG_CONTROLLER_ERROR_STACK_TRACE:false}" ws: send_timeout: "${TB_SERVER_WS_SEND_TIMEOUT:5000}" + ping_timeout: "${TB_SERVER_WS_PING_TIMEOUT:30000}" limits: # Limit the amount of sessions and subscriptions available on each server. Put values to zero to disable particular limitation max_sessions_per_tenant: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SESSIONS_PER_TENANT:0}" From f68158c550b2b04c9a9032359a3c0d39af0393d8 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Wed, 3 Mar 2021 12:15:21 +0200 Subject: [PATCH 2/4] improvements --- .../controller/plugin/TbWebSocketHandler.java | 42 +++++++++---------- .../DefaultTelemetryWebSocketService.java | 29 ++++++++++--- .../TelemetryWebSocketMsgEndpoint.java | 2 + 3 files changed, 47 insertions(+), 26 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java index 7c6a2f8c9d..13a54e5110 100644 --- a/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java +++ b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java @@ -41,7 +41,6 @@ import org.thingsboard.server.service.telemetry.TelemetryWebSocketMsgEndpoint; import org.thingsboard.server.service.telemetry.TelemetryWebSocketService; import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef; -import javax.annotation.PreDestroy; import javax.websocket.RemoteEndpoint; import javax.websocket.SendHandler; import javax.websocket.SendResult; @@ -55,11 +54,7 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicLong; @Service @TbCoreComponent @@ -99,8 +94,6 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr private ConcurrentMap> regularUserSessionsMap = new ConcurrentHashMap<>(); private ConcurrentMap> publicUserSessionsMap = new ConcurrentHashMap<>(); - private ScheduledExecutorService pingExecutor = Executors.newSingleThreadScheduledExecutor(); - @Override public void handleTextMessage(WebSocketSession session, TextMessage message) { try { @@ -135,8 +128,6 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr } internalSessionMap.put(internalSessionId, new SessionMetaData(session, sessionRef, maxMsgQueuePerSession)); - pingExecutor.scheduleWithFixedDelay(() -> internalSessionMap.get(internalSessionId).sendPing(), 10000, 10000, TimeUnit.MILLISECONDS); - externalSessionMap.put(externalSessionId, internalSessionId); processInWebSocketService(sessionRef, SessionEvent.onEstablished()); log.info("[{}][{}][{}] Session is opened", sessionRef.getSecurityCtx().getTenantId(), externalSessionId, session.getId()); @@ -206,7 +197,7 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr private volatile boolean isSending = false; private final Queue msgQueue; - private final AtomicLong lastActivityTime; + private volatile long lastActivityTime; SessionMetaData(WebSocketSession session, TelemetryWebSocketSessionRef sessionRef, int maxMsgQueuePerSession) { super(); @@ -215,14 +206,14 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr this.asyncRemote = nativeSession.getAsyncRemote(); this.sessionRef = sessionRef; this.msgQueue = new LinkedBlockingQueue<>(maxMsgQueuePerSession); - this.lastActivityTime = new AtomicLong(System.currentTimeMillis()); + this.lastActivityTime = System.currentTimeMillis(); } synchronized void sendPing() { try { - if (System.currentTimeMillis() - lastActivityTime.get() >= pingTimeout) { + if (System.currentTimeMillis() - lastActivityTime >= pingTimeout) { this.asyncRemote.sendPing(ByteBuffer.wrap(new byte[]{})); - lastActivityTime.set(System.currentTimeMillis()); + lastActivityTime = System.currentTimeMillis(); } } catch (Exception e) { log.trace("[{}] Failed to send ping msg", session.getId(), e); @@ -238,7 +229,7 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr if (isSending) { try { msgQueue.add(msg); - lastActivityTime.set(System.currentTimeMillis()); + lastActivityTime = System.currentTimeMillis(); } catch (RuntimeException e) { if (log.isTraceEnabled()) { log.trace("[{}][{}] Session closed due to queue error", sessionRef.getSecurityCtx().getTenantId(), session.getId(), e); @@ -321,6 +312,22 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr } } + @Override + public void sendPing(TelemetryWebSocketSessionRef sessionRef) throws IOException { + String externalId = sessionRef.getSessionId(); + String internalId = externalSessionMap.get(externalId); + if (internalId != null) { + SessionMetaData sessionMd = internalSessionMap.get(internalId); + if (sessionMd != null) { + sessionMd.sendPing(); + } else { + log.warn("[{}][{}] Failed to find session by internal id", externalId, internalId); + } + } else { + log.warn("[{}] Failed to find session by external id", externalId); + } + } + @Override public void close(TelemetryWebSocketSessionRef sessionRef, CloseStatus reason) throws IOException { String externalId = sessionRef.getSessionId(); @@ -430,11 +437,4 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr } } - @PreDestroy - private void destroy() { - if (pingExecutor != null) { - pingExecutor.shutdownNow(); - } - } - } \ No newline at end of file diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java index 2efbc2d0a8..1987bc354b 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java @@ -51,22 +51,20 @@ import org.thingsboard.server.service.security.ValidationResult; import org.thingsboard.server.service.security.ValidationResultCode; import org.thingsboard.server.service.security.model.UserPrincipal; import org.thingsboard.server.service.security.permission.Operation; +import org.thingsboard.server.service.subscription.TbAttributeSubscription; +import org.thingsboard.server.service.subscription.TbAttributeSubscriptionScope; import org.thingsboard.server.service.subscription.TbEntityDataSubscriptionService; import org.thingsboard.server.service.subscription.TbLocalSubscriptionService; -import org.thingsboard.server.service.subscription.TbAttributeSubscriptionScope; -import org.thingsboard.server.service.subscription.TbAttributeSubscription; import org.thingsboard.server.service.subscription.TbTimeseriesSubscription; +import org.thingsboard.server.service.telemetry.cmd.TelemetryPluginCmdsWrapper; import org.thingsboard.server.service.telemetry.cmd.v1.AttributesSubscriptionCmd; import org.thingsboard.server.service.telemetry.cmd.v1.GetHistoryCmd; import org.thingsboard.server.service.telemetry.cmd.v1.SubscriptionCmd; import org.thingsboard.server.service.telemetry.cmd.v1.TelemetryPluginCmd; -import org.thingsboard.server.service.telemetry.cmd.TelemetryPluginCmdsWrapper; import org.thingsboard.server.service.telemetry.cmd.v1.TimeseriesSubscriptionCmd; import org.thingsboard.server.service.telemetry.cmd.v2.AlarmDataCmd; -import org.thingsboard.server.service.telemetry.cmd.v2.AlarmDataUnsubscribeCmd; import org.thingsboard.server.service.telemetry.cmd.v2.DataUpdate; import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd; -import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUnsubscribeCmd; import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate; import org.thingsboard.server.service.telemetry.cmd.v2.UnsubscribeCmd; import org.thingsboard.server.service.telemetry.exception.UnauthorizedException; @@ -89,6 +87,8 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -151,14 +151,23 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi private ExecutorService executor; private String serviceId; + private ScheduledExecutorService pingExecutor; + @PostConstruct public void initExecutor() { serviceId = serviceInfoProvider.getServiceId(); executor = Executors.newWorkStealingPool(50); + + pingExecutor = Executors.newSingleThreadScheduledExecutor(); + pingExecutor.scheduleWithFixedDelay(this::sendPing, 10000, 10000, TimeUnit.MILLISECONDS); } @PreDestroy public void shutdownExecutor() { + if (pingExecutor != null) { + pingExecutor.shutdownNow(); + } + if (executor != null) { executor.shutdownNow(); } @@ -744,6 +753,16 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi } } + private void sendPing() { + wsSessionsMap.values().forEach(md -> + executor.submit(() -> { + try { + msgEndpoint.sendPing(md.getSessionRef()); + } catch (IOException e) { + log.warn("[{}] Failed to send ping: {}", md.getSessionRef().getSessionId(), e); + } + })); + } private static Optional> getKeys(TelemetryPluginCmd cmd) { if (!StringUtils.isEmpty(cmd.getKeys())) { diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketMsgEndpoint.java b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketMsgEndpoint.java index 2c132c745c..9a69e25009 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketMsgEndpoint.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketMsgEndpoint.java @@ -26,5 +26,7 @@ public interface TelemetryWebSocketMsgEndpoint { void send(TelemetryWebSocketSessionRef sessionRef, int subscriptionId, String msg) throws IOException; + void sendPing(TelemetryWebSocketSessionRef sessionRef) throws IOException; + void close(TelemetryWebSocketSessionRef sessionRef, CloseStatus withReason) throws IOException; } From e318b49c2dbfa810400f937b9f4d78989ce81b9c Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Wed, 3 Mar 2021 13:22:31 +0200 Subject: [PATCH 3/4] refactoring --- .../server/controller/plugin/TbWebSocketHandler.java | 10 +++++----- .../telemetry/DefaultTelemetryWebSocketService.java | 3 ++- .../telemetry/TelemetryWebSocketMsgEndpoint.java | 2 +- 3 files changed, 8 insertions(+), 7 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java index 13a54e5110..aeb595a193 100644 --- a/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java +++ b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java @@ -209,11 +209,11 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr this.lastActivityTime = System.currentTimeMillis(); } - synchronized void sendPing() { + synchronized void sendPing(long currentTime) { try { - if (System.currentTimeMillis() - lastActivityTime >= pingTimeout) { + if (currentTime - lastActivityTime >= pingTimeout) { this.asyncRemote.sendPing(ByteBuffer.wrap(new byte[]{})); - lastActivityTime = System.currentTimeMillis(); + lastActivityTime = currentTime; } } catch (Exception e) { log.trace("[{}] Failed to send ping msg", session.getId(), e); @@ -313,13 +313,13 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr } @Override - public void sendPing(TelemetryWebSocketSessionRef sessionRef) throws IOException { + public void sendPing(TelemetryWebSocketSessionRef sessionRef, long currentTime) throws IOException { String externalId = sessionRef.getSessionId(); String internalId = externalSessionMap.get(externalId); if (internalId != null) { SessionMetaData sessionMd = internalSessionMap.get(internalId); if (sessionMd != null) { - sessionMd.sendPing(); + sessionMd.sendPing(currentTime); } else { log.warn("[{}][{}] Failed to find session by internal id", externalId, internalId); } diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java index 1987bc354b..07d7af3165 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java @@ -754,10 +754,11 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi } private void sendPing() { + long currentTime = System.currentTimeMillis(); wsSessionsMap.values().forEach(md -> executor.submit(() -> { try { - msgEndpoint.sendPing(md.getSessionRef()); + msgEndpoint.sendPing(md.getSessionRef(), currentTime); } catch (IOException e) { log.warn("[{}] Failed to send ping: {}", md.getSessionRef().getSessionId(), e); } diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketMsgEndpoint.java b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketMsgEndpoint.java index 9a69e25009..f10038f116 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketMsgEndpoint.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketMsgEndpoint.java @@ -26,7 +26,7 @@ public interface TelemetryWebSocketMsgEndpoint { void send(TelemetryWebSocketSessionRef sessionRef, int subscriptionId, String msg) throws IOException; - void sendPing(TelemetryWebSocketSessionRef sessionRef) throws IOException; + void sendPing(TelemetryWebSocketSessionRef sessionRef, long currentTime) throws IOException; void close(TelemetryWebSocketSessionRef sessionRef, CloseStatus withReason) throws IOException; } From e3851dcaad4864ab9d00df383b2c5bc9750ae0dd Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Thu, 4 Mar 2021 10:50:57 +0200 Subject: [PATCH 4/4] improvements --- .../server/controller/plugin/TbWebSocketHandler.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java index aeb595a193..166403f417 100644 --- a/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java +++ b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java @@ -229,7 +229,6 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr if (isSending) { try { msgQueue.add(msg); - lastActivityTime = System.currentTimeMillis(); } catch (RuntimeException e) { if (log.isTraceEnabled()) { log.trace("[{}][{}] Session closed due to queue error", sessionRef.getSecurityCtx().getTenantId(), session.getId(), e); @@ -271,6 +270,7 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr log.trace("[{}] Session transport error", session.getId(), ioe); } } else { + lastActivityTime = System.currentTimeMillis(); String msg = msgQueue.poll(); if (msg != null) { sendMsgInternal(msg);