feat: 代码封装上的微调

Author: meethigher

src/main/java/top/meethigher/proxy/tcp/tunnel/ReverseTcpProxyTunnelServer.java

@@ -14,15 +14,13 @@
 import top.meethigher.proxy.tcp.tunnel.handler.TunnelHandler;
 import top.meethigher.proxy.tcp.tunnel.proto.TunnelMessage;
 
-import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
- *
- *
  * <p>背景：</p><p>我近期买了个树莓派，但是又不想随身带着树莓派，因此希望可以公网访问。</p>
  * <p>
  * 但是使用<a href="https://github.com/fatedier/frp">fatedier/frp</a>的过程中，不管在Windows还是Linux，都被扫出病毒了。
@@ -42,35 +40,35 @@ public class ReverseTcpProxyTunnelServer extends TunnelServer {
 
     protected String host = "0.0.0.0";
     protected int port = 44444;
-    protected Map<NetSocket, DataProxyServer> authedSockets = new LinkedHashMap<>();// 授权成功的socket列表
+    protected Map<NetSocket, DataProxyServer> authedSockets = new ConcurrentHashMap<>();// 授权成功的socket列表
 
     protected final String secret;
     protected final String name;
-    protected final Handler<NetSocket> connectHandler;
 
     public ReverseTcpProxyTunnelServer(Vertx vertx, NetServer netServer, String secret, String name) {
         super(vertx, netServer);
         this.secret = secret;
         this.name = name;
-        this.connectHandler = socket -> {
-            socket.pause();
-            socket.handler(decode(socket));
-            socket.closeHandler(v -> {
-                log.debug("closed {} -- {}", socket.remoteAddress(), socket.localAddress());
-                DataProxyServer removed = authedSockets.remove(socket);
-                if (removed != null) {
-                    removed.stop();
-                }
-            });
-            TunnelHandler connectedHandler = tunnelHandlers.get(null);
-            if (connectedHandler != null) {
-                connectedHandler.handle(vertx, socket, Buffer.buffer());
-            }
-            socket.resume();
-        };
         addMessageHandler();
     }
 
+    protected void handleConnect(NetSocket socket) {
+        socket.pause();
+        socket.handler(decode(socket));
+        socket.closeHandler(v -> {
+            log.debug("closed {} -- {}", socket.remoteAddress(), socket.localAddress());
+            DataProxyServer removed = authedSockets.remove(socket);
+            if (removed != null) {
+                removed.stop();
+            }
+        });
+        TunnelHandler connectedHandler = tunnelHandlers.get(null);
+        if (connectedHandler != null) {
+            connectedHandler.handle(vertx, socket, Buffer.buffer());
+        }
+        socket.resume();
+    }
+
     /**
      * 注册内网穿透的监听逻辑
      */
@@ -180,7 +178,7 @@ public void start() {
                 log.error("{} start failed", name, e);
             }
         };
-        netServer.connectHandler(connectHandler).exceptionHandler(e -> log.error("connect failed", e));
+        netServer.connectHandler(this::handleConnect).exceptionHandler(e -> log.error("connect failed", e));
         netServer.listen(port, host).onComplete(asyncResultHandler);
     }
 

src/main/java/top/meethigher/proxy/tcp/tunnel/TunnelClient.java

@@ -1,7 +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.net.NetClient;
 import io.vertx.core.net.NetSocket;
@@ -81,42 +80,7 @@ public TunnelMessageParser decode(final NetSocket socket) {
      */
     public void connect(final String host, final int port) {
         log.debug("client connect {}:{} ...", host, port);
-        Handler<AsyncResult<NetSocket>> asyncResultHandler = ar -> {
-            if (ar.succeeded()) {
-                setReconnectDelay(this.minDelay);
-
-                NetSocket socket = ar.result();
-                this.netSocket = socket;
-                socket.pause();
-                socket.closeHandler(v -> {
-                    log.warn("closed {} -- {}, after {} ms will reconnect",
-                            socket.localAddress(),
-                            socket.remoteAddress(),
-                            reconnectDelay);
-                    this.netSocket = null;
-                    reconnect(host, port);
-                });
-                socket.handler(decode(socket));
-                log.debug("client connected {}:{}", host, port);
-
-                // 执行连接成功的Handler
-                TunnelHandler tunnelHandler = tunnelHandlers.get(null);
-                if (tunnelHandler != null) {
-                    tunnelHandler.handle(vertx, socket, null);
-                }
-                socket.resume();
-            } else {
-                Throwable e = ar.cause();
-                log.error("client connect {}:{} error, after {} ms will reconnect",
-                        host,
-                        port,
-                        reconnectDelay,
-                        e);
-                this.netSocket = null;
-                reconnect(host, port);
-            }
-        };
-        netClient.connect(port, host).onComplete(asyncResultHandler);
+        netClient.connect(port, host).onComplete(ar -> handleConnectCompleteAsyncResult(ar, host, port));
     }
 
     /**
@@ -133,6 +97,50 @@ public void emit(final TunnelMessageType type, final byte[] bodyBytes) {
         }
     }
 
+    /**
+     * client完成连接后的业务逻辑
+     *
+     * @param ar   完成结果
+     * @param host 连接主机地址
+     * @param port 连接端口
+     */
+    protected void handleConnectCompleteAsyncResult(final AsyncResult<NetSocket> ar,
+                                                    final String host, final int port) {
+        if (ar.succeeded()) {
+            setReconnectDelay(this.minDelay);
+
+            NetSocket socket = ar.result();
+            this.netSocket = socket;
+            socket.pause();
+            socket.closeHandler(v -> {
+                log.warn("closed {} -- {}, after {} ms will reconnect",
+                        socket.localAddress(),
+                        socket.remoteAddress(),
+                        reconnectDelay);
+                this.netSocket = null;
+                reconnect(host, port);
+            });
+            socket.handler(decode(socket));
+            log.debug("client connected {}:{}", host, port);
+
+            // 执行连接成功的Handler
+            TunnelHandler tunnelHandler = tunnelHandlers.get(null);
+            if (tunnelHandler != null) {
+                tunnelHandler.handle(vertx, socket, null);
+            }
+            socket.resume();
+        } else {
+            Throwable e = ar.cause();
+            log.error("client connect {}:{} error, after {} ms will reconnect",
+                    host,
+                    port,
+                    reconnectDelay,
+                    e);
+            this.netSocket = null;
+            reconnect(host, port);
+        }
+    }
+
     protected void setReconnectDelay(final long delay) {
         this.reconnectDelay = delay;
     }