From 268e17273eeb7f645ab3c8bdda9ff91d3de2641c Mon Sep 17 00:00:00 2001
From: aoshiguchen <1052045476@qq.com>
Date: Sat, 23 Sep 2023 23:21:13 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BB=A3=E7=90=86=E4=BC=98=E5=8C=96=EF=BC=8C?=
 =?UTF-8?q?=E8=A7=A3=E5=86=B3=E8=83=8C=E5=8E=8B=E4=B8=8B=E8=BD=BD=E5=A4=A7?=
 =?UTF-8?q?=E6=96=87=E4=BB=B6=E4=B8=AD=E9=80=94=E6=96=AD=E5=BC=80=E7=9A=84?=
 =?UTF-8?q?=E9=97=AE=E9=A2=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../client/core/RealServerChannelHandler.java | 11 +++++++++
 .../client/core/TcpProxyChannelHandler.java   | 12 ++++++----
 .../handler/ProxyMessageTransferHandler.java  |  4 ++++
 .../src/main/resources/app.yml                |  6 ++---
 .../proxy/core/ProxyTunnelChannelHandler.java | 23 +++++++++++--------
 .../proxy/core/TcpVisitorChannelHandler.java  |  7 +++++-
 .../proxy/core/UdpVisitorChannelHandler.java  | 22 ++++++++++++++++++
 .../handler/ProxyMessageTransferHandler.java  | 11 +++++++++
 .../UdpProxyMessageTransferHandler.java       | 13 +++++++++++
 .../src/main/resources/app.yml                |  4 ++--
 10 files changed, 92 insertions(+), 21 deletions(-)

diff --git a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/RealServerChannelHandler.java b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/RealServerChannelHandler.java
index 69c47d45..aafec97a 100644
--- a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/RealServerChannelHandler.java
+++ b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/RealServerChannelHandler.java
@@ -49,6 +49,17 @@ public class RealServerChannelHandler extends SimpleChannelInboundHandler<ByteBu
             // 代理客户端连接断开
             ctx.channel().close();
         } else {
+
+            if (proxyChannel.isWritable()) {
+                if (!realServerChannel.config().isAutoRead()) {
+                    realServerChannel.config().setAutoRead(true);
+                }
+            } else {
+                if (realServerChannel.config().isAutoRead()) {
+                    realServerChannel.config().setAutoRead(false);
+                }
+            }
+
             byte[] bytes = new byte[buf.readableBytes()];
             buf.readBytes(bytes);
             String visitorId = ProxyUtil.getVisitorIdByRealServerChannel(realServerChannel);
diff --git a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/TcpProxyChannelHandler.java b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/TcpProxyChannelHandler.java
index 73283be3..c3710e83 100644
--- a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/TcpProxyChannelHandler.java
+++ b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/TcpProxyChannelHandler.java
@@ -63,16 +63,18 @@ public class TcpProxyChannelHandler extends SimpleChannelInboundHandler<ProxyMes
             IdleStateEvent event = (IdleStateEvent)evt;
             switch (event.state()) {
                 case READER_IDLE:
-                    // 读超时，断开连接
-                    log.info("[TCP Proxy Channel]Read timeout");
-                    ctx.channel().close();
+                    if (ctx.channel().isWritable()) {
+                        // 读超时，断开连接
+                        log.info("[TCP Proxy Channel]Read timeout");
+                        ctx.channel().close();
+                    }
                     break;
                 case WRITER_IDLE:
                     ctx.channel().writeAndFlush(ProxyMessage.buildHeartbeatMessage());
                     break;
                 case ALL_IDLE:
-                    log.debug("[TCP Proxy Channel]ReadWrite timeout");
-                    ctx.close();
+//                    log.debug("[TCP Proxy Channel]ReadWrite timeout");
+//                    ctx.close();
                     break;
             }
         }
