Merge pull request #21 from meethigher/issue-18

meethigher · web-flow · commit 91d713a21f35 · 2025-07-27T14:38:51.000+08:00
在pipe前添加针对连接的异常处理器
diff --git a/src/main/java/top/meethigher/proxy/http/ReverseHttpProxy.java b/src/main/java/top/meethigher/proxy/http/ReverseHttpProxy.java
@@ -1,7 +1,6 @@
 package top.meethigher.proxy.http;
 
 import io.vertx.core.AsyncResult;
-import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
 import io.vertx.core.http.*;
@@ -339,18 +338,11 @@ public ReverseHttpProxy host(String host) {
     }
 
     public void start() {
-        httpServer.requestHandler(router).exceptionHandler(e -> log.error("request failed", e));
-        Future<HttpServer> listenFuture = httpServer.listen(sourcePort, sourceHost);
-
-        Handler<AsyncResult<HttpServer>> asyncResultHandler = ar -> {
-            if (ar.succeeded()) {
-                log.info("{} started on {}:{}", name, sourceHost, sourcePort);
-            } else {
-                log.error("{} start failed", name, ar.cause());
-
-            }
-        };
-        listenFuture.onComplete(asyncResultHandler);
+        httpServer.requestHandler(router)
+                .exceptionHandler(e -> log.error("{} socket errors happening before the HTTP connection", name, e))
+                .listen(sourcePort, sourceHost)
+                .onFailure(e -> log.error("{} start failed", name, e))
+                .onSuccess(v -> log.info("{} started on {}:{}", name, sourceHost, sourcePort));
     }
 
     public void stop() {
@@ -635,10 +627,14 @@ protected Handler<AsyncResult<HttpClientRequest>> connectHandler(RoutingContext
                 setContextData(ctx, INTERNAL_CLIENT_REMOTE_ADDR, connection.remoteAddress().toString());
                 log.debug("target {} -- {} connected", getContextData(ctx, INTERNAL_CLIENT_LOCAL_ADDR), getContextData(ctx, INTERNAL_CLIENT_REMOTE_ADDR));
 
-                connection.closeHandler(v -> {
-                    setContextData(ctx, INTERNAL_CLIENT_CONNECTION_OPEN, false);
-                    log.debug("target {} -- {} closed", getContextData(ctx, INTERNAL_CLIENT_LOCAL_ADDR), getContextData(ctx, INTERNAL_CLIENT_REMOTE_ADDR));
-                });
+                // 由于内部都是使用pipe来进行数据传输，所以exceptionHandler肯定是都重新注册过了，参考{@code io.vertx.core.streams.impl.PipeImpl.PipeImpl }
+                // 但如果还没进入pipe前，连接出现异常，那么就会触发此处的exceptionHandler。https://github.com/meethigher/tcp-reverse-proxy/issues/18
+                connection.exceptionHandler(e ->
+                                log.error("target {} -- {} exception occurred", getContextData(ctx, INTERNAL_CLIENT_LOCAL_ADDR), getContextData(ctx, INTERNAL_CLIENT_REMOTE_ADDR), e))
+                        .closeHandler(v -> {
+                            setContextData(ctx, INTERNAL_CLIENT_CONNECTION_OPEN, false);
+                            log.debug("target {} -- {} closed", getContextData(ctx, INTERNAL_CLIENT_LOCAL_ADDR), getContextData(ctx, INTERNAL_CLIENT_REMOTE_ADDR));
+                        });
 
 
                 // 复制请求头。复制的过程中忽略逐跳标头
@@ -675,15 +671,21 @@ protected Handler<RoutingContext> routingContextHandler(HttpClient httpClient) {
         return ctx -> {
             // 暂停流读取
             ctx.request().pause();
-
             HttpConnection connection = ctx.request().connection();
             setContextData(ctx, INTERNAL_SERVER_REMOTE_ADDR, connection.remoteAddress().toString());
             setContextData(ctx, INTERNAL_SERVER_LOCAL_ADDR, connection.localAddress().toString());
+            log.debug("source {} -- {} connected", getContextData(ctx, INTERNAL_SERVER_LOCAL_ADDR), getContextData(ctx, INTERNAL_SERVER_REMOTE_ADDR));
+            // 由于内部都是使用pipe来进行数据传输，所以exceptionHandler肯定是都重新注册过了，参考{@code io.vertx.core.streams.impl.PipeImpl.PipeImpl }
+            // 但如果还没进入pipe前，连接出现异常，那么就会触发此处的exceptionHandler。https://github.com/meethigher/tcp-reverse-proxy/issues/18
+            connection.exceptionHandler(e -> log.error("source {} -- {} exception occurred", getContextData(ctx, INTERNAL_SERVER_LOCAL_ADDR), getContextData(ctx, INTERNAL_SERVER_REMOTE_ADDR), e))
+                    .closeHandler(v -> {
+                        setContextData(ctx, INTERNAL_SERVER_CONNECTION_OPEN, false);
+                        log.debug("source {} -- {} closed", getContextData(ctx, INTERNAL_SERVER_LOCAL_ADDR), getContextData(ctx, INTERNAL_SERVER_REMOTE_ADDR));
+                    });
             // 记录请求开始时间
             setContextData(ctx, INTERNAL_SEND_TIMESTAMP, System.currentTimeMillis());
             // 记录连接状态
             setContextData(ctx, INTERNAL_SERVER_CONNECTION_OPEN, true);
-            log.debug("source {} -- {} connected", getContextData(ctx, INTERNAL_SERVER_LOCAL_ADDR), getContextData(ctx, INTERNAL_SERVER_REMOTE_ADDR));
 
             // vertx的uri()是包含query参数的。而path()才是我们常说的不带有query的uri
             // route不是线程安全的。route里的metadata应以路由为单元存储，而不是以请求为单元存储。一个路由会有很多请求。
@@ -708,10 +710,6 @@ protected Handler<RoutingContext> routingContextHandler(HttpClient httpClient) {
             requestOptions.setMethod(ctx.request().method());
             requestOptions.setFollowRedirects(getContextData(ctx, P_FOLLOW_REDIRECTS) != null && Boolean.parseBoolean(getContextData(ctx, P_FOLLOW_REDIRECTS).toString()));
 
-            connection.closeHandler(v -> {
-                setContextData(ctx, INTERNAL_SERVER_CONNECTION_OPEN, false);
-                log.debug("source {} -- {} closed", getContextData(ctx, INTERNAL_SERVER_LOCAL_ADDR), getContextData(ctx, INTERNAL_SERVER_REMOTE_ADDR));
-            });
 
             // 如果跨域由代理服务接管，那么针对跨域使用的OPTIONS预检请求，就由代理服务接管，而不经过实际的后端服务
             if (HttpMethod.OPTIONS.name().equalsIgnoreCase(ctx.request().method().name()) &&
diff --git a/src/main/java/top/meethigher/proxy/tcp/ReverseTcpProxy.java b/src/main/java/top/meethigher/proxy/tcp/ReverseTcpProxy.java
@@ -1,7 +1,5 @@
 package top.meethigher.proxy.tcp;
 
-import io.vertx.core.AsyncResult;
-import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
 import io.vertx.core.net.NetClient;
@@ -54,7 +52,10 @@ protected ReverseTcpProxy(NetServer netServer, NetClient netClient,
             sourceSocket.pause();
             SocketAddress sourceRemote = sourceSocket.remoteAddress();
             SocketAddress sourceLocal = sourceSocket.localAddress();
-            sourceSocket.closeHandler(v -> log.debug("source {} -- {} closed", sourceLocal, sourceRemote));
+            // 由于内部都是使用pipe来进行数据传输，所以exceptionHandler肯定是都重新注册过了，参考{@code io.vertx.core.streams.impl.PipeImpl.PipeImpl }
+            // 但如果还没进入pipe前，连接出现异常，那么就会触发此处的exceptionHandler。https://github.com/meethigher/tcp-reverse-proxy/issues/18
+            sourceSocket.exceptionHandler(e -> log.error("source {} -- {}  exception occurred", sourceLocal, sourceRemote, e))
+                    .closeHandler(v -> log.debug("source {} -- {} closed", sourceLocal, sourceRemote));
             NetAddress next = lb.next();
             String targetHost = next.getHost();
             int targetPort = next.getPort();
@@ -64,7 +65,7 @@ protected ReverseTcpProxy(NetServer netServer, NetClient netClient,
             );
             netClient.connect(targetPort, targetHost)
                     .onFailure(e -> {
-                        log.error("failed to connect to {}:{}", targetHost, targetPort, e);
+                        log.error("source {} -- {} failed to connect to {}:{}", sourceLocal, sourceRemote, targetHost, targetPort, e);
                         // 若连接目标服务失败，需要断开源头服务
                         sourceSocket.close();
                     })
@@ -76,7 +77,10 @@ protected ReverseTcpProxy(NetServer netServer, NetClient netClient,
 
                         // feat: v1.0.5以前的版本，在closeHandler里面，将对端连接也关闭。比如targetSocket关闭时，则将sourceSocket也关闭。
                         // 结果导致在转发短连接时，出现了bug。参考https://github.com/meethigher/tcp-reverse-proxy/issues/6
-                        targetSocket.closeHandler(v -> log.debug("target {} -- {} closed", targetLocal, targetRemote));
+                        // 由于内部都是使用pipe来进行数据传输，所以exceptionHandler肯定是都重新注册过了，参考{@code io.vertx.core.streams.impl.PipeImpl.PipeImpl }
+                        // 但如果还没进入pipe前，连接出现异常，那么就会触发此处的exceptionHandler。https://github.com/meethigher/tcp-reverse-proxy/issues/18
+                        targetSocket.exceptionHandler(e -> log.error("target {} -- {}  exception occurred", targetLocal, targetRemote, e))
+                                .closeHandler(v -> log.debug("target {} -- {} closed", targetLocal, targetRemote));
 
                         // https://github.com/meethigher/tcp-reverse-proxy/issues/12
                         // 将日志记录详细，便于排查问题
@@ -193,18 +197,11 @@ public void start() {
         if (netAddresses.size() <= 0) {
             throw new IllegalStateException("netAddresses size must be greater than 0");
         }
-        netServer.connectHandler(connectHandler).exceptionHandler(e -> log.error("connect failed", e));
-        Future<NetServer> listenFuture = netServer.listen(sourcePort, sourceHost);
-
-        Handler<AsyncResult<NetServer>> asyncResultHandler = ar -> {
-            if (ar.succeeded()) {
-                log.info("{} started on {}:{}\nLB-Mode: {}\n  {}", name, sourceHost, sourcePort, lb.name(), netAddresses);
-            } else {
-                Throwable e = ar.cause();
-                log.error("{} start failed", name, e);
-            }
-        };
-        listenFuture.onComplete(asyncResultHandler);
+        netServer.connectHandler(connectHandler)
+                .exceptionHandler(e -> log.error("{} socket errors happening before the connection is passed to the connectHandler", name, e))
+                .listen(sourcePort, sourceHost)
+                .onFailure(e -> log.error("{} start failed", name, e))
+                .onSuccess(v -> log.info("{} started on {}:{}\nLB-Mode: {}\n  {}", name, sourceHost, sourcePort, lb.name(), netAddresses));
     }
 
     public void stop() {
diff --git a/src/main/java/top/meethigher/proxy/tcp/tunnel/ReverseTcpProxyTunnelClient.java b/src/main/java/top/meethigher/proxy/tcp/tunnel/ReverseTcpProxyTunnelClient.java
@@ -247,7 +247,10 @@ protected boolean doHandle(Vertx vertx, NetSocket netSocket, TunnelMessageType t
                                             // 双向生命周期绑定、双向数据转发
                                             // feat: v1.0.5以前的版本，在closeHandler里面，将对端连接也关闭。比如targetSocket关闭时，则将sourceSocket也关闭。
                                             // 结果导致在转发短连接时，出现了bug。参考https://github.com/meethigher/tcp-reverse-proxy/issues/6
-                                            dataSocket.closeHandler(v -> log.debug("{}: sessionId {}, data connection {} -- {} closed", dataProxyName, sessionId, dataSocket.remoteAddress(), dataSocket.localAddress()))
+                                            // 由于内部都是使用pipe来进行数据传输，所以exceptionHandler肯定是都重新注册过了，参考{@code io.vertx.core.streams.impl.PipeImpl.PipeImpl }
+                                            // 但如果还没进入pipe前，连接出现异常，那么就会触发此处的exceptionHandler。https://github.com/meethigher/tcp-reverse-proxy/issues/18
+                                            dataSocket.exceptionHandler(e -> log.error("{}: sessionId {}, data connection {} -- {} exception occurred", dataProxyName, sessionId, dataSocket.remoteAddress(), dataSocket.localAddress(), e))
+                                                    .closeHandler(v -> log.debug("{}: sessionId {}, data connection {} -- {} closed", dataProxyName, sessionId, dataSocket.remoteAddress(), dataSocket.localAddress()))
                                                     .pipeTo(backendSocket)
                                                     .onFailure(e -> log.error("{}: sessionId {}, data connection {} -- {} pipe to backend connection {} -- {} failed",
                                                             dataProxyName,
@@ -260,7 +263,10 @@ protected boolean doHandle(Vertx vertx, NetSocket netSocket, TunnelMessageType t
                                                             sessionId,
                                                             dataSocket.remoteAddress(), dataSocket.localAddress(),
                                                             backendSocket.remoteAddress(), backendSocket.localAddress()));
-                                            backendSocket.closeHandler(v -> log.debug("{}: sessionId {}, backend connection {} -- {} closed", dataProxyName, sessionId, backendSocket.remoteAddress(), backendSocket.localAddress()))
+                                            // 由于内部都是使用pipe来进行数据传输，所以exceptionHandler肯定是都重新注册过了，参考{@code io.vertx.core.streams.impl.PipeImpl.PipeImpl }
+                                            // 但如果还没进入pipe前，连接出现异常，那么就会触发此处的exceptionHandler。https://github.com/meethigher/tcp-reverse-proxy/issues/18
+                                            backendSocket.exceptionHandler(e -> log.error("{}: sessionId {}, backend connection {} -- {} exception occurred", dataProxyName, sessionId, backendSocket.remoteAddress(), backendSocket.localAddress(), e))
+                                                    .closeHandler(v -> log.debug("{}: sessionId {}, backend connection {} -- {} closed", dataProxyName, sessionId, backendSocket.remoteAddress(), backendSocket.localAddress()))
                                                     .pipeTo(dataSocket)
                                                     .onFailure(e -> log.error("{}: sessionId {}, backend connection {} -- {} pipe to data connection {} -- {} failed",
                                                             dataProxyName,
diff --git a/src/main/java/top/meethigher/proxy/tcp/tunnel/ReverseTcpProxyTunnelServer.java b/src/main/java/top/meethigher/proxy/tcp/tunnel/ReverseTcpProxyTunnelServer.java
@@ -1,8 +1,6 @@
 package top.meethigher.proxy.tcp.tunnel;
 
 
-import io.vertx.core.AsyncResult;
-import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.net.NetServer;
@@ -145,16 +143,11 @@ public static String generateName() {
     }
 
     public void start() {
-        Handler<AsyncResult<NetServer>> asyncResultHandler = ar -> {
-            if (ar.succeeded()) {
-                log.info("{} started on {}:{}", name, host, port);
-            } else {
-                Throwable e = ar.cause();
-                log.error("{} start failed", name, e);
-            }
-        };
-        netServer.connectHandler(this::handleConnect).exceptionHandler(e -> log.error("connect failed", e));
-        netServer.listen(port, host).onComplete(asyncResultHandler);
+        netServer.connectHandler(this::handleConnect)
+                .exceptionHandler(e -> log.error("{}: socket errors happening before the connection is passed to the connectHandler", name, e))
+                .listen(port, host)
+                .onFailure(e -> log.error("{} start failed", name, e))
+                .onSuccess(v -> log.info("{} started on {}:{}", name, host, port));
     }
 
     public void stop() {
@@ -301,7 +294,10 @@ protected void bindConnections(UserConnection userConn, NetSocket dataSocket, in
             // 双向生命周期绑定、双向数据转发
             // feat: v1.0.5以前的版本，在closeHandler里面，将对端连接也关闭。比如targetSocket关闭时，则将sourceSocket也关闭。
             // 结果导致在转发短连接时，出现了bug。参考https://github.com/meethigher/tcp-reverse-proxy/issues/6
-            userSocket.closeHandler(v -> log.debug("{}: sessionId {}, user connection {} -- {} closed", name, sessionId, userSocket.remoteAddress(), userSocket.localAddress()))
+            // 由于内部都是使用pipe来进行数据传输，所以exceptionHandler肯定是都重新注册过了，参考{@code io.vertx.core.streams.impl.PipeImpl.PipeImpl }
+            // 但如果还没进入pipe前，连接出现异常，那么就会触发此处的exceptionHandler。https://github.com/meethigher/tcp-reverse-proxy/issues/18
+            userSocket.exceptionHandler(e -> log.error("{}: sessionId {}, user connection {} -- {} exception occurred", name, sessionId, userSocket.remoteAddress(), userSocket.localAddress(), e))
+                    .closeHandler(v -> log.debug("{}: sessionId {}, user connection {} -- {} closed", name, sessionId, userSocket.remoteAddress(), userSocket.localAddress()))
                     .pipeTo(dataSocket)
                     .onFailure(e -> log.error("{}: sessionId {}, user connection {} -- {} pipe to data connection {} -- {} failed",
                             name,
@@ -311,10 +307,10 @@ protected void bindConnections(UserConnection userConn, NetSocket dataSocket, in
                             name,
                             sessionId,
                             userSocket.remoteAddress(), userSocket.localAddress(), dataSocket.remoteAddress(), dataSocket.localAddress()));
-            dataSocket.closeHandler(v -> log.debug("{}: sessionId {}, data connection {} -- {} closed",
-                            name,
-                            sessionId,
-                            dataSocket.remoteAddress(), dataSocket.localAddress()))
+            // 由于内部都是使用pipe来进行数据传输，所以exceptionHandler肯定是都重新注册过了，参考{@code io.vertx.core.streams.impl.PipeImpl.PipeImpl }
+            // 但如果还没进入pipe前，连接出现异常，那么就会触发此处的exceptionHandler。https://github.com/meethigher/tcp-reverse-proxy/issues/18
+            dataSocket.exceptionHandler(e -> log.error("{}: sessionId {}, data connection {} -- {} exception occurred", name, sessionId, dataSocket.remoteAddress(), dataSocket.localAddress(), e))
+                    .closeHandler(v -> log.debug("{}: sessionId {}, data connection {} -- {} closed", name, sessionId, dataSocket.remoteAddress(), dataSocket.localAddress()))
                     .pipeTo(userSocket)
                     .onFailure(e -> log.error("{}: sessionId {}, data connection {} -- {} pipe to user connection {} -- {} failed",
                             name,
@@ -347,6 +343,7 @@ protected void bindConnections(UserConnection userConn, NetSocket dataSocket, in
         public void start() {
             this.netServer
                     .connectHandler(this::handleConnect)
+                    .exceptionHandler(e -> log.error("{} socket errors happening before the connection is passed to the connectHandler", name, e))
                     .listen(port, host)
                     .onComplete(ar -> {
                         if (ar.succeeded()) {
@@ -369,6 +366,7 @@ public boolean startSync() {
             AtomicBoolean success = new AtomicBoolean(false);
             this.netServer
                     .connectHandler(this::handleConnect)
+                    .exceptionHandler(e -> log.error("{} socket errors happening before the connection is passed to the connectHandler", name, e))
                     .listen(port, host)
                     .onComplete(ar -> {
                         if (ar.succeeded()) {
diff --git a/src/test/java/top/meethigher/proxy/tcp/Issue18Test.java b/src/test/java/top/meethigher/proxy/tcp/Issue18Test.java
diff --git a/src/test/java/top/meethigher/proxy/tcp/issue18/Issue18Client.java b/src/test/java/top/meethigher/proxy/tcp/issue18/Issue18Client.java
