yeyinlaoying/neutrino-proxy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

客户端新增UDP隧道消息处理器

Browse Source
This commit is contained in:
aoshiguchen
2023-09-19 13:16:43 +08:00
parent 5578874df7
commit 6880898321
7 changed files with 121 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/config/ProxyConfiguration.java
+26 -4
View File
@@ -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<SocketChannel>() {
@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;
}
neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/CmdChannelHandler.java
+5 -5
View File
@@ -32,7 +32,7 @@ public class CmdChannelHandler extends SimpleChannelInboundHandler<ProxyMessage>
@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<ProxyMessage>
@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<ProxyMessage>
@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<ProxyMessage>
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;
}
neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/ProxyChannelHandler.java → neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/TcpProxyChannelHandler.java
+6 -6
View File
@@ -18,13 +18,13 @@ import org.noear.solon.Solon;
* @date: 2022/6/16
*/
@Slf4j
public class ProxyChannelHandler extends SimpleChannelInboundHandler<ProxyMessage> {
public class TcpProxyChannelHandler extends SimpleChannelInboundHandler<ProxyMessage> {
@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<ProxyMessag
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
log.error("Client ProxyChannel Error channelId:{}", ctx.channel().id().asLongText(), cause);
log.error("[TCP Proxy Channel]Client ProxyChannel Error channelId:{}", ctx.channel().id().asLongText(), cause);
ctx.close();
}
@@ -64,14 +64,14 @@ public class ProxyChannelHandler extends SimpleChannelInboundHandler<ProxyMessag
switch (event.state()) {
case READER_IDLE:
// 读超时，断开连接
// log.info("Read timeout");
// ctx.channel().close();
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("ReadWrite timeout");
log.debug("[TCP Proxy Channel]ReadWrite timeout");
ctx.close();
break;
}
neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/UdpProxyChannelHandler.java
+80
View File
@@ -0,0 +1,80 @@
package org.dromara.neutrinoproxy.client.core;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOption;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import org.dromara.neutrinoproxy.client.util.ProxyUtil;
import org.dromara.neutrinoproxy.core.Constants;
import org.dromara.neutrinoproxy.core.ProxyMessage;
import org.dromara.neutrinoproxy.core.dispatcher.Dispatcher;
import org.noear.solon.Solon;
/**
* 处理与服务端之间的数据传输
* @author: aoshiguchen
* @date: 2022/6/16
*/
@Slf4j
public class UdpProxyChannelHandler extends SimpleChannelInboundHandler<ProxyMessage> {
@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;
}
}
}
}
neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/handler/ProxyMessageConnectHandler.java
+1 -1
View File
@@ -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) {
neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/util/ProxyUtil.java
+2 -2
View File
@@ -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 {
neutrino-proxy-core/src/main/java/org/dromara/neutrinoproxy/core/ProxyMessage.java
+1 -1
View File
@@ -69,7 +69,7 @@ public class ProxyMessage {
/**
* UDP数据传输
*/
private static final byte TYPE_UDP_TRANSFER = 0x07;
private static final byte TYPE_UDP_TRANSFER = 0x08;
/**
* 消息类型