针对HTTP代理附加对象的优化

neutrino-proxy-core/src/main/java/org/dromara/neutrinoproxy/core/Constants.java

@@ -4,6 +4,7 @@ import io.netty.channel.Channel;
 import io.netty.util.AttributeKey;
 
 import java.net.InetSocketAddress;
+import java.util.function.BiConsumer;
 
 /**
  *
@@ -33,6 +34,8 @@ public interface Constants {
 
     AttributeKey<Boolean> FLOW_LIMITER_FLAG = AttributeKey.newInstance("flowLimiterFlag");
 
+    AttributeKey<ProxyAttachment> PROXY_CONNECT_ATTACHMENT = AttributeKey.newInstance("proxyConnectAttachment");;
+
 
     int HEADER_SIZE = 4;
     int TYPE_SIZE = 1;
@@ -51,4 +54,20 @@ public interface Constants {
         String ERROR = "ERROR";
         String PORT_MAPPING_SYNC = "PORT_MAPPING_SYNC";
     }
+
+    class ProxyAttachment {
+        private byte[] bytes;
+        private BiConsumer<Channel, byte[]> executor;
+
+        public ProxyAttachment(byte[] bytes, BiConsumer<Channel, byte[]> executor) {
+            this.bytes = bytes;
+            this.executor = executor;
+        }
+
+        public void execute(Channel channel) {
+            if (null != executor && null != channel && channel.isActive()) {
+                this.executor.accept(channel, bytes);
+            }
+        }
+    }
 }

neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/UdpVisitorChannelHandler.java

@@ -6,20 +6,16 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.channel.socket.DatagramPacket;
-import io.netty.handler.codec.DecoderException;
 import lombok.extern.slf4j.Slf4j;
 import org.dromara.neutrinoproxy.core.Constants;
 import org.dromara.neutrinoproxy.core.ProxyMessage;
 import org.dromara.neutrinoproxy.server.constant.NetworkProtocolEnum;
-import org.dromara.neutrinoproxy.server.proxy.domain.ProxyAttachment;
 import org.dromara.neutrinoproxy.server.proxy.domain.VisitorChannelAttachInfo;
 import org.dromara.neutrinoproxy.server.service.FlowReportService;
 import org.dromara.neutrinoproxy.server.util.ProxyUtil;
 import org.noear.solon.Solon;
 
-import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.net.SocketException;
 
 /**
  * @author: aoshiguchen
@@ -36,7 +32,7 @@ public class UdpVisitorChannelHandler extends SimpleChannelInboundHandler<Datagr
         byte[] bytes = new byte[datagramPacket.content().readableBytes()];
         datagramPacket.content().readBytes(bytes);
         datagramPacket.content().resetReaderIndex();
-        ProxyAttachment proxyAttachment = new ProxyAttachment(ctx.channel(), bytes, (channel, buf) -> {
+        Constants.ProxyAttachment proxyAttachment = new Constants.ProxyAttachment(bytes, (channel, buf) -> {
             String visitorId = ProxyUtil.getVisitorIdBySocketAddress(datagramPacket.sender());
             if (StrUtil.isBlank(visitorId)) {
                 return;
@@ -76,7 +72,7 @@ public class UdpVisitorChannelHandler extends SimpleChannelInboundHandler<Datagr
         Channel proxyChannel = ProxyUtil.getTunnelChannelByVisitorId(visitorId);
         if (StrUtil.isNotBlank(visitorId) && null != proxyChannel && proxyChannel.isActive()) {
             // UDP代理隧道已就绪，直接转发
-            proxyAttachment.execute();
+            proxyAttachment.execute(ctx.channel());
             return;
         }
 

neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/domain/ProxyAttachment.java

@@ -1,29 +1,28 @@
-package org.dromara.neutrinoproxy.server.proxy.domain;
+//package org.dromara.neutrinoproxy.server.proxy.domain;
-
+//
-import io.netty.buffer.ByteBuf;
+//import io.netty.channel.Channel;
-import io.netty.channel.Channel;
+//
-
+//import java.util.function.BiConsumer;
-import java.util.function.BiConsumer;
+//
-
+///**
-/**
+// * 代理连接附件
- * 代理连接附件
+// * @author: aoshiguchen
- * @author: aoshiguchen
+// * @date: 2023/4/2
- * @date: 2023/4/2
+// */
- */
+//public class ProxyAttachment {
-public class ProxyAttachment {
+//    private Channel channel;
-    private Channel channel;
+//    private byte[] bytes;
-    private byte[] bytes;
+//    private BiConsumer<Channel, byte[]> executor;
-    private BiConsumer<Channel, byte[]> executor;
+//
-
+//    public ProxyAttachment(Channel channel, byte[] bytes, BiConsumer<Channel, byte[]> executor) {
-    public ProxyAttachment(Channel channel, byte[] bytes, BiConsumer<Channel, byte[]> executor) {
+//        this.channel = channel;
-        this.channel = channel;
+//        this.bytes = bytes;
-        this.bytes = bytes;
+//        this.executor = executor;
-        this.executor = executor;
+//    }
-    }
+//
-
+//    public void execute() {
-    public void execute() {
+//        if (null != executor) {
-        if (null != executor) {
+//            this.executor.accept(channel, bytes);
-            this.executor.accept(channel, bytes);
+//        }
-        }
+//    }
-    }
+//}
-}

neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/enhance/HttpVisitorChannelHandler.java

@@ -9,7 +9,6 @@ import org.apache.commons.lang3.StringUtils;
 import org.dromara.neutrinoproxy.core.Constants;
 import org.dromara.neutrinoproxy.core.ProxyMessage;
 import org.dromara.neutrinoproxy.server.constant.NetworkProtocolEnum;
-import org.dromara.neutrinoproxy.server.proxy.domain.ProxyAttachment;
 import org.dromara.neutrinoproxy.server.proxy.domain.VisitorChannelAttachInfo;
 import org.dromara.neutrinoproxy.server.service.FlowReportService;
 import org.dromara.neutrinoproxy.server.util.ProxyUtil;
@@ -31,7 +30,7 @@ public class HttpVisitorChannelHandler extends SimpleChannelInboundHandler<ByteB
         byte[] bytes = new byte[byteBuf.readableBytes()];
         byteBuf.readBytes(bytes);
         byteBuf.resetReaderIndex();
-        ProxyAttachment proxyAttachment = new ProxyAttachment(ctx.channel(), bytes, (channel, buf) -> {
+        Constants.ProxyAttachment proxyAttachment = new Constants.ProxyAttachment(bytes, (channel, buf) -> {
             Channel proxyChannel = channel.attr(Constants.NEXT_CHANNEL).get();
             if (null == proxyChannel) {
                 // 该端口还没有代理客户端
@@ -48,7 +47,7 @@ public class HttpVisitorChannelHandler extends SimpleChannelInboundHandler<ByteB
 
         String visitorId = ProxyUtil.getVisitorIdByChannel(ctx.channel());
         if (StringUtils.isNotBlank(visitorId)) {
-            proxyAttachment.execute();
+            proxyAttachment.execute(ctx.channel());
             return;
         }
 
@@ -71,7 +70,8 @@ public class HttpVisitorChannelHandler extends SimpleChannelInboundHandler<ByteB
 
         visitorId = ProxyUtil.newVisitorId();
         ProxyUtil.addVisitorChannelToCmdChannel(NetworkProtocolEnum.HTTP, cmdChannel, visitorId, ctx.channel(), serverPort);
-        ProxyUtil.addProxyConnectAttachment(visitorId, proxyAttachment);
+        // ProxyUtil.addProxyConnectAttachment(visitorId, proxyAttachment);
+        ctx.channel().attr(Constants.PROXY_CONNECT_ATTACHMENT).set(proxyAttachment);
         cmdChannel.writeAndFlush(ProxyMessage.buildConnectMessage(visitorId).setData(lanInfo.getBytes()));
     }
 

neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/handler/ProxyMessageConnectHandler.java

@@ -7,7 +7,6 @@ import org.dromara.neutrinoproxy.core.dispatcher.Match;
 import org.dromara.neutrinoproxy.server.constant.EnableStatusEnum;
 import org.dromara.neutrinoproxy.server.dal.entity.LicenseDO;
 import org.dromara.neutrinoproxy.server.dal.entity.UserDO;
-import org.dromara.neutrinoproxy.server.proxy.domain.ProxyAttachment;
 import org.dromara.neutrinoproxy.server.service.LicenseService;
 import org.dromara.neutrinoproxy.server.service.UserService;
 import org.dromara.neutrinoproxy.server.util.ProxyUtil;
@@ -76,8 +75,18 @@ public class ProxyMessageConnectHandler implements ProxyMessageHandler {
 
 		Channel visitorChannel = ProxyUtil.getVisitorChannel(cmdChannel, visitorId);
 		if (null == visitorChannel) {
+            ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "server error，visitor channel not found!"));
+            ctx.channel().close();
 			return;
 		}
+        // 获取代理附加对象
+        Constants.ProxyAttachment proxyAttachment = visitorChannel.attr(Constants.PROXY_CONNECT_ATTACHMENT).get();
+        if (null == proxyAttachment) {
+            ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "server error，visitor channel proxy attachment not found!"));
+            ctx.channel().close();
+            visitorChannel.close();
+            return;
+        }
 		ctx.channel().attr(Constants.VISITOR_ID).set(visitorId);
 		ctx.channel().attr(Constants.LICENSE_ID).set(licenseDO.getId());
 		ctx.channel().attr(Constants.NEXT_CHANNEL).set(visitorChannel);
@@ -86,13 +95,16 @@ public class ProxyMessageConnectHandler implements ProxyMessageHandler {
         // 代理客户端与后端服务器连接成功，修改用户连接为可读状态
 		visitorChannel.config().setOption(ChannelOption.AUTO_READ, true);
 
-		// 获取代理附加对象
+        // 转发来自visitor的首次代理数据
-		ProxyAttachment proxyAttachment = ProxyUtil.getProxyConnectAttachment(visitorId);
+        proxyAttachment.execute(visitorChannel);
-		if (null != proxyAttachment) {
+
-			// 及时释放
+//		// 获取代理附加对象
-			ProxyUtil.remoteProxyConnectAttachment(visitorId);
+//		ProxyAttachment proxyAttachment = ProxyUtil.getProxyConnectAttachment(visitorId);
-			proxyAttachment.execute();
+//		if (null != proxyAttachment) {
-		}
+//			// 及时释放
+//			ProxyUtil.remoteProxyConnectAttachment(visitorId);
+//			proxyAttachment.execute();
+//		}
 	}
 
 	@Override

neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/handler/UdpProxyMessageConnectHandler.java

@@ -14,7 +14,6 @@ import org.dromara.neutrinoproxy.server.dal.PortMappingMapper;
 import org.dromara.neutrinoproxy.server.dal.entity.LicenseDO;
 import org.dromara.neutrinoproxy.server.dal.entity.PortMappingDO;
 import org.dromara.neutrinoproxy.server.dal.entity.UserDO;
-import org.dromara.neutrinoproxy.server.proxy.domain.ProxyAttachment;
 import org.dromara.neutrinoproxy.server.service.LicenseService;
 import org.dromara.neutrinoproxy.server.service.UserService;
 import org.dromara.neutrinoproxy.server.util.ProxyUtil;
@@ -94,11 +93,11 @@ public class UdpProxyMessageConnectHandler implements ProxyMessageHandler {
 //        // 代理客户端与后端服务器连接成功，修改用户连接为可读状态
 //        visitorChannel.config().setOption(ChannelOption.AUTO_READ, true);
         // 获取代理附加对象
-        ProxyAttachment proxyAttachment = ProxyUtil.getProxyConnectAttachment(udpBaseInfo.getVisitorId());
+        Constants.ProxyAttachment proxyAttachment = ProxyUtil.getProxyConnectAttachment(udpBaseInfo.getVisitorId());
         if (null != proxyAttachment) {
             // 及时释放
             ProxyUtil.remoteProxyConnectAttachment(udpBaseInfo.getVisitorId());
-            proxyAttachment.execute();
+            proxyAttachment.execute(visitorChannel);
         }
     }
 

neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/util/ProxyUtil.java

@@ -4,9 +4,9 @@ import cn.hutool.core.collection.CollectionUtil;
 import com.google.common.collect.Sets;
 import org.apache.commons.lang3.StringUtils;
 import org.dromara.neutrinoproxy.core.ChannelAttribute;
+import org.dromara.neutrinoproxy.core.Constants;
 import org.dromara.neutrinoproxy.server.constant.NetworkProtocolEnum;
 import org.dromara.neutrinoproxy.server.proxy.domain.CmdChannelAttachInfo;
-import org.dromara.neutrinoproxy.server.proxy.domain.ProxyAttachment;
 import org.dromara.neutrinoproxy.server.proxy.domain.ProxyMapping;
 import org.dromara.neutrinoproxy.server.proxy.domain.VisitorChannelAttachInfo;
 import io.netty.channel.Channel;
@@ -59,8 +59,10 @@ public class ProxyUtil {
 	private static AtomicLong visitorIdProducer = new AtomicLong(0);
 	/**
 	 * 代理 - connect附加映射
+     * （TCP/HTTP场景下，已优化不是用这个。UDP由于无连接，暂时还是要用这个）
+     * UDP场景时，需要记录最后读写时间，使用定时器定期检查，超时需要释放，待后续优化
 	 */
-	private static Map<String, ProxyAttachment> proxyConnectAttachmentMap = new HashMap<>();
+	private static Map<String, Constants.ProxyAttachment> proxyConnectAttachmentMap = new HashMap<>();
 	/**
 	 * 完整域名 - 服务端端口映射
 	 */
@@ -334,7 +336,7 @@ public class ProxyUtil {
 	 * @param visitorId
 	 * @param proxyAttachment
 	 */
-	public static void addProxyConnectAttachment(String visitorId, ProxyAttachment proxyAttachment) {
+	public static void addProxyConnectAttachment(String visitorId, Constants.ProxyAttachment proxyAttachment) {
 		proxyConnectAttachmentMap.put(visitorId, proxyAttachment);
 	}
 
@@ -343,7 +345,7 @@ public class ProxyUtil {
 	 * @param visitorId
 	 * @return
 	 */
-	public static ProxyAttachment getProxyConnectAttachment(String visitorId) {
+	public static Constants.ProxyAttachment getProxyConnectAttachment(String visitorId) {
 		return proxyConnectAttachmentMap.get(visitorId);
 	}
 
@@ -449,7 +451,7 @@ public class ProxyUtil {
 		if (StringUtils.isBlank(visitorId)) {
 			return;
 		}
-		ProxyAttachment proxyAttachment = ProxyUtil.getProxyConnectAttachment(visitorId);
+        Constants.ProxyAttachment proxyAttachment = ProxyUtil.getProxyConnectAttachment(visitorId);
 		if (null != proxyAttachment) {
 			tryClose(channel);
 		}