feat: ReverseHttpProxy日志调整

meethigher · meethigher · commit 2c21601144b5 · 2025-06-14T17:23:53.000+08:00
diff --git a/src/main/java/top/meethigher/proxy/http/ReverseHttpProxy.java b/src/main/java/top/meethigher/proxy/http/ReverseHttpProxy.java
@@ -289,12 +289,12 @@ protected Object getRouteMetadata(Route route, String key) {
      * @param value 数据值
      */
     protected void setContextData(RoutingContext ctx, String key, Object value) {
-        ctx.put(key, value == null ? "" : value);
+        ctx.put(key, value == null ? "null" : value);
     }
 
     protected Object getContextData(RoutingContext ctx, String key) {
         Object metadata = ctx.get(key);
-        return metadata == null ? "" : metadata;
+        return metadata == null ? "null" : metadata;
     }
 
     protected HttpServerResponse setStatusCode(RoutingContext ctx, HttpServerResponse resp, int code) {
@@ -630,14 +630,14 @@ protected Handler<AsyncResult<HttpClientRequest>> connectHandler(RoutingContext
                 setContextData(ctx, INTERNAL_CLIENT_CONNECTION_OPEN, true);
 
                 // 注册客户端与代理服务之间连接的断开监听事件。可监听主动关闭和被动关闭
-                setContextData(ctx, INTERNAL_CLIENT_LOCAL_ADDR, clientReq.connection().localAddress().toString());
-                setContextData(ctx, INTERNAL_CLIENT_REMOTE_ADDR, clientReq.connection().remoteAddress().toString());
-                log.debug("{} --> {} connected", getContextData(ctx, INTERNAL_CLIENT_LOCAL_ADDR), getContextData(ctx, INTERNAL_CLIENT_REMOTE_ADDR));
-
+                HttpConnection connection = clientReq.connection();
+                setContextData(ctx, INTERNAL_CLIENT_LOCAL_ADDR, connection.localAddress().toString());
+                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));
 
-                clientReq.connection().closeHandler(v -> {
+                connection.closeHandler(v -> {
                     setContextData(ctx, INTERNAL_CLIENT_CONNECTION_OPEN, false);
-                    log.debug("{} --> {} closed", getContextData(ctx, INTERNAL_CLIENT_LOCAL_ADDR), getContextData(ctx, INTERNAL_CLIENT_REMOTE_ADDR));
+                    log.debug("target {} -- {} closed", getContextData(ctx, INTERNAL_CLIENT_LOCAL_ADDR), getContextData(ctx, INTERNAL_CLIENT_REMOTE_ADDR));
                 });
 
 
@@ -679,6 +679,15 @@ protected Handler<RoutingContext> routingContextHandler(HttpClient httpClient) {
             // 暂停流读取
             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());
+            // 记录请求开始时间
+            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应以路由为单元存储，而不是以请求为单元存储。一个路由会有很多请求。
             // 若想要以请求为单元存储数据，应该使用routingContext.put
@@ -687,11 +696,6 @@ protected Handler<RoutingContext> routingContextHandler(HttpClient httpClient) {
                 setContextData(ctx, key, ctx.currentRoute().getMetadata(key));
             }
 
-            // 记录请求开始时间
-            setContextData(ctx, INTERNAL_SEND_TIMESTAMP, System.currentTimeMillis());
-            // 记录连接状态
-            setContextData(ctx, INTERNAL_SERVER_CONNECTION_OPEN, true);
-
             // 获取代理地址
             String proxyUrl = getProxyUrl(ctx, ctx.request(), ctx.response());
             setContextData(ctx, INTERNAL_PROXY_URL, proxyUrl);
@@ -707,15 +711,9 @@ 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()));
 
-            // 注册客户端与代理服务之间连接的断开监听事件。可监听主动关闭和被动关闭
-            setContextData(ctx, INTERNAL_SERVER_REMOTE_ADDR, ctx.request().connection().remoteAddress().toString());
-            setContextData(ctx, INTERNAL_SERVER_LOCAL_ADDR, ctx.request().connection().localAddress().toString());
-
-            log.debug("{} <-- {} connected", getContextData(ctx, INTERNAL_SERVER_LOCAL_ADDR), getContextData(ctx, INTERNAL_SERVER_REMOTE_ADDR));
-
-            ctx.request().connection().closeHandler(v -> {
+            connection.closeHandler(v -> {
                 setContextData(ctx, INTERNAL_SERVER_CONNECTION_OPEN, false);
-                log.debug("{} <-- {} closed", getContextData(ctx, INTERNAL_SERVER_LOCAL_ADDR), getContextData(ctx, INTERNAL_SERVER_REMOTE_ADDR));
+                log.debug("source {} -- {} closed", getContextData(ctx, INTERNAL_SERVER_LOCAL_ADDR), getContextData(ctx, INTERNAL_SERVER_REMOTE_ADDR));
             });
 
             // 如果跨域由代理服务接管，那么针对跨域使用的OPTIONS预检请求，就由代理服务接管，而不经过实际的后端服务
diff --git a/src/test/java/top/meethigher/proxy/tcp/Pipe.java b/src/test/java/top/meethigher/proxy/tcp/Pipe.java
@@ -0,0 +1,163 @@
+/*
+ * Copyright (c) 2011-2019 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+ * which is available at https://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+ */
+package top.meethigher.proxy.tcp;
+
+import io.vertx.core.*;
+import io.vertx.core.net.NetSocket;
+import io.vertx.core.streams.ReadStream;
+import io.vertx.core.streams.WriteStream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class Pipe<T> implements io.vertx.core.streams.Pipe<T> {
+
+    private static final Logger log = LoggerFactory.getLogger(Pipe.class);
+    private final Promise<Void> result;
+    private final ReadStream<T> src;
+    private boolean endOnSuccess = true;
+    private boolean endOnFailure = true;
+    private WriteStream<T> dst;
+
+    public Pipe(ReadStream<T> src) {
+        this.src = src;
+        this.result = Promise.promise();
+
+        if (src instanceof NetSocket) {
+            NetSocket tsrc = (NetSocket) src;
+            tsrc.remoteAddress();
+            tsrc.localAddress();
+        }
+
+        // Set handlers now
+        src.endHandler(result::tryComplete);
+        src.exceptionHandler(result::tryFail);
+    }
+
+    @Override
+    public synchronized io.vertx.core.streams.Pipe<T> endOnFailure(boolean end) {
+        endOnFailure = end;
+        return this;
+    }
+
+    @Override
+    public synchronized io.vertx.core.streams.Pipe<T> endOnSuccess(boolean end) {
+        endOnSuccess = end;
+        return this;
+    }
+
+    @Override
+    public synchronized io.vertx.core.streams.Pipe<T> endOnComplete(boolean end) {
+        endOnSuccess = end;
+        endOnFailure = end;
+        return this;
+    }
+
+    private void handleWriteResult(AsyncResult<Void> ack) {
+        if (ack.failed()) {
+            result.tryFail(new WriteException(ack.cause()));
+        }
+    }
+
+    @Override
+    public void to(WriteStream<T> ws, Handler<AsyncResult<Void>> completionHandler) {
+        if (ws == null) {
+            throw new NullPointerException();
+        }
+        boolean endOnSuccess;
+        boolean endOnFailure;
+        synchronized (Pipe.this) {
+            if (dst != null) {
+                throw new IllegalStateException();
+            }
+            dst = ws;
+            endOnSuccess = this.endOnSuccess;
+            endOnFailure = this.endOnFailure;
+        }
+        Handler<Void> drainHandler = v -> src.resume();
+        src.handler(item -> {
+            if (src instanceof NetSocket) {
+                NetSocket tsrc = (NetSocket) src;
+                log.info("{} -- {} received:\n{}", tsrc.remoteAddress(), tsrc.localAddress(),
+                        item.toString());
+            }
+            ws.write(item, this::handleWriteResult);
+            if (ws.writeQueueFull()) {
+                src.pause();
+                ws.drainHandler(drainHandler);
+            }
+        });
+        src.resume();
+        result.future().onComplete(ar -> {
+            try {
+                src.handler(null);
+            } catch (Exception ignore) {
+            }
+            try {
+                src.exceptionHandler(null);
+            } catch (Exception ignore) {
+            }
+            try {
+                src.endHandler(null);
+            } catch (Exception ignore) {
+            }
+            if (ar.succeeded()) {
+                handleSuccess(completionHandler);
+            } else {
+                Throwable err = ar.cause();
+                if (err instanceof WriteException) {
+                    src.resume();
+                    err = err.getCause();
+                }
+                handleFailure(err, completionHandler);
+            }
+        });
+    }
+
+    private void handleSuccess(Handler<AsyncResult<Void>> completionHandler) {
+        if (endOnSuccess) {
+            dst.end(completionHandler);
+        } else {
+            completionHandler.handle(Future.succeededFuture());
+        }
+    }
+
+    private void handleFailure(Throwable cause, Handler<AsyncResult<Void>> completionHandler) {
+        Future<Void> res = Future.failedFuture(cause);
+        if (endOnFailure) {
+            dst.end(ignore -> {
+                completionHandler.handle(res);
+            });
+        } else {
+            completionHandler.handle(res);
+        }
+    }
+
+    public void close() {
+        synchronized (this) {
+            src.exceptionHandler(null);
+            src.handler(null);
+            if (dst != null) {
+                dst.drainHandler(null);
+                dst.exceptionHandler(null);
+            }
+        }
+        VertxException err = new VertxException("Pipe closed", true);
+        if (result.tryFail(err)) {
+            src.resume();
+        }
+    }
+
+    private static class WriteException extends VertxException {
+        private WriteException(Throwable cause) {
+            super(cause, true);
+        }
+    }
+}