diff --git a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/handler/ProxyMessageTransferHandler.java b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/handler/ProxyMessageTransferHandler.java
index 06a24f2b..6e8f2122 100644
--- a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/handler/ProxyMessageTransferHandler.java
+++ b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/handler/ProxyMessageTransferHandler.java
@@ -23,6 +23,10 @@ public class ProxyMessageTransferHandler implements ProxyMessageHandler {
 	public void handle(ChannelHandlerContext ctx, ProxyMessage proxyMessage) {
 		Channel realServerChannel = ctx.channel().attr(Constants.NEXT_CHANNEL).get();
 		if (realServerChannel != null) {
+
+			// 自己可写，则设置来源可读。自己不可写，则设置来源不可读
+			realServerChannel.config().setAutoRead(ctx.channel().isWritable());
+
 			ByteBuf buf = ctx.alloc().buffer(proxyMessage.getData().length);
 			buf.writeBytes(proxyMessage.getData());
 			realServerChannel.writeAndFlush(buf);
diff --git a/neutrino-proxy-client/src/main/resources/app.yml b/neutrino-proxy-client/src/main/resources/app.yml
index 3b869fb3..29e02100 100644
--- a/neutrino-proxy-client/src/main/resources/app.yml
+++ b/neutrino-proxy-client/src/main/resources/app.yml
@@ -9,9 +9,9 @@ neutrino:
       length-field-length: 4
       initial-bytes-to-strip: 0
       length-adjustment: 0
-      read-idle-time: 8
-      write-idle-time: 3
-      all-idle-time-seconds: 8
+      read-idle-time: 120
+      write-idle-time: 20
+      all-idle-time-seconds: 0
     logger:
       # 日志级别
       level: ${LOG_LEVEL:info}
diff --git a/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/ProxyTunnelChannelHandler.java b/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/ProxyTunnelChannelHandler.java
index d44cc9e2..18412de4 100644
--- a/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/ProxyTunnelChannelHandler.java
+++ b/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/ProxyTunnelChannelHandler.java
@@ -70,10 +70,10 @@ public class ProxyTunnelChannelHandler extends SimpleChannelInboundHandler<Proxy
 
     @Override
     public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
-        Channel userChannel = ctx.channel().attr(Constants.NEXT_CHANNEL).get();
-        if (userChannel != null) {
-            userChannel.config().setOption(ChannelOption.AUTO_READ, ctx.channel().isWritable());
-        }
+//        Channel userChannel = ctx.channel().attr(Constants.NEXT_CHANNEL).get();
+//        if (userChannel != null) {
+//            userChannel.config().setOption(ChannelOption.AUTO_READ, ctx.channel().isWritable());
+//        }
 
         super.channelWritabilityChanged(ctx);
     }
@@ -127,9 +127,10 @@ public class ProxyTunnelChannelHandler extends SimpleChannelInboundHandler<Proxy
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
 //        super.exceptionCaught(ctx, cause);
-        if (ctx.channel().isActive()) {
-            ctx.channel().close();
-        }
+//        if (ctx.channel().isActive()) {
+//            ctx.channel().close();
+//        }
+        log.error("[Tunnel Channel] error", cause);
     }
 
     @Override
@@ -138,9 +139,11 @@ public class ProxyTunnelChannelHandler extends SimpleChannelInboundHandler<Proxy
             IdleStateEvent event = (IdleStateEvent)evt;
             switch (event.state()) {
                 case READER_IDLE:
-                    // 读超时，断开连接
-                    log.debug("Read timeout");
-                    ctx.channel().close();
+                    if (ctx.channel().isWritable()) {
+                        // 读超时，断开连接
+                        log.warn("[Tunnel Channel]Read timeout");
+                        ctx.channel().close();
+                    }
                     break;
                 case WRITER_IDLE:
                     ctx.channel().writeAndFlush(ProxyMessage.buildHeartbeatMessage());
diff --git a/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/TcpVisitorChannelHandler.java b/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/TcpVisitorChannelHandler.java
index 8f0b2186..8f723802 100644
--- a/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/TcpVisitorChannelHandler.java
+++ b/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/TcpVisitorChannelHandler.java
@@ -43,6 +43,10 @@ public class TcpVisitorChannelHandler extends SimpleChannelInboundHandler<ByteBu
             ctx.channel().close();
             return;
         }
+
+        // 代理通道可写，则设置访问通道可读。代理通道不可写，则设置访问通道不可读
+        visitorChannel.config().setAutoRead(proxyChannel.isWritable());
+
         // 转发代理数据
         byte[] bytes = new byte[buf.readableBytes()];
         buf.readBytes(bytes);
@@ -129,7 +133,8 @@ public class TcpVisitorChannelHandler extends SimpleChannelInboundHandler<ByteBu
         if (null == cmdChannel) {
             // 该端口还没有代理客户端
             ctx.channel().close();
-        } else {
+        }
+        else {
             Channel proxyChannel = visitorChannel.attr(Constants.NEXT_CHANNEL).get();
             if (null != proxyChannel) {
                 proxyChannel.config().setOption(ChannelOption.AUTO_READ, visitorChannel.isWritable());
diff --git a/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/UdpVisitorChannelHandler.java b/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/UdpVisitorChannelHandler.java
index 07abc014..6493be65 100644
--- a/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/UdpVisitorChannelHandler.java
+++ b/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/UdpVisitorChannelHandler.java
@@ -127,4 +127,26 @@ public class UdpVisitorChannelHandler extends SimpleChannelInboundHandler<Datagr
         ctx.close();
         log.error("[UDP Visitor Channel]VisitorChannel error", cause);
     }
+
+    @Override
+    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
+
+        // 通知代理客户端
+        Channel visitorChannel = ctx.channel();
+        InetSocketAddress sa = (InetSocketAddress) visitorChannel.localAddress();
+        Channel cmdChannel = ProxyUtil.getCmdChannelByServerPort(sa.getPort());
+
+        if (null == cmdChannel) {
+            // 该端口还没有代理客户端
+            ctx.channel().close();
+        }
+        else {
+            Channel proxyChannel = visitorChannel.attr(Constants.NEXT_CHANNEL).get();
+            if (null != proxyChannel) {
+                proxyChannel.config().setOption(ChannelOption.AUTO_READ, visitorChannel.isWritable());
+            }
+        }
+
+        super.channelWritabilityChanged(ctx);
+    }
 }
