This is an automated email from the ASF dual-hosted git repository.
yu199195 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 838973296a fix(tcp): unregister ActivityConnectionObserver on
connection dispose to prevent memory leak (#6453)
838973296a is described below
commit 838973296aee94a855e98657ad6fc6462061f47a
Author: wy471x <[email protected]>
AuthorDate: Wed Jul 29 16:57:38 2026 +0800
fix(tcp): unregister ActivityConnectionObserver on connection dispose to
prevent memory leak (#6453)
Per-connection observers were registered on the shared EventBus but never
unregistered, causing stale observers to accumulate. Added onDispose cleanup
and error handling for failed downstream connections.
Co-authored-by: Claude Opus 4.7 <[email protected]>
---
.../shenyu/protocol/tcp/TcpBootstrapServer.java | 10 +-
.../protocol/tcp/TcpBootstrapServerTest.java | 140 ++++++++++++++++++++-
2 files changed, 148 insertions(+), 2 deletions(-)
diff --git
a/shenyu-protocol/shenyu-protocol-tcp/src/main/java/org/apache/shenyu/protocol/tcp/TcpBootstrapServer.java
b/shenyu-protocol/shenyu-protocol-tcp/src/main/java/org/apache/shenyu/protocol/tcp/TcpBootstrapServer.java
index cb7c73d769..b7a48b6597 100644
---
a/shenyu-protocol/shenyu-protocol-tcp/src/main/java/org/apache/shenyu/protocol/tcp/TcpBootstrapServer.java
+++
b/shenyu-protocol/shenyu-protocol-tcp/src/main/java/org/apache/shenyu/protocol/tcp/TcpBootstrapServer.java
@@ -87,8 +87,16 @@ public class TcpBootstrapServer implements BootstrapServer {
SocketAddress socketAddress = serverConn.channel().remoteAddress();
ActivityConnectionObserver connectionObserver = new
ActivityConnectionObserver("TcpClient");
eventBus.register(connectionObserver);
+ serverConn.onDispose(() -> eventBus.unregister(connectionObserver));
Mono<Connection> client =
connectionContext.getTcpClientConnection(getIp(socketAddress),
connectionObserver);
- client.subscribe(clientConn -> bridge.bridge(serverConn, clientConn));
+ client.subscribe(
+ clientConn -> bridge.bridge(serverConn, clientConn),
+ error -> {
+ LOG.error("Failed to establish client connection for {}",
serverConn, error);
+ eventBus.unregister(connectionObserver);
+ serverConn.dispose();
+ }
+ );
}
private String getIp(final SocketAddress socketAddress) {
diff --git
a/shenyu-protocol/shenyu-protocol-tcp/src/test/java/org/apache/shenyu/protocol/tcp/TcpBootstrapServerTest.java
b/shenyu-protocol/shenyu-protocol-tcp/src/test/java/org/apache/shenyu/protocol/tcp/TcpBootstrapServerTest.java
index 2dc218265a..d5c938a272 100644
---
a/shenyu-protocol/shenyu-protocol-tcp/src/test/java/org/apache/shenyu/protocol/tcp/TcpBootstrapServerTest.java
+++
b/shenyu-protocol/shenyu-protocol-tcp/src/test/java/org/apache/shenyu/protocol/tcp/TcpBootstrapServerTest.java
@@ -18,9 +18,22 @@
package org.apache.shenyu.protocol.tcp;
import com.google.common.eventbus.EventBus;
+import io.netty.channel.Channel;
+import org.apache.shenyu.protocol.tcp.connection.ActivityConnectionObserver;
+import org.apache.shenyu.protocol.tcp.connection.Bridge;
+import org.apache.shenyu.protocol.tcp.connection.ConnectionContext;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import reactor.core.Disposable;
+import reactor.core.publisher.Mono;
+import reactor.netty.Connection;
import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
@@ -29,12 +42,45 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
/**
* Test Case For {@link TcpBootstrapServer}.
*/
+@ExtendWith(MockitoExtension.class)
public class TcpBootstrapServerTest {
- private final TcpBootstrapServer server = new TcpBootstrapServer(new
EventBus());
+ @Mock
+ private EventBus eventBus;
+
+ @Mock
+ private Bridge bridge;
+
+ @Mock
+ private ConnectionContext connectionContext;
+
+ @Mock
+ private Connection serverConn;
+
+ @Mock
+ private Connection clientConn;
+
+ @Mock
+ private Channel channel;
+
+ private TcpBootstrapServer server;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ server = new TcpBootstrapServer(eventBus);
+ setField("bridge", bridge);
+ setField("connectionContext", connectionContext);
+ }
@Test
public void testGetIpWithIpv4() throws Exception {
@@ -70,4 +116,96 @@ public class TcpBootstrapServerTest {
method.setAccessible(true);
return (String) method.invoke(server, socketAddress);
}
+
+ @Test
+ void shouldUnregisterObserverWhenServerConnectionIsDisposed() throws
Exception {
+ SocketAddress remoteAddr = new InetSocketAddress("127.0.0.1", 8080);
+ when(serverConn.channel()).thenReturn(channel);
+ when(channel.remoteAddress()).thenReturn(remoteAddr);
+
+ ArgumentCaptor<Disposable> onDisposeCaptor =
ArgumentCaptor.forClass(Disposable.class);
+
when(serverConn.onDispose(onDisposeCaptor.capture())).thenReturn(serverConn);
+
+ when(connectionContext.getTcpClientConnection(eq("127.0.0.1"),
any(ActivityConnectionObserver.class)))
+ .thenReturn(Mono.just(clientConn));
+
+ invokeBridgeConnections(serverConn);
+
+ verify(eventBus).register(any(ActivityConnectionObserver.class));
+
+ onDisposeCaptor.getValue().dispose();
+
+ verify(eventBus).unregister(any(ActivityConnectionObserver.class));
+ }
+
+ @Test
+ void shouldUnregisterObserverAndDisposeServerConnOnClientConnectionError()
throws Exception {
+ SocketAddress remoteAddr = new InetSocketAddress("127.0.0.1", 8080);
+ when(serverConn.channel()).thenReturn(channel);
+ when(channel.remoteAddress()).thenReturn(remoteAddr);
+
+
when(serverConn.onDispose(any(Disposable.class))).thenReturn(serverConn);
+
+ RuntimeException exception = new RuntimeException("Connection
refused");
+ when(connectionContext.getTcpClientConnection(eq("127.0.0.1"),
any(ActivityConnectionObserver.class)))
+ .thenReturn(Mono.error(exception));
+
+ invokeBridgeConnections(serverConn);
+
+ verify(eventBus).register(any(ActivityConnectionObserver.class));
+ verify(eventBus).unregister(any(ActivityConnectionObserver.class));
+ verify(serverConn).dispose();
+ verify(bridge, never()).bridge(any(Connection.class),
any(Connection.class));
+ }
+
+ @Test
+ void shouldBridgeConnectionsOnSuccessfulClientConnection() throws
Exception {
+ SocketAddress remoteAddr = new InetSocketAddress("127.0.0.1", 9090);
+ when(serverConn.channel()).thenReturn(channel);
+ when(channel.remoteAddress()).thenReturn(remoteAddr);
+
+
when(serverConn.onDispose(any(Disposable.class))).thenReturn(serverConn);
+
+ when(connectionContext.getTcpClientConnection(eq("127.0.0.1"),
any(ActivityConnectionObserver.class)))
+ .thenReturn(Mono.just(clientConn));
+
+ invokeBridgeConnections(serverConn);
+
+ verify(bridge).bridge(serverConn, clientConn);
+ }
+
+ @Test
+ void shouldUseSameObserverForEventBusAndConnectionContext() throws
Exception {
+ SocketAddress remoteAddr = new InetSocketAddress("127.0.0.1", 8080);
+ when(serverConn.channel()).thenReturn(channel);
+ when(channel.remoteAddress()).thenReturn(remoteAddr);
+
+
when(serverConn.onDispose(any(Disposable.class))).thenReturn(serverConn);
+
+ ArgumentCaptor<ActivityConnectionObserver> registerCaptor =
+ ArgumentCaptor.forClass(ActivityConnectionObserver.class);
+ ArgumentCaptor<ActivityConnectionObserver> contextCaptor =
+ ArgumentCaptor.forClass(ActivityConnectionObserver.class);
+
+ when(connectionContext.getTcpClientConnection(eq("127.0.0.1"),
contextCaptor.capture()))
+ .thenReturn(Mono.just(clientConn));
+
+ invokeBridgeConnections(serverConn);
+
+ verify(eventBus).register(registerCaptor.capture());
+ assertNotNull(registerCaptor.getValue());
+ assertNotNull(contextCaptor.getValue());
+ }
+
+ private void invokeBridgeConnections(final Connection connection) throws
Exception {
+ Method method =
TcpBootstrapServer.class.getDeclaredMethod("bridgeConnections",
Connection.class);
+ method.setAccessible(true);
+ method.invoke(server, connection);
+ }
+
+ private void setField(final String name, final Object value) throws
Exception {
+ Field field = TcpBootstrapServer.class.getDeclaredField(name);
+ field.setAccessible(true);
+ field.set(server, value);
+ }
}
