From 688089832118652e141aa7748e622908161e5d9c Mon Sep 17 00:00:00 2001 From: aoshiguchen <1052045476@qq.com> Date: Tue, 19 Sep 2023 13:16:43 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=A2=E6=88=B7=E7=AB=AF=E6=96=B0=E5=A2=9EUD?= =?UTF-8?q?P=E9=9A=A7=E9=81=93=E6=B6=88=E6=81=AF=E5=A4=84=E7=90=86?= =?UTF-8?q?=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../client/config/ProxyConfiguration.java | 30 ++++++- .../client/core/CmdChannelHandler.java | 10 +-- ...ndler.java => TcpProxyChannelHandler.java} | 12 +-- .../client/core/UdpProxyChannelHandler.java | 80 +++++++++++++++++++ .../handler/ProxyMessageConnectHandler.java | 2 +- .../neutrinoproxy/client/util/ProxyUtil.java | 4 +- .../neutrinoproxy/core/ProxyMessage.java | 2 +- 7 files changed, 121 insertions(+), 19 deletions(-) rename neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/{ProxyChannelHandler.java => TcpProxyChannelHandler.java} (83%) create mode 100644 neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/UdpProxyChannelHandler.java diff --git a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/config/ProxyConfiguration.java b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/config/ProxyConfiguration.java index 7dc39597..07ca1890 100644 --- a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/config/ProxyConfiguration.java +++ b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/config/ProxyConfiguration.java @@ -7,8 +7,9 @@ import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.logging.LoggingHandler; import io.netty.handler.timeout.IdleStateHandler; import org.dromara.neutrinoproxy.client.core.CmdChannelHandler; -import org.dromara.neutrinoproxy.client.core.ProxyChannelHandler; +import org.dromara.neutrinoproxy.client.core.TcpProxyChannelHandler; import org.dromara.neutrinoproxy.client.core.RealServerChannelHandler; +import org.dromara.neutrinoproxy.client.core.UdpProxyChannelHandler; import org.dromara.neutrinoproxy.client.util.ProxyUtil; import org.dromara.neutrinoproxy.core.*; import org.dromara.neutrinoproxy.core.dispatcher.DefaultDispatcher; @@ -103,23 +104,44 @@ public class ProxyConfiguration implements LifecycleBean { ch.pipeline().addLast(ProxyUtil.createSslHandler(proxyConfig)); } if (null != proxyConfig.getTunnel().getTransferLogEnable() && proxyConfig.getTunnel().getTransferLogEnable()) { - ch.pipeline().addFirst(new LoggingHandler(ProxyChannelHandler.class)); + ch.pipeline().addFirst(new LoggingHandler(TcpProxyChannelHandler.class)); } ch.pipeline().addLast(new ProxyMessageDecoder(proxyConfig.getProtocol().getMaxFrameLength(), proxyConfig.getProtocol().getLengthFieldOffset(), proxyConfig.getProtocol().getLengthFieldLength(), proxyConfig.getProtocol().getLengthAdjustment(), proxyConfig.getProtocol().getInitialBytesToStrip())); ch.pipeline().addLast(new ProxyMessageEncoder()); ch.pipeline().addLast(new IdleStateHandler(proxyConfig.getProtocol().getReadIdleTime(), proxyConfig.getProtocol().getWriteIdleTime(), proxyConfig.getProtocol().getAllIdleTimeSeconds())); - ch.pipeline().addLast(new ProxyChannelHandler()); + ch.pipeline().addLast(new TcpProxyChannelHandler()); } }); return bootstrap; } @Bean("udpProxyTunnelBootstrap") - private Bootstrap udpProxyTunnelBootstrap() { + private Bootstrap udpProxyTunnelBootstrap(@Inject ProxyConfig proxyConfig, + @Inject("tunnelWorkGroup") NioEventLoopGroup tunnelWorkGroup) { Bootstrap bootstrap = new Bootstrap(); + bootstrap.group(tunnelWorkGroup); + bootstrap.channel(NioSocketChannel.class); + bootstrap.remoteAddress(InetSocketAddress.createUnresolved(proxyConfig.getTunnel().getServerIp(), proxyConfig.getTunnel().getServerPort())); + bootstrap.handler(new ChannelInitializer() { + @Override + public void initChannel(SocketChannel ch) throws Exception { + if (proxyConfig.getTunnel().getSslEnable()) { + ch.pipeline().addLast(ProxyUtil.createSslHandler(proxyConfig)); + } + if (null != proxyConfig.getTunnel().getTransferLogEnable() && proxyConfig.getTunnel().getTransferLogEnable()) { + ch.pipeline().addFirst(new LoggingHandler(TcpProxyChannelHandler.class)); + } + ch.pipeline().addLast(new ProxyMessageDecoder(proxyConfig.getProtocol().getMaxFrameLength(), + proxyConfig.getProtocol().getLengthFieldOffset(), proxyConfig.getProtocol().getLengthFieldLength(), + proxyConfig.getProtocol().getLengthAdjustment(), proxyConfig.getProtocol().getInitialBytesToStrip())); + ch.pipeline().addLast(new ProxyMessageEncoder()); + ch.pipeline().addLast(new IdleStateHandler(proxyConfig.getProtocol().getReadIdleTime(), proxyConfig.getProtocol().getWriteIdleTime(), proxyConfig.getProtocol().getAllIdleTimeSeconds())); + ch.pipeline().addLast(new UdpProxyChannelHandler()); + } + }); return bootstrap; } diff --git a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/CmdChannelHandler.java b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/CmdChannelHandler.java index c5c153e9..68db2a0e 100644 --- a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/CmdChannelHandler.java +++ b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/CmdChannelHandler.java @@ -32,7 +32,7 @@ public class CmdChannelHandler extends SimpleChannelInboundHandler @Override protected void channelRead0(ChannelHandlerContext ctx, ProxyMessage proxyMessage) throws Exception { if (ProxyMessage.TYPE_HEARTBEAT != proxyMessage.getType() || transferLogEnable) { - log.debug("Client CmdChannel recieved proxy message, type is {}", proxyMessage.getType()); + log.debug("[CMD Channel]Client CmdChannel recieved proxy message, type is {}", proxyMessage.getType()); } Solon.context().getBean(Dispatcher.class).dispatch(ctx, proxyMessage); } @@ -49,7 +49,7 @@ public class CmdChannelHandler extends SimpleChannelInboundHandler @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { - log.info("Client CmdChannel disconnect"); + log.info("[CMD Channel]Client CmdChannel disconnect"); ProxyUtil.setCmdChannel(null); ProxyUtil.clearRealServerChannels(); @@ -58,7 +58,7 @@ public class CmdChannelHandler extends SimpleChannelInboundHandler @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { - log.error("Client CmdChannel Error channelId:{}", ctx.channel().id().asLongText(), cause); + log.error("[CMD Channel]Client CmdChannel Error channelId:{}", ctx.channel().id().asLongText(), cause); ctx.close(); } @@ -69,14 +69,14 @@ public class CmdChannelHandler extends SimpleChannelInboundHandler switch (event.state()) { case READER_IDLE: // 读超时，断开连接 - log.error("[Client CmdChannel] Read timeout disconnect"); + log.error("[CMD Channel] Read timeout disconnect"); ctx.channel().close(); break; case WRITER_IDLE: ctx.channel().writeAndFlush(ProxyMessage.buildHeartbeatMessage()); break; case ALL_IDLE: - log.error("[Client CmdChannel] ReadWrite timeout disconnect"); + log.error("[CMD Channel] ReadWrite timeout disconnect"); ctx.close(); break; } diff --git a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/ProxyChannelHandler.java b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/TcpProxyChannelHandler.java similarity index 83% rename from neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/ProxyChannelHandler.java rename to neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/TcpProxyChannelHandler.java index a8a17dd4..45373152 100644 --- a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/ProxyChannelHandler.java +++ b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/TcpProxyChannelHandler.java @@ -18,13 +18,13 @@ import org.noear.solon.Solon; * @date: 2022/6/16 */ @Slf4j -public class ProxyChannelHandler extends SimpleChannelInboundHandler { +public class TcpProxyChannelHandler extends SimpleChannelInboundHandler { @Override protected void channelRead0(ChannelHandlerContext ctx, ProxyMessage proxyMessage) throws Exception { if (ProxyMessage.TYPE_HEARTBEAT != proxyMessage.getType()) { - log.debug("Client ProxyChannel recieved proxy message, type is {}", proxyMessage.getType()); + log.debug("[TCP Proxy Channel]Client ProxyChannel recieved proxy message, type is {}", proxyMessage.getType()); } Solon.context().getBean(Dispatcher.class).dispatch(ctx, proxyMessage); } @@ -53,7 +53,7 @@ public class ProxyChannelHandler extends SimpleChannelInboundHandler { + + + @Override + protected void channelRead0(ChannelHandlerContext ctx, ProxyMessage proxyMessage) throws Exception { + if (ProxyMessage.TYPE_HEARTBEAT != proxyMessage.getType()) { + log.debug("[UDP Proxy Channel]Client ProxyChannel recieved proxy message, type is {}", proxyMessage.getType()); + } + Solon.context().getBean(Dispatcher.class).dispatch(ctx, proxyMessage); + } + + @Override + public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception { + Channel realServerChannel = ctx.channel().attr(Constants.NEXT_CHANNEL).get(); + if (realServerChannel != null) { + realServerChannel.config().setOption(ChannelOption.AUTO_READ, ctx.channel().isWritable()); + } + + super.channelWritabilityChanged(ctx); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + // 数据传输连接 + Channel realServerChannel = ctx.channel().attr(Constants.NEXT_CHANNEL).get(); + if (realServerChannel != null && realServerChannel.isActive()) { + realServerChannel.close(); + } + + ProxyUtil.removeProxyChanel(ctx.channel()); + super.channelInactive(ctx); + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { + log.error("[UDP Proxy Channel]Client ProxyChannel Error channelId:{}", ctx.channel().id().asLongText(), cause); + ctx.close(); + } + + @Override + public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { + if(evt instanceof IdleStateEvent) { + IdleStateEvent event = (IdleStateEvent)evt; + switch (event.state()) { + case READER_IDLE: + // 读超时，断开连接 + log.info("[UDP Proxy Channel]Read timeout"); + ctx.channel().close(); + break; + case WRITER_IDLE: + ctx.channel().writeAndFlush(ProxyMessage.buildHeartbeatMessage()); + break; + case ALL_IDLE: + log.debug("[UDP Proxy Channel]ReadWrite timeout"); + ctx.close(); + break; + } + } + } +} diff --git a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/handler/ProxyMessageConnectHandler.java b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/handler/ProxyMessageConnectHandler.java index 50bff9ba..2bca8667 100644 --- a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/handler/ProxyMessageConnectHandler.java +++ b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/handler/ProxyMessageConnectHandler.java @@ -48,7 +48,7 @@ public class ProxyMessageConnectHandler implements ProxyMessageHandler { realServerChannel.config().setOption(ChannelOption.AUTO_READ, false); // 获取连接 - ProxyUtil.borrowProxyChanel(tcpProxyTunnelBootstrap, new ProxyChannelBorrowListener() { + ProxyUtil.borrowTcpProxyChanel(tcpProxyTunnelBootstrap, new ProxyChannelBorrowListener() { @Override public void success(Channel channel) { diff --git a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/util/ProxyUtil.java b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/util/ProxyUtil.java index 789dc9b9..d42a5cb5 100644 --- a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/util/ProxyUtil.java +++ b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/util/ProxyUtil.java @@ -71,14 +71,14 @@ public class ProxyUtil { private static String clientId; private static final String CLIENT_ID_FILE = ".NEUTRINO_PROXY_CLIENT_ID"; - public static void borrowProxyChanel(Bootstrap proxyTunnelBootstrap, final ProxyChannelBorrowListener borrowListener) { + public static void borrowTcpProxyChanel(Bootstrap tcpProxyTunnelBootstrap, final ProxyChannelBorrowListener borrowListener) { Channel channel = proxyChannelPool.poll(); if (null != channel) { borrowListener.success(channel); return; } - proxyTunnelBootstrap.connect().addListener((ChannelFutureListener) future -> { + tcpProxyTunnelBootstrap.connect().addListener((ChannelFutureListener) future -> { if (future.isSuccess()) { borrowListener.success(future.channel()); } else { diff --git a/neutrino-proxy-core/src/main/java/org/dromara/neutrinoproxy/core/ProxyMessage.java b/neutrino-proxy-core/src/main/java/org/dromara/neutrinoproxy/core/ProxyMessage.java index 716557c4..43d138e8 100644 --- a/neutrino-proxy-core/src/main/java/org/dromara/neutrinoproxy/core/ProxyMessage.java +++ b/neutrino-proxy-core/src/main/java/org/dromara/neutrinoproxy/core/ProxyMessage.java @@ -69,7 +69,7 @@ public class ProxyMessage { /** * UDP数据传输 */ - private static final byte TYPE_UDP_TRANSFER = 0x07; + private static final byte TYPE_UDP_TRANSFER = 0x08; /** * 消息类型