diff --git a/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/handler/ProxyMessageTransferHandler.java b/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/handler/ProxyMessageTransferHandler.java
index af39b9f8..45d70b40 100644
--- a/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/handler/ProxyMessageTransferHandler.java
+++ b/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/handler/ProxyMessageTransferHandler.java
@@ -27,6 +27,17 @@ public class ProxyMessageTransferHandler implements ProxyMessageHandler {
 	public void handle(ChannelHandlerContext ctx, ProxyMessage proxyMessage) {
 		Channel visitorChannel = ctx.channel().attr(Constants.NEXT_CHANNEL).get();
 		if (null != visitorChannel) {
+			if (!visitorChannel.isWritable()) {
+				//自己不可写，通道可以读，让通道关闭读
+				//自己可写，通道不可以读，让通道打开读
+				if (ctx.channel().config().isAutoRead()) {
+					ctx.channel().config().setAutoRead(false);
+				}
+			} else {
+				if (ctx.channel().config().isAutoRead()) {
+					ctx.channel().config().setAutoRead(true);
+				}
+			}
 			ByteBuf buf = ctx.alloc().buffer(proxyMessage.getData().length);
 			buf.writeBytes(proxyMessage.getData());
 			visitorChannel.writeAndFlush(buf);
diff --git a/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/handler/UdpProxyMessageTransferHandler.java b/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/handler/UdpProxyMessageTransferHandler.java
index 978bae9d..a8fd0ba2 100644
--- a/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/handler/UdpProxyMessageTransferHandler.java
+++ b/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/handler/UdpProxyMessageTransferHandler.java
@@ -35,6 +35,19 @@ public class UdpProxyMessageTransferHandler implements ProxyMessageHandler {
 
         Channel visitorChannel = ctx.channel().attr(Constants.NEXT_CHANNEL).get();
         if (null != visitorChannel) {
+
+            if (!visitorChannel.isWritable()) {
+                //自己不可写，通道可以读，让通道关闭读
+                //自己可写，通道不可以读，让通道打开读
+                if (ctx.channel().config().isAutoRead()) {
+                    ctx.channel().config().setAutoRead(false);
+                }
+            } else {
+                if (ctx.channel().config().isAutoRead()) {
+                    ctx.channel().config().setAutoRead(true);
+                }
+            }
+
 //            InetSocketAddress address = new InetSocketAddress(udpBaseInfo.getVisitorIp(), udpBaseInfo.getVisitorPort());
 //            ByteBuf byteBuf = Unpooled.copiedBuffer(proxyMessage.getData());
 //            visitorChannel.writeAndFlush(new DatagramPacket(byteBuf, address));
diff --git a/neutrino-proxy-server/src/main/resources/app.yml b/neutrino-proxy-server/src/main/resources/app.yml
index d268da3b..a8a71f21 100644
--- a/neutrino-proxy-server/src/main/resources/app.yml
+++ b/neutrino-proxy-server/src/main/resources/app.yml
@@ -11,8 +11,8 @@ neutrino:
       length-field-length: 4
       initial-bytes-to-strip: 0
       length-adjustment: 0
-      read-idle-time: 30
-      write-idle-time: 5
+      read-idle-time: 120
+      write-idle-time: 20
       all-idle-time-seconds: 0
     logger:
       # 日志级别