This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new ec9d1c113a fix(admin): drop closed websocket sessions from the
namespace map (#7317)
ec9d1c113a is described below
commit ec9d1c113a9c81290410f4b13c8d55524bb21328
Author: Burak Kalaycı <[email protected]>
AuthorDate: Sat Sep 26 14:16:21 2026 +0300
fix(admin): drop closed websocket sessions from the namespace map (#7317)
@OnClose runs after the session is already closed, so reading the
namespace from the session skipped NAMESPACE_SESSION_MAP removal.
Remember the namespace at registration and remove every index from
that record. Reject a missing namespace before any index insert.
Fixes #7315
Co-authored-by: aias00 <[email protected]>
---
.../listener/websocket/WebsocketCollector.java | 62 +++++++++---
.../listener/websocket/WebsocketCollectorTest.java | 105 +++++++++++++++++++++
2 files changed, 153 insertions(+), 14 deletions(-)
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/websocket/WebsocketCollector.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/websocket/WebsocketCollector.java
index 7260b5ac70..c4d7b578c5 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/websocket/WebsocketCollector.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/websocket/WebsocketCollector.java
@@ -47,6 +47,7 @@ import jakarta.websocket.Session;
import jakarta.websocket.server.ServerEndpoint;
import java.util.ArrayDeque;
+import java.util.ArrayList;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
@@ -69,6 +70,12 @@ public class WebsocketCollector {
private static final Map<String, Set<Session>> NAMESPACE_SESSION_MAP =
Maps.newConcurrentMap();
private static final Map<Session, SessionSendQueue> SESSION_SEND_QUEUES =
Maps.newConcurrentMap();
+
+ /**
+ * Namespace captured at registration. {@code Session#isOpen()} is already
false when
+ * {@code @OnClose} runs, so the namespace cannot be read from the session
at teardown.
+ */
+ private static final Map<Session, String> SESSION_NAMESPACE_IDS =
Maps.newConcurrentMap();
private static final String SESSION_KEY = "sessionKey";
@@ -82,14 +89,18 @@ public class WebsocketCollector {
String clientIp = getClientIp(session);
LOG.info("websocket on client[{}] open successful,
maxTextMessageBufferSize: {}",
clientIp, session.getMaxTextMessageBufferSize());
- SESSION_SET.add(session);
-
String namespaceId = getNamespaceId(session);
if (StringUtils.isBlank(namespaceId)) {
throw new ShenyuException("websocket on client open failed,
namespaceId is null");
}
+ SESSION_SET.add(session);
+ SESSION_NAMESPACE_IDS.put(session, namespaceId);
LOG.info("websocket on client[{}] open successful, namespaceId: {}",
clientIp, namespaceId);
- NAMESPACE_SESSION_MAP.computeIfAbsent(namespaceId, k ->
Sets.newConcurrentHashSet()).add(session);
+ NAMESPACE_SESSION_MAP.compute(namespaceId, (id, sessions) -> {
+ Set<Session> registered = Objects.isNull(sessions) ?
Sets.newConcurrentHashSet() : sessions;
+ registered.add(session);
+ return registered;
+ });
}
private static String getClientIp(final Session session) {
@@ -251,12 +262,17 @@ public class WebsocketCollector {
if (session.isOpen()) {
sendMessageBySession(session, message);
} else {
- SESSION_SET.remove(session);
- removeSessionSendQueue(session);
+ removeSessionIndexes(session);
}
}
} else {
- SESSION_SET.forEach(session -> sendMessageBySession(session,
message));
+ for (Session registered : new ArrayList<>(SESSION_SET)) {
+ if (registered.isOpen()) {
+ sendMessageBySession(registered, message);
+ } else {
+ removeSessionIndexes(registered);
+ }
+ }
}
}
@@ -282,13 +298,21 @@ public class WebsocketCollector {
if (session.isOpen()) {
sendMessageBySession(session, message);
} else {
- NAMESPACE_SESSION_MAP.getOrDefault(namespaceId,
Sets.newConcurrentHashSet()).remove(session);
- removeSessionSendQueue(session);
+ removeSessionIndexes(session);
}
}
} else {
- NAMESPACE_SESSION_MAP.getOrDefault(namespaceId,
Sets.newConcurrentHashSet())
- .forEach(session -> sendMessageBySession(session,
message));
+ Set<Session> sessions = NAMESPACE_SESSION_MAP.get(namespaceId);
+ if (Objects.isNull(sessions) || sessions.isEmpty()) {
+ return;
+ }
+ for (Session registered : new ArrayList<>(sessions)) {
+ if (registered.isOpen()) {
+ sendMessageBySession(registered, message);
+ } else {
+ removeSessionIndexes(registered);
+ }
+ }
}
}
@@ -304,14 +328,24 @@ public class WebsocketCollector {
}
}
- private void clearSession(final Session session) {
+ private static void clearSession(final Session session) {
+ removeSessionIndexes(session);
+ ThreadLocalUtils.clear();
+ }
+
+ private static void removeSessionIndexes(final Session session) {
SESSION_SET.remove(session);
removeSessionSendQueue(session);
- String namespaceId = getNamespaceId(session);
+ String namespaceId = SESSION_NAMESPACE_IDS.remove(session);
if (StringUtils.isNotBlank(namespaceId)) {
- NAMESPACE_SESSION_MAP.getOrDefault(namespaceId,
Sets.newConcurrentHashSet()).remove(session);
+ NAMESPACE_SESSION_MAP.compute(namespaceId, (id, sessions) -> {
+ if (Objects.isNull(sessions)) {
+ return null;
+ }
+ sessions.remove(session);
+ return sessions.isEmpty() ? null : sessions;
+ });
}
- ThreadLocalUtils.clear();
}
private static String maskSensitive(final String json) {
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/websocket/WebsocketCollectorTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/websocket/WebsocketCollectorTest.java
index 72e1f1ac7e..0c3cba451c 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/websocket/WebsocketCollectorTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/websocket/WebsocketCollectorTest.java
@@ -53,8 +53,10 @@ import java.util.Objects;
import java.util.Set;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
@@ -132,6 +134,11 @@ public final class WebsocketCollectorTest {
if (Objects.nonNull(sessionSendQueues)) {
sessionSendQueues.clear();
}
+ Map<Session, ?> sessionNamespaces =
+ (Map<Session, ?>)
ReflectionTestUtils.getField(WebsocketCollector.class, "SESSION_NAMESPACE_IDS");
+ if (Objects.nonNull(sessionNamespaces)) {
+ sessionNamespaces.clear();
+ }
}
@Test
@@ -249,6 +256,89 @@ public final class WebsocketCollectorTest {
assertNull(getSession());
}
+ @Test
+ void testOnCloseRemovesAlreadyClosedSessionFromNamespaceMap() {
+ websocketCollector.onOpen(session);
+ assertEquals(1,
namespaceSessionCount(Constants.SYS_DEFAULT_NAMESPACE_ID));
+ when(session.isOpen()).thenReturn(false);
+ doNothing().when(loggerSpy).warn(anyString(), anyString());
+
+ websocketCollector.onClose(session);
+
+ assertEquals(0L, getSessionSetSize());
+
assertFalse(namespaceMap().containsKey(Constants.SYS_DEFAULT_NAMESPACE_ID));
+ assertEquals(0, sendQueueSize());
+ }
+
+ @Test
+ void testOnErrorRemovesAlreadyClosedSessionFromNamespaceMap() {
+ websocketCollector.onOpen(session);
+ when(session.isOpen()).thenReturn(false);
+ doNothing().when(loggerSpy).error(anyString(), anyString(),
isA(Throwable.class));
+
+ websocketCollector.onError(session, new
IllegalStateException("closed"));
+
+ assertEquals(0L, getSessionSetSize());
+
assertFalse(namespaceMap().containsKey(Constants.SYS_DEFAULT_NAMESPACE_ID));
+ }
+
+ @Test
+ void testRepeatedCloseOfAlreadyClosedSessionIsIdempotent() {
+ websocketCollector.onOpen(session);
+ when(session.isOpen()).thenReturn(false);
+ doNothing().when(loggerSpy).warn(anyString(), anyString());
+
+ websocketCollector.onClose(session);
+ websocketCollector.onClose(session);
+ websocketCollector.onError(session, new IllegalStateException("closed
again"));
+
+ assertEquals(0L, getSessionSetSize());
+ assertTrue(namespaceMap().isEmpty());
+ }
+
+ @Test
+ void testMissingNamespaceLeavesNoPartialRegistration() {
+ Map<String, Object> userProperties = new HashMap<>();
+ when(session.getUserProperties()).thenReturn(userProperties);
+
+ assertThrows(ShenyuException.class, () ->
websocketCollector.onOpen(session));
+
+ assertEquals(0L, getSessionSetSize());
+ assertTrue(namespaceMap().isEmpty());
+ assertEquals(0, sendQueueSize());
+ }
+
+ @Test
+ void testReconnectDoesNotAccumulateClosedSessions() {
+ doNothing().when(loggerSpy).warn(anyString(), anyString());
+ for (int i = 0; i < 3; i++) {
+ Session reconnect = mock(Session.class);
+ Map<String, Object> props = new HashMap<>();
+ props.put(Constants.SHENYU_NAMESPACE_ID,
Constants.SYS_DEFAULT_NAMESPACE_ID);
+ when(reconnect.isOpen()).thenReturn(true);
+ when(reconnect.getUserProperties()).thenReturn(props);
+ websocketCollector.onOpen(reconnect);
+ when(reconnect.isOpen()).thenReturn(false);
+ websocketCollector.onClose(reconnect);
+ }
+
+ assertEquals(0L, getSessionSetSize());
+ assertTrue(namespaceMap().isEmpty());
+ }
+
+ @Test
+ void testNamespaceBroadcastSkipsClosedSession() {
+ RemoteEndpoint.Async async = mock(RemoteEndpoint.Async.class);
+ when(session.getAsyncRemote()).thenReturn(async);
+ websocketCollector.onOpen(session);
+ when(session.isOpen()).thenReturn(false);
+
+ WebsocketCollector.send(Constants.SYS_DEFAULT_NAMESPACE_ID,
"stale-msg", DataEventTypeEnum.CREATE);
+
+ verify(async, never()).sendText(anyString(), any(SendHandler.class));
+
assertFalse(namespaceMap().containsKey(Constants.SYS_DEFAULT_NAMESPACE_ID));
+ }
+
@Test
void testOnCloseWithBlankNamespaceId() {
// Session with no namespace — clearSession should still work (blank
namespaceId branch)
@@ -454,6 +544,21 @@ public final class WebsocketCollectorTest {
return Objects.isNull(sessionSet) ? -1 : sessionSet.size();
}
+ @SuppressWarnings("unchecked")
+ private Map<String, Set<Session>> namespaceMap() {
+ return (Map<String, Set<Session>>)
ReflectionTestUtils.getField(WebsocketCollector.class, "NAMESPACE_SESSION_MAP");
+ }
+
+ private int namespaceSessionCount(final String namespaceId) {
+ Set<Session> sessions = namespaceMap().get(namespaceId);
+ return Objects.isNull(sessions) ? 0 : sessions.size();
+ }
+
+ private int sendQueueSize() {
+ Map<?, ?> queues = (Map<?, ?>)
ReflectionTestUtils.getField(WebsocketCollector.class, "SESSION_SEND_QUEUES");
+ return Objects.isNull(queues) ? 0 : queues.size();
+ }
+
private Session getSession() {
return (Session) ThreadLocalUtils.get("sessionKey");
}
