[fix] socks、http代理 添加流量计费

wu-lazy-cloud-heartbeat-client/README.md

@@ -10,8 +10,8 @@ mvn native:build -Pnative
 ### 构建docker镜像
 
 ```shell
-docker build -t docker-registry.wujiawei.com/lazy/lazy-under-cloud-heartbeat-client:lazy-2.4.2-native-SNAPSHOT_latest -f Native-Dockerfile .
+docker build -t docker-registry.wujiawei.com/lazy/lazy-under-cloud-heartbeat-client:lazy-2.4.2-NATIVE-SNAPSHOT_latest -f Native-Dockerfile .
-docker push docker-registry.wujiawei.com/lazy/lazy-under-cloud-heartbeat-client:lazy-2.4.2-native-SNAPSHOT_latest
+docker push docker-registry.wujiawei.com/lazy/lazy-under-cloud-heartbeat-client:lazy-2.4.2-NATIVE-SNAPSHOT_latest
 ```
 
 ## BUILD IMAGE

wu-lazy-cloud-heartbeat-client/src/main/java/org/framework/lazy/cloud/network/heartbeat/client/netty/proxy/http/handler/NettyHttpClientProxyClientRealHandler.java

@@ -25,12 +25,14 @@ public class NettyHttpClientProxyClientRealHandler extends SimpleChannelInboundH
         log.debug("客户端代理客户端，接收目标客户端真实服务数据:{}", new String(bytes));
         String visitorId = ChannelAttributeKeyUtils.getVisitorId(ctx.channel());
         String clientId = ChannelAttributeKeyUtils.getClientId(ctx.channel());
+        Integer visitorPort = ChannelAttributeKeyUtils.getVisitorPort(ctx.channel());
         // 访客通信通道 上报服务端代理完成
         Channel nextChannel = ChannelAttributeKeyUtils.getNextChannel(ctx.channel());
         NettyProxyMsg returnMessage = new NettyProxyMsg();
         returnMessage.setType(ProxyMessageType.HTTP_REPORT_CLIENT_PROXY_CLIENT_TRANSFER_RESPONSE_);
         returnMessage.setVisitorId(visitorId);
         returnMessage.setClientId(clientId);
+        returnMessage.setVisitorPort(visitorPort);
         returnMessage.setData(bytes);
 
         nextChannel.writeAndFlush(returnMessage);

wu-lazy-cloud-heartbeat-client/src/main/java/org/framework/lazy/cloud/network/heartbeat/client/netty/proxy/http/socket/NettyHttpClientProxyClientRealSocket.java

@@ -28,8 +28,8 @@ public class NettyHttpClientProxyClientRealSocket {
 
 
     public static void buildRealServer(String clientId,
-                                       String clientTargetIp,
+                                       String targetIp,
-                                       Integer clientTargetPort,
+                                       Integer targetPort,
                                        String visitorId,
                                        NettyClientProperties nettyClientProperties,
                                        List<HandleChannelTypeAdvanced> handleChannelTypeAdvancedList) {
@@ -52,25 +52,26 @@ public class NettyHttpClientProxyClientRealSocket {
             ;
 
 
-            bootstrap.connect(clientTargetIp, clientTargetPort).addListener((ChannelFutureListener) future -> {
+            bootstrap.connect(targetIp, targetPort).addListener((ChannelFutureListener) future -> {
                 if (future.isSuccess()) {
                     // 客户端链接真实服务成功 设置自动读写false 等待访客连接成功后设置成true
                     Channel realChannel = future.channel();
 //                    realChannel.config().setOption(ChannelOption.AUTO_READ, false);
 
-                    log.info("访客通过 客户端:【{}】,visitorId:{},绑定本地服务,IP:{},端口:{} 新建通道成功", clientId, visitorId, clientTargetIp, clientTargetPort);
+                    log.info("访客通过 客户端:【{}】,visitorId:{},绑定本地服务,IP:{},端口:{} 新建通道成功", clientId, visitorId, targetIp, targetPort);
                     // 客户端真实通道
                     NettyRealIdContext.pushReal(realChannel, visitorId);
                     // 绑定访客ID到当前真实通道属性
                     ChannelAttributeKeyUtils.buildVisitorId(realChannel, visitorId);
                     ChannelAttributeKeyUtils.buildClientId(realChannel, clientId);
+                    ChannelAttributeKeyUtils.buildVisitorPort(realChannel, targetPort);
 
                     // 连接服务端 然后绑定通道
                     // 新建一个通道处理
                     newVisitorConnect2Server(
                             clientId,
-                            clientTargetIp,
+                            targetIp,
-                            clientTargetPort,
+                            targetPort,
                             visitorId,
                             realChannel,
                             nettyClientProperties,
@@ -79,7 +80,7 @@ public class NettyHttpClientProxyClientRealSocket {
 
 
                 } else {
-                    log.error("客户：【{}】,无法连接当前网络内的目标IP：【{}】,目标端口:【{}】", clientId, clientTargetIp, clientTargetPort);
+                    log.error("客户：【{}】,无法连接当前网络内的目标IP：【{}】,目标端口:【{}】", clientId, targetIp, targetPort);
                 }
             });
         } catch (Exception e) {

wu-lazy-cloud-heartbeat-client/src/main/java/org/framework/lazy/cloud/network/heartbeat/client/netty/proxy/http/socket/NettyHttpClientProxyServerProxySocket.java

@@ -72,7 +72,7 @@ public class NettyHttpClientProxyServerProxySocket {
             if (futureListener.isSuccess()) {
 
                 NettyProxyMsg nettyProxyMsg = new NettyProxyMsg();
-                nettyProxyMsg.setType(ProxyMessageType.HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_);
+                nettyProxyMsg.setType(ProxyMessageType.HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_REQUEST_);
                 // other clientId
                 nettyProxyMsg.setClientId(clientId);
 

wu-lazy-cloud-heartbeat-common/src/main/java/org/framework/lazy/cloud/network/heartbeat/common/adapter/ChannelProxyFlowAdapter.java

@@ -0,0 +1,65 @@
+package org.framework.lazy.cloud.network.heartbeat.common.adapter;
+
+import io.netty.channel.Channel;
+import lombok.extern.slf4j.Slf4j;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.HandleChannelFlowAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.proxy.ChannelProxyFlow;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.proxy.HandleChannelProxyFlowAdvanced;
+
+import java.util.List;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 通道流量适配器
+ *
+ * @see HandleChannelFlowAdvanced
+ */
+@Slf4j
+public class ChannelProxyFlowAdapter {
+
+
+    ThreadPoolExecutor CHANNEL_FLOW_ADAPTER_EXECUTOR =
+            new ThreadPoolExecutor(20, 200, 3L, TimeUnit.MINUTES,
+                    new LinkedBlockingDeque<>(500),new ThreadPoolExecutor.AbortPolicy());
+    // 线程使用完后使用主线程执行
+
+
+
+    protected final List<HandleChannelProxyFlowAdvanced> handleChannelProxyFlowAdvancedList;
+
+    public ChannelProxyFlowAdapter(List<HandleChannelProxyFlowAdvanced> handleChannelProxyFlowAdvancedList) {
+        this.handleChannelProxyFlowAdvancedList = handleChannelProxyFlowAdvancedList;
+    }
+
+
+    /**
+     * 处理当前数据
+     *
+     * @param channelProxyFlow 通道数据
+     */
+    public void handler(Channel channel, ChannelProxyFlow channelProxyFlow) {
+        for (HandleChannelProxyFlowAdvanced handleChannelProxyFlowAdvanced : handleChannelProxyFlowAdvancedList) {
+            if (handleChannelProxyFlowAdvanced.support(channelProxyFlow)) {
+                try {
+                    handleChannelProxyFlowAdvanced.handler(channel, channelProxyFlow);
+                } catch (Exception e) {
+                    log.error("流量统计失败:{}", e.getMessage());
+                }
+                return;
+            }
+        }
+    }
+
+    /**
+     * 异步处理当前数据
+     *
+     * @param channelProxyFlow 通道数据
+     */
+    public void asyncHandler(Channel channel, ChannelProxyFlow channelProxyFlow) {
+     // TODO 流量并发异常
+        CHANNEL_FLOW_ADAPTER_EXECUTOR.submit(() -> handler(channel, channelProxyFlow));
+    }
+
+}

wu-lazy-cloud-heartbeat-common/src/main/java/org/framework/lazy/cloud/network/heartbeat/common/advanced/flow/proxy/AbstractHandleChannelProxyFlowAdvanced.java

@@ -0,0 +1,49 @@
+package org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.proxy;
+
+import io.netty.channel.Channel;
+
+/**
+ * 处理通道流量适配者 抽象类
+ *
+ * @see HandleChannelProxyFlowAdvanced
+ */
+public abstract class AbstractHandleChannelProxyFlowAdvanced implements HandleChannelProxyFlowAdvanced {
+
+    /**
+     * 是否支持当前这种类型
+     *
+     * @param channelProxyFlow 数据
+     * @return boolean
+     */
+    @Override
+    public boolean support(ChannelProxyFlow channelProxyFlow) {
+        return doSupport(channelProxyFlow);
+    }
+
+    /**
+     * 处理是否支持这种类型
+     *
+     * @param channelProxyFlow 数据
+     * @return boolean
+     */
+    protected abstract boolean doSupport(ChannelProxyFlow channelProxyFlow);
+
+    /**
+     * 处理当前数据
+     *
+     * @param channel     当前通道
+     * @param channelProxyFlow 通道数据
+     */
+    @Override
+    public void handler(Channel channel, ChannelProxyFlow channelProxyFlow) {
+        doHandler(channel, channelProxyFlow);
+    }
+
+    /**
+     * 处理当前数据
+     *
+     * @param channel     当前通道
+     * @param channelProxyFlow 通道数据
+     */
+    protected abstract void doHandler(Channel channel, ChannelProxyFlow channelProxyFlow);
+}

wu-lazy-cloud-heartbeat-common/src/main/java/org/framework/lazy/cloud/network/heartbeat/common/advanced/flow/proxy/ChannelProxyFlow.java

@@ -0,0 +1,43 @@
+package org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.proxy;
+
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ChannelFlowEnum;
+
+public interface ChannelProxyFlow {
+
+    /**
+     * 通道客户端ID
+     *
+     * @return 通道客户端ID
+     */
+    String clientId();
+
+    /**
+     * ip
+     * @return ip
+     */
+    String ip();
+
+    /**
+     * 通道使用的端口（服务端访客端口、客户端真实端口）
+     *
+     * @return 端口
+     */
+    Integer port();
+
+    /**
+     * 通道流量类型
+     *
+     * @return ChannelFlowEnum
+     * @see ChannelFlowEnum
+     */
+    ChannelFlowEnum channelFlowEnum();
+
+    /**
+     * 流量
+     *
+     * @return 流量
+     */
+    Integer flow();
+}
+
+

wu-lazy-cloud-heartbeat-common/src/main/java/org/framework/lazy/cloud/network/heartbeat/common/advanced/flow/proxy/HandleChannelProxyFlowAdvanced.java

@@ -0,0 +1,25 @@
+package org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.proxy;
+
+import io.netty.channel.Channel;
+
+/**
+ * 处理通道流量适配者
+ */
+public interface HandleChannelProxyFlowAdvanced {
+
+    /**
+     * 是否支持当前这种类型
+     *
+     * @param channelProxyFlow 数据
+     * @return boolean
+     */
+    boolean support(ChannelProxyFlow channelProxyFlow);
+
+    /**
+     * 处理当前数据
+     *
+     * @param channel     当前通道
+     * @param channelProxyFlow 通道数据
+     */
+    void handler(Channel channel, ChannelProxyFlow channelProxyFlow);
+}

wu-lazy-cloud-heartbeat-common/src/main/java/org/framework/lazy/cloud/network/heartbeat/common/advanced/payload/NettyProxyMsg.java

@@ -201,6 +201,24 @@ public class NettyProxyMsg {
         return new String(clientTargetPort, StandardCharsets.UTF_8);
     }
 
+    /**
+     * 获取访客ID
+     * @return
+     */
+    public String getVisitorIdString() {
+        if (ObjectUtils.isEmpty(visitorId)) {
+            return null;
+        }
+        return new String(visitorId, StandardCharsets.UTF_8);
+    }
+
+    public Integer getVisitorPortInt() {
+        if (ObjectUtils.isEmpty(visitorPort)) {
+            return null;
+        }
+        return Integer.valueOf(new String(visitorPort, StandardCharsets.UTF_8));
+    }
+
     public void setClientTargetIp(byte[] clientTargetIp) {
         this.clientTargetIp = clientTargetIp;
     }
@@ -253,4 +271,6 @@ public class NettyProxyMsg {
     public void setIsSsl(byte isSsl) {
         this.isSsl = isSsl;
     }
+
+
 }

wu-lazy-cloud-heartbeat-common/src/main/java/org/framework/lazy/cloud/network/heartbeat/common/advanced/proxy/http/server/AbstractHandleHttpReportClientProxyServerTransferRequestTypeAdvanced.java

@@ -9,7 +9,7 @@ import org.framework.lazy.cloud.network.heartbeat.common.enums.ProxyMessageTypeE
  * 服务端处理客户端上报的代理请求数据
  * @param <MSG>
  */
-public abstract class AbstractHandleHttpReportClientProxyServerTransferTypeAdvanced<MSG> extends AbstractHandleChannelTypeAdvanced<NettyProxyMsg> implements HandleChannelTypeAdvanced {
+public abstract class AbstractHandleHttpReportClientProxyServerTransferRequestTypeAdvanced<MSG> extends AbstractHandleChannelTypeAdvanced<NettyProxyMsg> implements HandleChannelTypeAdvanced {
 
 
     /**
@@ -20,6 +20,6 @@ public abstract class AbstractHandleHttpReportClientProxyServerTransferTypeAdvan
      */
     @Override
     protected boolean doSupport(NettyProxyMsg nettyProxyMsg) {
-        return ProxyMessageTypeEnums.HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_.getTypeByte() == nettyProxyMsg.getType();
+        return ProxyMessageTypeEnums.HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_REQUEST_.getTypeByte() == nettyProxyMsg.getType();
     }
 }

wu-lazy-cloud-heartbeat-common/src/main/java/org/framework/lazy/cloud/network/heartbeat/common/advanced/proxy/http/server/AbstractHandleHttpReportClientProxyServerTransferResponseTypeAdvanced.java

@@ -0,0 +1,25 @@
+package org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.http.server;
+
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.AbstractHandleChannelTypeAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.HandleChannelTypeAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ProxyMessageTypeEnums;
+
+/**
+ * 服务端处理客户端上报的代理请求数据
+ * @param <MSG>
+ */
+public abstract class AbstractHandleHttpReportClientProxyServerTransferResponseTypeAdvanced<MSG> extends AbstractHandleChannelTypeAdvanced<NettyProxyMsg> implements HandleChannelTypeAdvanced {
+
+
+    /**
+     * 是否支持当前类型
+     *
+     * @param nettyProxyMsg 通道数据
+     * @return 布尔类型 是、否
+     */
+    @Override
+    protected boolean doSupport(NettyProxyMsg nettyProxyMsg) {
+        return ProxyMessageTypeEnums.HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_.getTypeByte() == nettyProxyMsg.getType();
+    }
+}

wu-lazy-cloud-heartbeat-common/src/main/java/org/framework/lazy/cloud/network/heartbeat/common/advanced/proxy/socks/server/AbstractHandleSocksReportClientProxyServerTransferResponseTypeAdvanced.java

@@ -0,0 +1,21 @@
+package org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.socks.server;
+
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.AbstractHandleChannelTypeAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.HandleChannelTypeAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ProxyMessageTypeEnums;
+
+public abstract class AbstractHandleSocksReportClientProxyServerTransferResponseTypeAdvanced<MSG> extends AbstractHandleChannelTypeAdvanced<NettyProxyMsg> implements HandleChannelTypeAdvanced {
+
+
+    /**
+     * 是否支持当前类型
+     *
+     * @param nettyProxyMsg 通道数据
+     * @return 布尔类型 是、否
+     */
+    @Override
+    protected boolean doSupport(NettyProxyMsg nettyProxyMsg) {
+        return ProxyMessageTypeEnums.SOCKS_REPORT_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_.getTypeByte() == nettyProxyMsg.getType();
+    }
+}

wu-lazy-cloud-heartbeat-common/src/main/java/org/framework/lazy/cloud/network/heartbeat/common/advanced/proxy/socks/server/AbstractHandleSocksReportServerProxyClientRequestTypeAdvanced.java

@@ -0,0 +1,21 @@
+package org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.socks.server;
+
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.AbstractHandleChannelTypeAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.HandleChannelTypeAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ProxyMessageTypeEnums;
+
+public abstract class AbstractHandleSocksReportServerProxyClientRequestTypeAdvanced<MSG> extends AbstractHandleChannelTypeAdvanced<NettyProxyMsg> implements HandleChannelTypeAdvanced {
+
+
+    /**
+     * 是否支持当前类型
+     *
+     * @param nettyProxyMsg 通道数据
+     * @return 布尔类型 是、否
+     */
+    @Override
+    protected boolean doSupport(NettyProxyMsg nettyProxyMsg) {
+        return ProxyMessageTypeEnums.SOCKS_REPORT_SERVER_PROXY_CLIENT_TRANSFER_REQUEST_.getTypeByte() == nettyProxyMsg.getType();
+    }
+}

wu-lazy-cloud-heartbeat-common/src/main/java/org/framework/lazy/cloud/network/heartbeat/common/constant/ProxyMessageType.java

@@ -42,19 +42,29 @@ public class ProxyMessageType {
      */
     public static final byte HTTP_CLIENT_PROXY_SERVER_ = HTTP_CLIENT_PROXY_CLIENT_ + 1;
     /**
-     * http 远程服务端代理传输数据上报
+     * http 远程服务端代理传输数据上报请求
      *
-     * @see ProxyMessageTypeEnums#HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_
+     * @see ProxyMessageTypeEnums#HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_REQUEST_
-     * @see AbstractHandleHttpReportClientProxyServerTransferTypeAdvanced
+     * @see AbstractHandleHttpReportClientProxyServerTransferRequestTypeAdvanced
      */
-    public static final byte HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_ = HTTP_CLIENT_PROXY_SERVER_ + 1;
+    public static final byte HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_REQUEST_ = HTTP_CLIENT_PROXY_SERVER_ + 1;
+
+    /**
+     * http 远程服务端代理传输数据上报请求结果
+     *
+     * @see ProxyMessageTypeEnums#HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_
+     * @see AbstractHandleHttpReportClientProxyServerTransferResponseTypeAdvanced
+     */
+    public static final byte HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_ = HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_REQUEST_ + 1;
+
+
     /**
      * http 远程代理服务通道关闭
      *
      * @see ProxyMessageTypeEnums#HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_CLOSE_
      * @see AbstractHandleHttpReportClientProxyServerTransferCloseTypeAdvanced
      */
-    public static final byte HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_CLOSE_ = HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_ + 1;
+    public static final byte HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_CLOSE_ = HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_ + 1;
 
 
     /**
@@ -165,6 +175,14 @@ public class ProxyMessageType {
      */
     public static final byte SOCKS_REPORT_CLIENT_PROXY_SERVER_TRANSFER_REQUEST_ = SOCKS_REPORT_CLIENT_PROXY_SERVER_TRANSFER_CONNECTION_ + 1;
 
+    /**
+     * socket 上报客户端代理服务端结果
+     *
+     * @see ProxyMessageTypeEnums#SOCKS_REPORT_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_
+     * @see AbstractHandleSocksReportClientProxyServerTransferResponseTypeAdvanced
+     */
+    public static final byte SOCKS_REPORT_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_ = SOCKS_REPORT_CLIENT_PROXY_SERVER_TRANSFER_REQUEST_ + 1;
+
 
     /**
      * socket 上报客户端代理服务端传输通道关闭
@@ -172,7 +190,7 @@ public class ProxyMessageType {
      * @see ProxyMessageTypeEnums#SOCKS_REPORT_CLIENT_PROXY_SERVER_TRANSFER_CLOSE_
      * @see AbstractHandleSocksReportClientProxyServerTransferCloseTypeAdvanced
      */
-    public static final byte SOCKS_REPORT_CLIENT_PROXY_SERVER_TRANSFER_CLOSE_ = SOCKS_REPORT_CLIENT_PROXY_SERVER_TRANSFER_REQUEST_ + 1;
+    public static final byte SOCKS_REPORT_CLIENT_PROXY_SERVER_TRANSFER_CLOSE_ = SOCKS_REPORT_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_ + 1;
 
 
 
@@ -253,13 +271,21 @@ public class ProxyMessageType {
     public static final byte SOCKS_REPORT_SERVER_PROXY_CLIENT_CONNECTION_SUCCESS_ = SOCKS_SERVER_PROXY_CLIENT_ + 1;
 
 
+    /**
+     * socket 上报客户端代理服务端 请求
+     *
+     * @see ProxyMessageTypeEnums#SOCKS_REPORT_SERVER_PROXY_CLIENT_TRANSFER_REQUEST_
+     * @see AbstractHandleSocksReportServerProxyClientRequestTypeAdvanced
+     */
+    public static final byte SOCKS_REPORT_SERVER_PROXY_CLIENT_TRANSFER_REQUEST_ = SOCKS_REPORT_SERVER_PROXY_CLIENT_CONNECTION_SUCCESS_ + 1;
+
     /**
      * socket 上报客户端代理服务端 结果返回
      *
      * @see ProxyMessageTypeEnums#SOCKS_REPORT_SERVER_PROXY_CLIENT_TRANSFER_RESPONSE_
      * @see AbstractHandleSocksReportServerProxyClientResponseTypeAdvanced
      */
-    public static final byte SOCKS_REPORT_SERVER_PROXY_CLIENT_TRANSFER_RESPONSE_ = SOCKS_REPORT_SERVER_PROXY_CLIENT_CONNECTION_SUCCESS_ + 1;
+    public static final byte SOCKS_REPORT_SERVER_PROXY_CLIENT_TRANSFER_RESPONSE_ = SOCKS_REPORT_SERVER_PROXY_CLIENT_TRANSFER_REQUEST_ + 1;
 
     /**
      * socket 上报客户端代理服务端 传输通道关闭

wu-lazy-cloud-heartbeat-common/src/main/java/org/framework/lazy/cloud/network/heartbeat/common/enums/ProxyMessageTypeEnums.java

@@ -40,9 +40,16 @@ public enum ProxyMessageTypeEnums {
     HTTP_CLIENT_PROXY_SERVER_(ProxyMessageType.HTTP_CLIENT_PROXY_SERVER_, "http远程服务端代理"),
     /**
      * http 远程服务端代理传输数据上报
-     * @see AbstractHandleHttpReportClientProxyServerTransferTypeAdvanced
+     * @see AbstractHandleHttpReportClientProxyServerTransferRequestTypeAdvanced
      */
-    HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_(ProxyMessageType.HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_, "http远程服务端代理传输数据上报"),
+    HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_REQUEST_(ProxyMessageType.HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_REQUEST_, "远程服务端代理传输数据上报请求"),
+    /**
+     * http 远程服务端代理传输数据上报请求结果
+     *
+     * @see ProxyMessageTypeEnums#HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_
+     * @see AbstractHandleHttpReportClientProxyServerTransferResponseTypeAdvanced
+     */
+    HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_(ProxyMessageType.HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_, "远程服务端代理传输数据上报请求结果"),
     /**
      * http 远程服务端代理传输数据下发
      * @see AbstractHandleHttpDistributeClientProxyServerTransferTypeAdvanced
@@ -214,6 +221,13 @@ public enum ProxyMessageTypeEnums {
      * @see AbstractHandleSocksReportClientProxyServerTransferRequestTypeAdvanced
      */
     SOCKS_REPORT_CLIENT_PROXY_SERVER_TRANSFER_REQUEST_(ProxyMessageType.SOCKS_REPORT_CLIENT_PROXY_SERVER_TRANSFER_REQUEST_, "socket 上报客户端代理服务端请求"),
+    /**
+     * socket 上报客户端代理服务端结果
+     *
+     * @see ProxyMessageTypeEnums#SOCKS_REPORT_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_
+     * @see AbstractHandleSocksReportClientProxyServerTransferResponseTypeAdvanced
+     */
+    SOCKS_REPORT_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_(ProxyMessageType.SOCKS_REPORT_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_, "socket 上报客户端代理服务端结果"),
     /**
      * socket 下发客户端代理服务端数据返回
      *
@@ -344,6 +358,14 @@ public enum ProxyMessageTypeEnums {
      * @see AbstractHandleSocksDistributeServerProxyClientRequestTypeAdvanced
      */
     SOCKS_DISTRIBUTE_SERVER_PROXY_CLIENT_TRANSFER_REQUEST_(ProxyMessageType.SOCKS_DISTRIBUTE_SERVER_PROXY_CLIENT_TRANSFER_REQUEST_, "socket 下发客户端代理服务端 请求数据"),
+    /**
+     * socket 上报客户端代理服务端 请求
+     *
+     * @see ProxyMessageTypeEnums#SOCKS_REPORT_SERVER_PROXY_CLIENT_TRANSFER_REQUEST_
+     * @see AbstractHandleSocksReportServerProxyClientRequestTypeAdvanced
+     */
+    SOCKS_REPORT_SERVER_PROXY_CLIENT_TRANSFER_REQUEST_(ProxyMessageType.SOCKS_REPORT_SERVER_PROXY_CLIENT_TRANSFER_REQUEST_, "socket 上报客户端代理服务端 请求"),
+
     /**
      * socket 上报客户端代理服务端 结果返回
      *

wu-lazy-cloud-heartbeat-common/src/main/java/org/framework/lazy/cloud/network/heartbeat/common/utils/ChannelAttributeKeyUtils.java

@@ -16,6 +16,8 @@ public class ChannelAttributeKeyUtils {
     private static final AttributeKey<String> APP_KEY = AttributeKey.newInstance("appKey");
     private static final AttributeKey<String> APP_SECRET = AttributeKey.newInstance("appSecret");
     private static final AttributeKey<String> ORIGINAL_IP = AttributeKey.newInstance("originalIp");
+    private static final AttributeKey<String> TARGET_IP = AttributeKey.newInstance("targetIp");
+    private static final AttributeKey<Integer> TARGET_PORT = AttributeKey.newInstance("targetPort");
     private static final AttributeKey<Integer> OUT_FLOW = AttributeKey.newInstance("outFlow");
     private static final AttributeKey<Integer> IN_FLOW = AttributeKey.newInstance("inFlow");
 
@@ -259,6 +261,45 @@ public class ChannelAttributeKeyUtils {
         return channel.attr(ORIGINAL_IP).get();
     }
 
+
+    /**
+     * 为通道绑定 目标IP
+     *
+     * @param channel    通道
+     * @param targetIp 目标IP
+     */
+    public static void buildTargetIp(Channel channel, String targetIp) {
+        channel.attr(TARGET_IP).set(targetIp);
+    }
+
+    /**
+     * 获取 通道中 目标IP
+     *
+     * @param channel 通道
+     */
+    public static String getTargetIp(Channel channel) {
+        return channel.attr(TARGET_IP).get();
+    }
+
+    /**
+     * 为通道绑定 目标端口
+     *
+     * @param channel    通道
+     * @param targetIp 目标端口
+     */
+    public static void buildTargetPort(Channel channel, Integer targetPort) {
+        channel.attr(TARGET_PORT).set(targetPort);
+    }
+
+    /**
+     * 获取 通道中 目标端口
+     *
+     * @param channel 通道
+     */
+    public static Integer getTargetPort(Channel channel) {
+        return channel.attr(TARGET_PORT).get();
+    }
+
     /**
      * 为通道绑定 请求地址类型
      *

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/config/ServerAutoConfiguration.java

@@ -2,6 +2,7 @@ package org.framework.lazy.cloud.network.heartbeat.server.config;
 
 
 import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelFlowAdapter;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelProxyFlowAdapter;
 import org.framework.lazy.cloud.network.heartbeat.server.netty.permeate.tcp.advanced.*;
 import org.framework.lazy.cloud.network.heartbeat.server.netty.permeate.udp.advanced.*;
 import org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.http.advanced.*;
@@ -202,18 +203,24 @@ public class ServerAutoConfiguration {
     @Configuration()
     static class ServerHttpProxyConfiguration {
         @Bean
-        public ServerHandleHttpReportClientProxyServerProxyTransferTypeAdvanced serverHandleHttpReportClientProxyServerProxyTransferTypeAdvanced() {
+        public ServerHandleHttpReportClientProxyServerProxyTransferRequestTypeAdvanced serverHandleHttpReportClientProxyServerProxyTransferTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
-            return new ServerHandleHttpReportClientProxyServerProxyTransferTypeAdvanced();
+            return new ServerHandleHttpReportClientProxyServerProxyTransferRequestTypeAdvanced(channelProxyFlowAdapter);
         }
 
+        @Bean
+        public ServerHandleHttpReportClientProxyServerProxyTransferResponseTypeAdvanced serverHandleHttpReportClientProxyServerProxyTransferResponseTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
+            return new ServerHandleHttpReportClientProxyServerProxyTransferResponseTypeAdvanced(channelProxyFlowAdapter);
+        }
+
+
         @Bean
         public ServerHandleHttpReportClientProxyServerTransferCloseTypeAdvanced serverHandleHttpReportClientProxyServerTransferCloseTypeAdvanced() {
             return new ServerHandleHttpReportClientProxyServerTransferCloseTypeAdvanced();
         }
 
         @Bean
-        public ServerHandleHttpReportClientProxyClientConnectTransferTypeAdvanced serverHandleHttpReportClientProxyClientConnectTransferTypeAdvanced() {
+        public ServerHandleHttpReportClientProxyClientConnectTransferTypeAdvanced serverHandleHttpReportClientProxyClientConnectTransferTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
-            return new ServerHandleHttpReportClientProxyClientConnectTransferTypeAdvanced();
+            return new ServerHandleHttpReportClientProxyClientConnectTransferTypeAdvanced(channelProxyFlowAdapter);
         }
 
         @Bean
@@ -222,8 +229,8 @@ public class ServerAutoConfiguration {
         }
 
         @Bean
-        public ServerHandleHttpReportClientProxyClientTransferResponseTypeAdvanced serverHandleHttpReportClientProxyClientTransferResponseTypeAdvanced() {
+        public ServerHandleHttpReportClientProxyClientTransferResponseTypeAdvanced serverHandleHttpReportClientProxyClientTransferResponseTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
-            return new ServerHandleHttpReportClientProxyClientTransferResponseTypeAdvanced();
+            return new ServerHandleHttpReportClientProxyClientTransferResponseTypeAdvanced(channelProxyFlowAdapter);
         }
         @Bean
         public ServerHandleHttpReportServerProxyClientTransferTypeAdvanced serverHandleHttpReportClientProxyServerProxyTransferTypeAdvanced1() {
@@ -235,13 +242,13 @@ public class ServerAutoConfiguration {
         }
 
         @Bean
-        public ServerHandleHttpReportServerProxyClientTransferChannelInitSuccessfulTypeAdvanced serverHandleHttpReportServerProxyClientTransferChannelInitSuccessfulTypeAdvanced() {
+        public ServerHandleHttpReportServerProxyClientTransferChannelInitSuccessfulTypeAdvanced serverHandleHttpReportServerProxyClientTransferChannelInitSuccessfulTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
-            return new ServerHandleHttpReportServerProxyClientTransferChannelInitSuccessfulTypeAdvanced();
+            return new ServerHandleHttpReportServerProxyClientTransferChannelInitSuccessfulTypeAdvanced(channelProxyFlowAdapter);
         }
 
         @Bean
-        public ServerHandleHttpReportServerProxyClientTransferResponseTypeAdvanced serverHandleHttpReportServerProxyClientTransferResponseTypeAdvanced() {
+        public ServerHandleHttpReportServerProxyClientTransferResponseTypeAdvanced serverHandleHttpReportServerProxyClientTransferResponseTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
-            return new ServerHandleHttpReportServerProxyClientTransferResponseTypeAdvanced();
+            return new ServerHandleHttpReportServerProxyClientTransferResponseTypeAdvanced(channelProxyFlowAdapter);
         }
 
         @Bean
@@ -261,8 +268,13 @@ public class ServerAutoConfiguration {
             return new ServerHandleSocksReportClientProxyServerTransferCloseTypeAdvanced();
         }
         @Bean
-        public ServerHandleSocksReportClientProxyServerTransferRequestTypeAdvanced serverHandleSocksReportClientProxyServerTransferRequestTypeAdvanced(){
+        public ServerHandleSocksReportClientProxyServerTransferRequestTypeAdvanced serverHandleSocksReportClientProxyServerTransferRequestTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
-            return new ServerHandleSocksReportClientProxyServerTransferRequestTypeAdvanced();
+            return new ServerHandleSocksReportClientProxyServerTransferRequestTypeAdvanced(channelProxyFlowAdapter);
+        }
+
+        @Bean
+        public ServerHandleSocksReportClientProxyServerTransferResponseTypeAdvanced serverHandleSocksReportClientProxyServerTransferResponseTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
+            return new ServerHandleSocksReportClientProxyServerTransferResponseTypeAdvanced(channelProxyFlowAdapter);
         }
 
         @Bean
@@ -276,8 +288,8 @@ public class ServerAutoConfiguration {
         }
 
         @Bean
-        public ServerHandleSocksReportClientProxyClientTransferRequestTypeAdvanced serverHandleSocksReportClientProxyClientTransferRequestTypeAdvanced() {
+        public ServerHandleSocksReportClientProxyClientTransferRequestTypeAdvanced serverHandleSocksReportClientProxyClientTransferRequestTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
-            return new ServerHandleSocksReportClientProxyClientTransferRequestTypeAdvanced();
+            return new ServerHandleSocksReportClientProxyClientTransferRequestTypeAdvanced(channelProxyFlowAdapter);
         }
 
         @Bean
@@ -285,8 +297,8 @@ public class ServerAutoConfiguration {
             return new ServerHandleSocksReportClientProxyClientOtherConnectionTransferSuccessTypeAdvanced();
         }
         @Bean
-        public ServerHandleSocksReportClientProxyClientTransferResponseTypeAdvanced serverHandleSocksReportClientProxyClientTransferResponseTypeAdvanced(){
+        public ServerHandleSocksReportClientProxyClientTransferResponseTypeAdvanced serverHandleSocksReportClientProxyClientTransferResponseTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter){
-            return new ServerHandleSocksReportClientProxyClientTransferResponseTypeAdvanced();
+            return new ServerHandleSocksReportClientProxyClientTransferResponseTypeAdvanced(channelProxyFlowAdapter);
         }
 
         @Bean
@@ -304,6 +316,10 @@ public class ServerAutoConfiguration {
             return new ServerHandleSocksReportServerProxyClientResponseTypeAdvanced();
         }
         @Bean
+        public  ServerHandleSocksReportServerProxyClientRequestTypeAdvanced serverHandleSocksReportServerProxyClientRequestTypeAdvanced(){
+            return new ServerHandleSocksReportServerProxyClientRequestTypeAdvanced();
+        }
+        @Bean
         public ServerHandleSocksReportServerProxyClientTransferCloseTypeAdvanced serverHandleSocksReportServerProxyClientTransferCloseTypeAdvanced(){
             return new ServerHandleSocksReportServerProxyClientTransferCloseTypeAdvanced();
         }

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/config/ServerFlowConfiguration.java

@@ -1,7 +1,11 @@
 package org.framework.lazy.cloud.network.heartbeat.server.config;
 
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelProxyFlowAdapter;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.proxy.HandleChannelProxyFlowAdvanced;
 import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.ServerHandlerInFlowHandler;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.proxy.ServerHandlerInProxyFlowHandler;
 import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.ServerHandlerOutFlowHandler;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.proxy.ServerHandlerOutProxyFlowHandler;
 import org.framework.lazy.cloud.network.heartbeat.server.properties.ServerNodeProperties;
 import org.framework.lazy.cloud.network.heartbeat.server.standalone.application.LazyVisitorPortFlowApplication;
 import org.springframework.beans.factory.config.BeanDefinition;
@@ -55,4 +59,43 @@ public class ServerFlowConfiguration {
         return new ChannelFlowAdapter(handleChannelFlowAdvancedList);
     }
 
+
+
+
+    /**
+     * 进口数据处理
+     *
+     * @return serverHandlerInProxyFlowHandler
+     */
+    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+    @Bean
+    public ServerHandlerInProxyFlowHandler serverHandlerInProxyFlowHandler(LazyVisitorPortFlowApplication lazyVisitorPortFlowApplication, ServerNodeProperties serverNodeProperties) {
+        return new ServerHandlerInProxyFlowHandler(lazyVisitorPortFlowApplication, serverNodeProperties);
+    }
+
+    /**
+     * 出口数据处理
+     *
+     * @return serverHandlerOutProxyFlowHandler
+     */
+    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+    @Bean
+    public ServerHandlerOutProxyFlowHandler serverHandlerOutProxyFlowHandler(LazyVisitorPortFlowApplication lazyVisitorPortFlowApplication, ServerNodeProperties serverNodeProperties) {
+        return new ServerHandlerOutProxyFlowHandler(lazyVisitorPortFlowApplication,serverNodeProperties);
+    }
+
+
+    /**
+     * 服务端流量适配器
+     *
+     * @param handleChannelProxyFlowAdvancedList 服务端流量适配者
+     * @return 服务端流量适配器
+     */
+    @ConditionalOnMissingBean(ChannelProxyFlowAdapter.class)
+    @Bean
+    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+    public ChannelProxyFlowAdapter channelProxyFlowAdapter(List<HandleChannelProxyFlowAdvanced> handleChannelProxyFlowAdvancedList) {
+        return new ChannelProxyFlowAdapter(handleChannelProxyFlowAdvancedList);
+    }
+
 }

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/flow/proxy/ServerChannelProxyFlow.java

@@ -0,0 +1,67 @@
+package org.framework.lazy.cloud.network.heartbeat.server.netty.flow.proxy;
+
+import lombok.Builder;
+import lombok.Data;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.proxy.ChannelProxyFlow;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ChannelFlowEnum;
+
+@Builder
+@Data
+public class ServerChannelProxyFlow implements ChannelProxyFlow {
+    private String clientId;
+    private String ip;
+    private Integer port;
+    private ChannelFlowEnum channelFlowEnum;
+    private Integer flow;
+
+    /**
+     * 通道客户端ID
+     *
+     * @return 通道客户端ID
+     */
+    @Override
+    public String clientId() {
+        return clientId;
+    }
+
+    /**
+     * ip
+     *
+     * @return
+     */
+    @Override
+    public String ip() {
+        return ip;
+    }
+
+    /**
+     * 通道使用的端口（服务端访客端口、客户端真实端口）
+     *
+     * @return 端口
+     */
+    @Override
+    public Integer port() {
+        return port;
+    }
+
+    /**
+     * 通道流量类型
+     *
+     * @return ChannelFlowEnum
+     * @see ChannelFlowEnum
+     */
+    @Override
+    public ChannelFlowEnum channelFlowEnum() {
+        return channelFlowEnum;
+    }
+
+    /**
+     * 流量
+     *
+     * @return 流量
+     */
+    @Override
+    public Integer flow() {
+        return flow;
+    }
+}

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/flow/proxy/ServerHandlerInProxyFlowHandler.java

@@ -0,0 +1,59 @@
+package org.framework.lazy.cloud.network.heartbeat.server.netty.flow.proxy;
+
+import io.netty.channel.Channel;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.proxy.AbstractHandleChannelProxyFlowAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.ChannelFlow;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.proxy.ChannelProxyFlow;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ChannelFlowEnum;
+import org.framework.lazy.cloud.network.heartbeat.server.properties.ServerNodeProperties;
+import org.framework.lazy.cloud.network.heartbeat.server.standalone.application.LazyVisitorPortFlowApplication;
+import org.framework.lazy.cloud.network.heartbeat.server.standalone.application.command.visitor.flow.LazyVisitorPortFlowStoryCommand;
+
+/**
+ * 进口流量处理
+ */
+public class ServerHandlerInProxyFlowHandler extends AbstractHandleChannelProxyFlowAdvanced {
+    private final LazyVisitorPortFlowApplication lazyVisitorPortFlowApplication;
+    private final ServerNodeProperties serverNodeProperties;
+
+    public ServerHandlerInProxyFlowHandler(LazyVisitorPortFlowApplication lazyVisitorPortFlowApplication, ServerNodeProperties serverNodeProperties) {
+        this.lazyVisitorPortFlowApplication = lazyVisitorPortFlowApplication;
+        this.serverNodeProperties = serverNodeProperties;
+    }
+
+    /**
+     * 处理是否支持这种类型
+     *
+     * @param channelProxyFlow 数据
+     * @return boolean
+     */
+    @Override
+    protected boolean doSupport(ChannelProxyFlow channelProxyFlow) {
+        if (serverNodeProperties.getEnableFlowControl()) {
+            return ChannelFlowEnum.IN_FLOW.equals(channelProxyFlow.channelFlowEnum());
+        } else {
+            return false;
+        }
+    }
+
+    /**
+     * 处理当前数据
+     *
+     * @param channel          当前通道
+     * @param channelProxyFlow 通道数据
+     */
+    @Override
+    protected void doHandler(Channel channel, ChannelProxyFlow channelProxyFlow) {
+        String clientId = channelProxyFlow.clientId();
+        Integer port = channelProxyFlow.port();
+        Integer flow = channelProxyFlow.flow();
+
+        // 进口流量处理
+        LazyVisitorPortFlowStoryCommand visitorPortFlow = new LazyVisitorPortFlowStoryCommand();
+        visitorPortFlow.setInFlow(flow);
+        visitorPortFlow.setClientId(clientId);
+        visitorPortFlow.setVisitorPort(port);
+        visitorPortFlow.setIsDeleted(false);
+        lazyVisitorPortFlowApplication.flowIncreaseStory(visitorPortFlow);
+    }
+}

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/flow/proxy/ServerHandlerOutProxyFlowHandler.java

@@ -0,0 +1,60 @@
+package org.framework.lazy.cloud.network.heartbeat.server.netty.flow.proxy;
+
+import io.netty.channel.Channel;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.proxy.AbstractHandleChannelProxyFlowAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.ChannelFlow;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.proxy.ChannelProxyFlow;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ChannelFlowEnum;
+import org.framework.lazy.cloud.network.heartbeat.server.properties.ServerNodeProperties;
+import org.framework.lazy.cloud.network.heartbeat.server.standalone.application.LazyVisitorPortFlowApplication;
+import org.framework.lazy.cloud.network.heartbeat.server.standalone.application.command.visitor.flow.LazyVisitorPortFlowStoryCommand;
+
+/**
+ * 出口流量处理
+ */
+public class ServerHandlerOutProxyFlowHandler extends AbstractHandleChannelProxyFlowAdvanced {
+    private final LazyVisitorPortFlowApplication lazyVisitorPortFlowApplication;
+    private final ServerNodeProperties serverNodeProperties;
+
+    public ServerHandlerOutProxyFlowHandler(LazyVisitorPortFlowApplication lazyVisitorPortFlowApplication, ServerNodeProperties serverNodeProperties) {
+        this.lazyVisitorPortFlowApplication = lazyVisitorPortFlowApplication;
+        this.serverNodeProperties = serverNodeProperties;
+    }
+
+    /**
+     * 处理是否支持这种类型
+     *
+     * @param channelProxyFlow 数据
+     * @return boolean
+     */
+    @Override
+    protected boolean doSupport(ChannelProxyFlow channelProxyFlow) {
+        if (serverNodeProperties.getEnableFlowControl()) {
+            return ChannelFlowEnum.OUT_FLOW.equals(channelProxyFlow.channelFlowEnum());
+        } else {
+            return false;
+        }
+    }
+
+    /**
+     * 处理当前数据
+     *
+     * @param channel     当前通道
+     * @param channelProxyFlow 通道数据
+     */
+    @Override
+    protected void doHandler(Channel channel, ChannelProxyFlow channelProxyFlow) {
+        String clientId = channelProxyFlow.clientId();
+        Integer port = channelProxyFlow.port();
+        Integer flow = channelProxyFlow.flow();
+
+        // 出口流量处理
+        LazyVisitorPortFlowStoryCommand visitorPortFlow = new LazyVisitorPortFlowStoryCommand();
+        visitorPortFlow.setOutFlow(flow);
+        visitorPortFlow.setClientId(clientId);
+        visitorPortFlow.setVisitorPort(port);
+        visitorPortFlow.setIsDeleted(false);
+        lazyVisitorPortFlowApplication.flowIncreaseStory(visitorPortFlow);
+
+    }
+}

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/http/NettyHttpClientProxyServerTransfer.java

@@ -43,6 +43,6 @@ public class NettyHttpClientProxyServerTransfer {
     /**
      * 是否是ssl
      */
-    private  boolean isSsl;
+    private boolean isSsl;
 
 }

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/http/advanced/ServerHandleHttpReportClientProxyClientConnectTransferTypeAdvanced.java

@@ -5,12 +5,17 @@ import io.netty.channel.Channel;
 import lombok.extern.slf4j.Slf4j;
 import org.framework.lazy.cloud.network.heartbeat.common.ChannelContext;
 import org.framework.lazy.cloud.network.heartbeat.common.NettyByteBuf;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelFlowAdapter;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelProxyFlowAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
 import org.framework.lazy.cloud.network.heartbeat.common.NettyTransferChannelContext;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyChannelContext;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.http.server.AbstractHandleHttpReportClientProxyClientConnectionTransferTypeAdvanced;
 import org.framework.lazy.cloud.network.heartbeat.common.constant.ProxyMessageType;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ChannelFlowEnum;
 import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.ServerChannelFlow;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.proxy.ServerChannelProxyFlow;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.context.annotation.Role;
 import org.springframework.stereotype.Component;
@@ -26,6 +31,12 @@ import org.springframework.stereotype.Component;
 public class ServerHandleHttpReportClientProxyClientConnectTransferTypeAdvanced
         extends AbstractHandleHttpReportClientProxyClientConnectionTransferTypeAdvanced<NettyProxyMsg> {
 
+    private final ChannelProxyFlowAdapter channelProxyFlowAdapter;
+
+    public ServerHandleHttpReportClientProxyClientConnectTransferTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
+        this.channelProxyFlowAdapter = channelProxyFlowAdapter;
+    }
+
 
     /**
      * 处理当前数据
@@ -65,6 +76,15 @@ public class ServerHandleHttpReportClientProxyClientConnectTransferTypeAdvanced
         }else {
             log.error("can not find target client:【】 channel",clientId);
         }
+        // 记录进口数据
+        ServerChannelProxyFlow serverChannelFlow = ServerChannelProxyFlow
+                .builder()
+                .channelFlowEnum(ChannelFlowEnum.IN_FLOW)
+                .port(Integer.parseInt(targetPortString))
+                .clientId(new String(clientId))
+                .flow(data.length)
+                .build();
+        channelProxyFlowAdapter.asyncHandler(channel, serverChannelFlow);
 
     }
 

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/http/advanced/ServerHandleHttpReportClientProxyClientTransferResponseTypeAdvanced.java

@@ -3,11 +3,16 @@ package org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.http.advan
 
 import io.netty.channel.Channel;
 import lombok.extern.slf4j.Slf4j;
-import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelFlowAdapter;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelProxyFlowAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyChannelContext;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.http.server.AbstractHandleHttpReportClientProxyClientTransferResponseTypeAdvanced;
 import org.framework.lazy.cloud.network.heartbeat.common.constant.ProxyMessageType;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ChannelFlowEnum;
 import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.ServerChannelFlow;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.proxy.ServerChannelProxyFlow;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.context.annotation.Role;
 import org.springframework.stereotype.Component;
@@ -23,6 +28,11 @@ import org.wu.framework.core.utils.ObjectUtils;
 @Component
 public class ServerHandleHttpReportClientProxyClientTransferResponseTypeAdvanced
         extends AbstractHandleHttpReportClientProxyClientTransferResponseTypeAdvanced<NettyProxyMsg> {
+    private final ChannelProxyFlowAdapter channelProxyFlowAdapter;
+
+    public ServerHandleHttpReportClientProxyClientTransferResponseTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
+        this.channelProxyFlowAdapter = channelProxyFlowAdapter;
+    }
 
 
     /**
@@ -37,12 +47,25 @@ public class ServerHandleHttpReportClientProxyClientTransferResponseTypeAdvanced
        // 将返回数据下发客户端
         Channel transferNextChannel = ChannelAttributeKeyUtils.getTransferNextChannel(channel);
 
+        String clientId = nettyProxyMsg.getClientIdString();
+        Integer visitorPort = nettyProxyMsg.getVisitorPortInt();
+        String visitorId = nettyProxyMsg.getVisitorIdString();
         if(ObjectUtils.isNotEmpty(transferNextChannel)) {
             log.info("目标客户端返回数据通过服务端下发到原始通道");
             NettyProxyMsg responseProxyMsg = new NettyProxyMsg();
             responseProxyMsg.setData(nettyProxyMsg.getData());
             responseProxyMsg.setType(ProxyMessageType.HTTP_DISTRIBUTE_CLIENT_PROXY_CLIENT_TRANSFER_RESPONSE_);
             transferNextChannel.writeAndFlush(responseProxyMsg);
+
+            // 记录出口数据
+            ServerChannelProxyFlow serverChannelFlow = ServerChannelProxyFlow
+                    .builder()
+                    .channelFlowEnum(ChannelFlowEnum.OUT_FLOW)
+                    .port(visitorPort)
+                    .clientId(clientId)
+                    .flow(responseProxyMsg.getData().length)
+                    .build();
+            channelProxyFlowAdapter.asyncHandler(channel, serverChannelFlow);
         }else {
             log.error("无法将数据下发给原始客户端");
         }

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/http/advanced/ServerHandleHttpReportClientProxyServerProxyTransferRequestTypeAdvanced.java

@@ -4,10 +4,14 @@ package org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.http.advan
 import io.netty.channel.Channel;
 import lombok.extern.slf4j.Slf4j;
 import org.framework.lazy.cloud.network.heartbeat.client.config.NettyClientProperties;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelProxyFlowAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.HandleChannelTypeAdvanced;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyChannelContext;
-import org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.http.server.AbstractHandleHttpReportClientProxyServerTransferTypeAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.http.server.AbstractHandleHttpReportClientProxyServerTransferRequestTypeAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ChannelFlowEnum;
+import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.proxy.ServerChannelProxyFlow;
 import org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.http.NettyHttpClientProxyServerTransfer;
 import org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.http.socket.NettyHttpClientProxyServerRealSocket;
 import org.springframework.beans.factory.config.BeanDefinition;
@@ -21,14 +25,19 @@ import java.util.List;
 
 /**
  * 服务端处理客户端代理到服务端的请求
- * HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_
+ * HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_REQUEST_
  */
 @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
 @Slf4j
 @Component
-public class ServerHandleHttpReportClientProxyServerProxyTransferTypeAdvanced
+public class ServerHandleHttpReportClientProxyServerProxyTransferRequestTypeAdvanced
-        extends AbstractHandleHttpReportClientProxyServerTransferTypeAdvanced<NettyProxyMsg> {
+        extends AbstractHandleHttpReportClientProxyServerTransferRequestTypeAdvanced<NettyProxyMsg> {
 
+    private final ChannelProxyFlowAdapter channelProxyFlowAdapter;
+
+    public ServerHandleHttpReportClientProxyServerProxyTransferRequestTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
+        this.channelProxyFlowAdapter = channelProxyFlowAdapter;
+    }
 
     /**
      * 处理当前数据
@@ -42,8 +51,13 @@ public class ServerHandleHttpReportClientProxyServerProxyTransferTypeAdvanced
         // 创建链接、发送数据
         String targetPortString = nettyProxyMsg.getTargetPortString();
         String targetIpString = nettyProxyMsg.getTargetIpString();
+        String clientIdString = nettyProxyMsg.getClientIdString();
         byte[] data = nettyProxyMsg.getData();
         byte[] visitorId = nettyProxyMsg.getVisitorId();
+
+
+        ChannelAttributeKeyUtils.buildClientId(channel, clientIdString);
+
         List<HandleChannelTypeAdvanced> handleChannelTypeAdvancedList = new ArrayList<>(SpringContextHolder.getApplicationContext().getBeansOfType(HandleChannelTypeAdvanced.class).values());
         NettyClientProperties nettyClientProperties = SpringContextHolder.getBean(NettyClientProperties.class);
         // 判断代理到客户端还是服务端
@@ -60,6 +74,16 @@ public class ServerHandleHttpReportClientProxyServerProxyTransferTypeAdvanced
         NettyHttpClientProxyServerRealSocket
                 .buildRealServer(nettyHttpClientProxyServerTransfer, channel,new String(visitorId));
 
+
+        // 记录进口数据
+        ServerChannelProxyFlow serverChannelFlow = ServerChannelProxyFlow
+                .builder()
+                .channelFlowEnum(ChannelFlowEnum.IN_FLOW)
+                .port(Integer.parseInt(targetPortString))
+                .clientId(clientIdString)
+                .flow(data.length)
+                .build();
+        channelProxyFlowAdapter.asyncHandler(channel, serverChannelFlow);
     }
 
 }

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/http/advanced/ServerHandleHttpReportClientProxyServerProxyTransferResponseTypeAdvanced.java

@@ -0,0 +1,69 @@
+package org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.http.advanced;
+
+
+import io.netty.channel.Channel;
+import lombok.extern.slf4j.Slf4j;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelProxyFlowAdapter;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyChannelContext;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.http.server.AbstractHandleHttpReportClientProxyServerTransferResponseTypeAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.constant.ProxyMessageType;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ChannelFlowEnum;
+import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.proxy.ServerChannelProxyFlow;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.context.annotation.Role;
+import org.springframework.stereotype.Component;
+
+
+/**
+ * 服务端处理客户端代理到服务端的请求
+ * HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_
+ */
+@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+@Slf4j
+@Component
+public class ServerHandleHttpReportClientProxyServerProxyTransferResponseTypeAdvanced
+        extends AbstractHandleHttpReportClientProxyServerTransferResponseTypeAdvanced<NettyProxyMsg> {
+
+    private final ChannelProxyFlowAdapter channelProxyFlowAdapter;
+
+    public ServerHandleHttpReportClientProxyServerProxyTransferResponseTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
+        this.channelProxyFlowAdapter = channelProxyFlowAdapter;
+    }
+
+    /**
+     * 处理当前数据
+     *
+     * @param nettyChannelContext 当前通道
+     * @param nettyProxyMsg        通道数据
+     */
+    @Override
+    public void doHandler(NettyChannelContext nettyChannelContext, NettyProxyMsg nettyProxyMsg) {
+        Channel channel = nettyChannelContext.channel();
+
+        byte[] data = nettyProxyMsg.getData();
+
+        Channel nextChannel = ChannelAttributeKeyUtils.getNextChannel(channel);
+        // 将数据返回给客户端
+        NettyProxyMsg responseNettyProxyMsg = new NettyProxyMsg();
+        responseNettyProxyMsg.setType(ProxyMessageType.HTTP_DISTRIBUTE_CLIENT_PROXY_SERVER_TRANSFER_);
+        responseNettyProxyMsg.setData(data);
+
+        nextChannel.writeAndFlush(responseNettyProxyMsg);
+
+        String clientId = ChannelAttributeKeyUtils.getClientId(nextChannel);
+        Integer visitorPort = ChannelAttributeKeyUtils.getVisitorPort(nextChannel);
+
+        // 记录出口数据
+        ServerChannelProxyFlow serverChannelFlow = ServerChannelProxyFlow
+                .builder()
+                .channelFlowEnum(ChannelFlowEnum.OUT_FLOW)
+                .port(visitorPort)
+                .clientId(clientId)
+                .flow(data.length)
+                .build();
+        channelProxyFlowAdapter.asyncHandler(channel, serverChannelFlow);
+    }
+
+}

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/http/advanced/ServerHandleHttpReportServerProxyClientTransferChannelInitSuccessfulTypeAdvanced.java

@@ -4,12 +4,15 @@ package org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.http.advan
 import io.netty.channel.Channel;
 import lombok.extern.slf4j.Slf4j;
 import org.framework.lazy.cloud.network.heartbeat.common.NettyByteBuf;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelProxyFlowAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
 import org.framework.lazy.cloud.network.heartbeat.common.NettyTransferChannelContext;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyChannelContext;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.http.server.AbstractHandleHttpReportServerProxyClientTransferChannelInitSuccessfulTypeAdvanced;
 import org.framework.lazy.cloud.network.heartbeat.common.constant.ProxyMessageType;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ChannelFlowEnum;
 import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.proxy.ServerChannelProxyFlow;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.context.annotation.Role;
 import org.springframework.stereotype.Component;
@@ -21,6 +24,11 @@ import org.springframework.stereotype.Component;
 public class ServerHandleHttpReportServerProxyClientTransferChannelInitSuccessfulTypeAdvanced
         extends AbstractHandleHttpReportServerProxyClientTransferChannelInitSuccessfulTypeAdvanced<NettyProxyMsg> {
 
+    private final ChannelProxyFlowAdapter channelProxyFlowAdapter;
+
+    public ServerHandleHttpReportServerProxyClientTransferChannelInitSuccessfulTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
+        this.channelProxyFlowAdapter = channelProxyFlowAdapter;
+    }
 
     /**
      * 处理当前数据
@@ -63,6 +71,16 @@ public class ServerHandleHttpReportServerProxyClientTransferChannelInitSuccessfu
 
         channel.writeAndFlush(clientConnectTagetNettyProxyMsg);
 
+        // 记录进口数据
+        ServerChannelProxyFlow serverChannelFlow = ServerChannelProxyFlow
+                .builder()
+                .channelFlowEnum(ChannelFlowEnum.IN_FLOW)
+                .port(Integer.parseInt(new String(msgClientTargetPort)))
+                .clientId(new String(clientId))
+                .flow(nettyByteBufData.getData().length)
+                .build();
+        channelProxyFlowAdapter.asyncHandler(channel, serverChannelFlow);
+
 
     }
 

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/http/advanced/ServerHandleHttpReportServerProxyClientTransferResponseTypeAdvanced.java

@@ -4,10 +4,13 @@ package org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.http.advan
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
 import lombok.extern.slf4j.Slf4j;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelProxyFlowAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyChannelContext;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.http.server.AbstractHandleHttpReportServerProxyClientTransferResponseTypeAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ChannelFlowEnum;
 import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.proxy.ServerChannelProxyFlow;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.context.annotation.Role;
 import org.springframework.stereotype.Component;
@@ -20,7 +23,11 @@ import org.wu.framework.core.utils.ObjectUtils;
 @Component
 public class ServerHandleHttpReportServerProxyClientTransferResponseTypeAdvanced
         extends AbstractHandleHttpReportServerProxyClientTransferResponseTypeAdvanced<NettyProxyMsg> {
+    private final ChannelProxyFlowAdapter channelProxyFlowAdapter;
 
+    public ServerHandleHttpReportServerProxyClientTransferResponseTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
+        this.channelProxyFlowAdapter = channelProxyFlowAdapter;
+    }
 
     /**
      * 处理当前数据
@@ -31,14 +38,27 @@ public class ServerHandleHttpReportServerProxyClientTransferResponseTypeAdvanced
     @Override
     public void doHandler(NettyChannelContext nettyChannelContext, NettyProxyMsg nettyProxyMsg) {
         Channel channel = nettyChannelContext.channel();
+
        // 将返回数据
         Channel nextChannel = ChannelAttributeKeyUtils.getNextChannel(channel);
+        String clientId = ChannelAttributeKeyUtils.getClientId(nextChannel);
+        Integer visitorPort = ChannelAttributeKeyUtils.getVisitorPort(nextChannel);
 
         if(ObjectUtils.isNotEmpty(nextChannel)) {
             log.info("目标客户端返回数据通过服务端下发到原始通道");
             ByteBuf buf = channel.config().getAllocator().buffer(nettyProxyMsg.getData().length);
             buf.writeBytes(nettyProxyMsg.getData());
             nextChannel.writeAndFlush(buf);
+
+            // 记录出口数据
+            ServerChannelProxyFlow serverChannelFlow = ServerChannelProxyFlow
+                    .builder()
+                    .channelFlowEnum(ChannelFlowEnum.OUT_FLOW)
+                    .port(visitorPort)
+                    .clientId(clientId)
+                    .flow(nettyProxyMsg.getData().length)
+                    .build();
+            channelProxyFlowAdapter.asyncHandler(channel, serverChannelFlow);
         }else {
             log.error("无法将数据下发给原始客户端");
         }

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/http/filter/NettyHttpClientProxyServerRealFilter.java

@@ -7,6 +7,7 @@ import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
 import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
 import lombok.extern.slf4j.Slf4j;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelTypeAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.decoder.TransferDecoder;
 import org.framework.lazy.cloud.network.heartbeat.common.encoder.TransferEncoder;
 import org.framework.lazy.cloud.network.heartbeat.common.filter.DebugChannelInitializer;
@@ -19,6 +20,7 @@ public class NettyHttpClientProxyServerRealFilter extends DebugChannelInitialize
 
     public NettyHttpClientProxyServerRealFilter(NettyHttpClientProxyServerTransfer nettyHttpClientProxyServerTransfer) {
         this.nettyHttpClientProxyServerTransfer = nettyHttpClientProxyServerTransfer;
+
     }
 
     /**
@@ -45,10 +47,12 @@ public class NettyHttpClientProxyServerRealFilter extends DebugChannelInitialize
             }
         }
 
+        ChannelTypeAdapter channelTypeAdapter = new ChannelTypeAdapter(nettyHttpClientProxyServerTransfer.getHandleChannelTypeAdvancedList());
+
         // 解码、编码
         pipeline.addLast(new TransferDecoder(Integer.MAX_VALUE, 1024 * 1024*10));
         pipeline.addLast(new TransferEncoder());
-        pipeline.addLast(new NettyHttpClientProxyServerRealHandler());
+        pipeline.addLast(new NettyHttpClientProxyServerRealHandler(channelTypeAdapter));
 
     }
 }

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/http/handler/NettyHttpClientProxyServerRealHandler.java

@@ -7,6 +7,7 @@ import io.netty.channel.ChannelOption;
 import io.netty.channel.SimpleChannelInboundHandler;
 import lombok.extern.slf4j.Slf4j;
 import org.framework.lazy.cloud.network.heartbeat.common.NettyByteBuf;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelTypeAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
 import org.framework.lazy.cloud.network.heartbeat.common.constant.ProxyMessageType;
 import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
@@ -17,6 +18,13 @@ import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeK
 @Slf4j
 public class NettyHttpClientProxyServerRealHandler extends SimpleChannelInboundHandler<NettyByteBuf> {
 
+
+    private final ChannelTypeAdapter channelTypeAdapter;
+
+    public NettyHttpClientProxyServerRealHandler(ChannelTypeAdapter channelTypeAdapter) {
+        this.channelTypeAdapter = channelTypeAdapter;
+    }
+
     @Override
     public void channelActive(ChannelHandlerContext ctx) throws Exception {
         // 根据访客ID 确认真实通道 读写打开
@@ -32,17 +40,16 @@ public class NettyHttpClientProxyServerRealHandler extends SimpleChannelInboundH
     @Override
     public void channelRead0(ChannelHandlerContext ctx,NettyByteBuf nettyByteBuf) {
 
-        Channel channel = ctx.channel();
         byte[] bytes = nettyByteBuf.getData();
         log.debug("bytes.length:{}",bytes.length);
         log.debug("客户端代理服务端接收服务端真实服务数据:{}", new String(bytes));
-        Channel nextChannel = ChannelAttributeKeyUtils.getNextChannel(channel);
+
         // 将数据返回给客户端
         NettyProxyMsg nettyProxyMsg = new NettyProxyMsg();
-        nettyProxyMsg.setType(ProxyMessageType.HTTP_DISTRIBUTE_CLIENT_PROXY_SERVER_TRANSFER_);
+        nettyProxyMsg.setType(ProxyMessageType.HTTP_REPORT_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_);
         nettyProxyMsg.setData(bytes);
 
-        nextChannel.writeAndFlush(nettyProxyMsg);
+        channelTypeAdapter.handler(ctx,nettyProxyMsg);
     }
 
 

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/http/socket/NettyHttpClientProxyServerRealSocket.java

@@ -7,6 +7,7 @@ import io.netty.channel.*;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioSocketChannel;
 import lombok.extern.slf4j.Slf4j;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelTypeAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
 import org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.http.NettyHttpClientProxyServerTransfer;
 import org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.http.filter.NettyHttpClientProxyServerRealFilter;

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/socks/advanced/ServerHandleSocksReportClientProxyClientConnectTransferTypeAdvanced.java

@@ -52,6 +52,8 @@ public class ServerHandleSocksReportClientProxyClientConnectTransferTypeAdvanced
         String msgVisitorId = new String(visitorId);
         ChannelAttributeKeyUtils.buildClientId(transferChannel, targetClientId);
         ChannelAttributeKeyUtils.buildVisitorId(transferChannel, msgVisitorId);
+        ChannelAttributeKeyUtils.buildTargetIp(transferChannel, host);
+        ChannelAttributeKeyUtils.buildTargetPort(transferChannel, port);
 
         NettyTransferChannelContext.pushVisitor(transferChannel, msgVisitorId);
 

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/socks/advanced/ServerHandleSocksReportClientProxyClientTransferRequestTypeAdvanced.java

@@ -3,12 +3,14 @@ package org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.socks.adva
 
 import io.netty.channel.Channel;
 import lombok.extern.slf4j.Slf4j;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelProxyFlowAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyChannelContext;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.socks.server.AbstractHandleSocksReportClientProxyClientTransferRequestTypeAdvanced;
 import org.framework.lazy.cloud.network.heartbeat.common.constant.ProxyMessageType;
-import org.framework.lazy.cloud.network.heartbeat.common.constant.TcpMessageType;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ChannelFlowEnum;
 import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.proxy.ServerChannelProxyFlow;
 import org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.socks.handler.NettySocksClientProxyServerRealHandler;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.context.annotation.Role;
@@ -21,6 +23,11 @@ import org.springframework.stereotype.Component;
 public class ServerHandleSocksReportClientProxyClientTransferRequestTypeAdvanced
         extends AbstractHandleSocksReportClientProxyClientTransferRequestTypeAdvanced<NettyProxyMsg> {
 
+    private final ChannelProxyFlowAdapter channelProxyFlowAdapter;
+
+    public ServerHandleSocksReportClientProxyClientTransferRequestTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
+        this.channelProxyFlowAdapter = channelProxyFlowAdapter;
+    }
 
     /**
      * 处理当前数据
@@ -33,15 +40,31 @@ public class ServerHandleSocksReportClientProxyClientTransferRequestTypeAdvanced
     public void doHandler(NettyChannelContext nettyChannelContext, NettyProxyMsg nettyProxyMsg) {
         Channel transferChannel = nettyChannelContext.channel();
 
+
+        Integer targetPort = ChannelAttributeKeyUtils.getTargetPort(transferChannel);
+        String targetIp = ChannelAttributeKeyUtils.getTargetIp(transferChannel);
+        String clientId = ChannelAttributeKeyUtils.getClientId(transferChannel);
+
         byte[] visitorId = nettyProxyMsg.getVisitorId();
         // 目标通道
         Channel nextChannel = ChannelAttributeKeyUtils.getTransferNextChannel(transferChannel);
+
         NettyProxyMsg requestMsg = new NettyProxyMsg();
         requestMsg.setVisitorId(visitorId);
         requestMsg.setData(nettyProxyMsg.getData());
         requestMsg.setType(ProxyMessageType.SOCKS_DISTRIBUTE_CLIENT_PROXY_CLIENT_TRANSFER_REQUEST_);
         if (nextChannel != null) {
             nextChannel.writeAndFlush(requestMsg);
+            // 记录进口数据
+            ServerChannelProxyFlow serverChannelFlow = ServerChannelProxyFlow
+                    .builder()
+                    .channelFlowEnum(ChannelFlowEnum.IN_FLOW)
+                    .port(targetPort)
+                    .ip(targetIp)
+                    .clientId(clientId)
+                    .flow(nettyProxyMsg.getData().length)
+                    .build();
+            channelProxyFlowAdapter.asyncHandler(transferChannel, serverChannelFlow);
         } else {
             log.error("can not find the  channel");
         }

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/socks/advanced/ServerHandleSocksReportClientProxyClientTransferResponseTypeAdvanced.java

@@ -3,11 +3,14 @@ package org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.socks.adva
 
 import io.netty.channel.Channel;
 import lombok.extern.slf4j.Slf4j;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelProxyFlowAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyChannelContext;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.socks.server.AbstractHandleSocksReportClientProxyClientTransferResponseTypeAdvanced;
 import org.framework.lazy.cloud.network.heartbeat.common.constant.ProxyMessageType;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ChannelFlowEnum;
 import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.proxy.ServerChannelProxyFlow;
 import org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.socks.handler.NettySocksClientProxyServerRealHandler;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.context.annotation.Role;
@@ -20,6 +23,11 @@ import org.springframework.stereotype.Component;
 public class ServerHandleSocksReportClientProxyClientTransferResponseTypeAdvanced
         extends AbstractHandleSocksReportClientProxyClientTransferResponseTypeAdvanced<NettyProxyMsg> {
 
+    private final ChannelProxyFlowAdapter channelProxyFlowAdapter;
+
+    public ServerHandleSocksReportClientProxyClientTransferResponseTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
+        this.channelProxyFlowAdapter = channelProxyFlowAdapter;
+    }
 
     /**
      * 处理当前数据
@@ -39,10 +47,23 @@ public class ServerHandleSocksReportClientProxyClientTransferResponseTypeAdvance
         responseMsg.setVisitorId(visitorId);
         responseMsg.setData(nettyProxyMsg.getData());
 
+        String targetIp = ChannelAttributeKeyUtils.getTargetIp(nextChannel);
+        Integer targetPort = ChannelAttributeKeyUtils.getTargetPort(nextChannel);
+        String clientId = ChannelAttributeKeyUtils.getClientId(nextChannel);
 
         responseMsg.setType(ProxyMessageType.SOCKS_DISTRIBUTE_CLIENT_PROXY_CLIENT_TRANSFER_RESPONSE_);
         if (nextChannel != null) {
             nextChannel.writeAndFlush(responseMsg);
+            // 记录出口数据
+            ServerChannelProxyFlow serverChannelFlow = ServerChannelProxyFlow
+                    .builder()
+                    .channelFlowEnum(ChannelFlowEnum.OUT_FLOW)
+                    .ip(targetIp)
+                    .port(targetPort)
+                    .clientId(clientId)
+                    .flow(nettyProxyMsg.getData().length)
+                    .build();
+            channelProxyFlowAdapter.asyncHandler(nextChannel, serverChannelFlow);
         }else {
             log.error("can not find the channel");
         }

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/socks/advanced/ServerHandleSocksReportClientProxyServerConnectTransferTypeAdvanced.java

@@ -9,7 +9,9 @@ import io.netty.handler.codec.socksx.v5.DefaultSocks5CommandResponse;
 import io.netty.handler.codec.socksx.v5.Socks5AddressType;
 import io.netty.handler.codec.socksx.v5.Socks5CommandStatus;
 import lombok.extern.slf4j.Slf4j;
+import org.framework.lazy.cloud.network.heartbeat.client.config.NettyClientProperties;
 import org.framework.lazy.cloud.network.heartbeat.common.NettyTransferChannelContext;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelTypeAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyChannelContext;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.socks.server.AbstractHandleSocksReportClientProxyServerConnectionTransferTypeAdvanced;
@@ -21,6 +23,7 @@ import org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.socks.handl
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.context.annotation.Role;
 import org.springframework.stereotype.Component;
+import org.wu.framework.spring.utils.SpringContextHolder;
 
 import java.net.InetSocketAddress;
 
@@ -51,10 +54,14 @@ public class ServerHandleSocksReportClientProxyServerConnectTransferTypeAdvanced
         String msgVisitorId = new String(visitorId);
         ChannelAttributeKeyUtils.buildClientId(transferChannel, clientId);
         ChannelAttributeKeyUtils.buildVisitorId(transferChannel, msgVisitorId);
+        ChannelAttributeKeyUtils.buildTargetIp(transferChannel, host);
+        ChannelAttributeKeyUtils.buildTargetPort(transferChannel, port);
 
         NettyTransferChannelContext.pushVisitor(transferChannel, msgVisitorId);
         Socks5AddressType socks5AddressType = Socks5AddressType.valueOf(data[0]);
 
+
+        ChannelTypeAdapter channelTypeAdapter = SpringContextHolder.getBean(ChannelTypeAdapter.class);
         // 创建真实代理链接
         EventLoopGroup group = EventLoopGroupFactory.createClientWorkGroup();
         Bootstrap b = new Bootstrap();
@@ -65,7 +72,7 @@ public class ServerHandleSocksReportClientProxyServerConnectTransferTypeAdvanced
                     @Override
                     protected void initChannel(SocketChannel ch) {
                         ch.pipeline().addLast(new TransferDecoder(Integer.MAX_VALUE, 1024 * 1024 * 10));
-                        ch.pipeline().addLast(new NettySocksClientProxyServerRealHandler());
+                        ch.pipeline().addLast(new NettySocksClientProxyServerRealHandler(channelTypeAdapter));
                     }
                 });
         log.info("准备连接目标服务器，ip = {},port = {}", host, port);

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/socks/advanced/ServerHandleSocksReportClientProxyServerTransferRequestTypeAdvanced.java

@@ -4,10 +4,13 @@ package org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.socks.adva
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
 import lombok.extern.slf4j.Slf4j;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelProxyFlowAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyChannelContext;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.socks.server.AbstractHandleSocksReportClientProxyServerTransferRequestTypeAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ChannelFlowEnum;
 import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.proxy.ServerChannelProxyFlow;
 import org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.socks.handler.NettySocksClientProxyServerRealHandler;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.context.annotation.Role;
@@ -22,6 +25,12 @@ public class ServerHandleSocksReportClientProxyServerTransferRequestTypeAdvanced
         extends AbstractHandleSocksReportClientProxyServerTransferRequestTypeAdvanced<NettyProxyMsg> {
 
 
+    private final ChannelProxyFlowAdapter channelProxyFlowAdapter;
+
+    public ServerHandleSocksReportClientProxyServerTransferRequestTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
+        this.channelProxyFlowAdapter = channelProxyFlowAdapter;
+    }
+
     /**
      * 处理当前数据
      *
@@ -32,6 +41,10 @@ public class ServerHandleSocksReportClientProxyServerTransferRequestTypeAdvanced
     @Override
     public void doHandler(NettyChannelContext nettyChannelContext, NettyProxyMsg nettyProxyMsg) {
         Channel transferChannel = nettyChannelContext.channel();
+
+        String targetIp = ChannelAttributeKeyUtils.getTargetIp(transferChannel);
+        Integer targetPort = ChannelAttributeKeyUtils.getTargetPort(transferChannel);
+        String clientId = ChannelAttributeKeyUtils.getClientId(transferChannel);
         // 目标通道
         Channel nextChannel = ChannelAttributeKeyUtils.getNextChannel(transferChannel);
         // 目标数据发送
@@ -39,6 +52,17 @@ public class ServerHandleSocksReportClientProxyServerTransferRequestTypeAdvanced
             ByteBuf buf = nextChannel.config().getAllocator().buffer(nettyProxyMsg.getData().length);
             buf.writeBytes(nettyProxyMsg.getData());
             nextChannel.writeAndFlush(buf);
+
+            // 记录进口数据
+            ServerChannelProxyFlow serverChannelFlow = ServerChannelProxyFlow
+                    .builder()
+                    .channelFlowEnum(ChannelFlowEnum.IN_FLOW)
+                    .port(targetPort)
+                    .ip(targetIp)
+                    .clientId(clientId)
+                    .flow(nettyProxyMsg.getData().length)
+                    .build();
+            channelProxyFlowAdapter.asyncHandler(transferChannel, serverChannelFlow);
         } else {
             log.error("当前目标通道已经关闭或者不存在");
         }

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/socks/advanced/ServerHandleSocksReportClientProxyServerTransferResponseTypeAdvanced.java

@@ -0,0 +1,75 @@
+package org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.socks.advanced;
+
+
+import io.netty.channel.Channel;
+import io.netty.util.ReferenceCountUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelProxyFlowAdapter;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyChannelContext;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.http.server.AbstractHandleHttpReportClientProxyServerTransferResponseTypeAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.socks.server.AbstractHandleSocksReportClientProxyServerTransferResponseTypeAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.constant.ProxyMessageType;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ChannelFlowEnum;
+import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.proxy.ServerChannelProxyFlow;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.socks.handler.NettySocksClientProxyServerRealHandler;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.context.annotation.Role;
+import org.springframework.stereotype.Component;
+
+
+@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+@Slf4j
+@Component
+public class ServerHandleSocksReportClientProxyServerTransferResponseTypeAdvanced
+        extends AbstractHandleSocksReportClientProxyServerTransferResponseTypeAdvanced<NettyProxyMsg> {
+
+
+    private final ChannelProxyFlowAdapter channelProxyFlowAdapter;
+
+    public ServerHandleSocksReportClientProxyServerTransferResponseTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
+        this.channelProxyFlowAdapter = channelProxyFlowAdapter;
+    }
+
+    /**
+     * 处理当前数据
+     *
+     * @param nettyChannelContext 当前通道
+     * @param nettyProxyMsg       通道数据
+     * @see NettySocksClientProxyServerRealHandler
+     */
+    @Override
+    public void doHandler(NettyChannelContext nettyChannelContext, NettyProxyMsg nettyProxyMsg) {
+        Channel channel = nettyChannelContext.channel();
+        byte[] data = nettyProxyMsg.getData();
+        Channel nextChannel = ChannelAttributeKeyUtils.getNextChannel(channel);
+
+        String targetIp = ChannelAttributeKeyUtils.getTargetIp(nextChannel);
+        String clientId = ChannelAttributeKeyUtils.getClientId(nextChannel);
+        Integer targetPort = ChannelAttributeKeyUtils.getTargetPort(nextChannel);
+        if (nextChannel.isActive()) {
+            // 将数据返回给客户端
+            NettyProxyMsg nettyProxyMsgResponse = new NettyProxyMsg();
+            nettyProxyMsgResponse.setType(ProxyMessageType.SOCKS_DISTRIBUTE_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_);
+            nettyProxyMsgResponse.setData(data);
+
+            nextChannel.writeAndFlush(nettyProxyMsgResponse);
+
+            // 记录出口数据
+            ServerChannelProxyFlow serverChannelFlow = ServerChannelProxyFlow
+                    .builder()
+                    .channelFlowEnum(ChannelFlowEnum.OUT_FLOW)
+                    .ip(targetIp)
+                    .port(targetPort)
+                    .clientId(clientId)
+                    .flow(nettyProxyMsg.getData().length)
+                    .build();
+            channelProxyFlowAdapter.asyncHandler(nextChannel, serverChannelFlow);
+        } else {
+            log.info("释放内存");
+            ReferenceCountUtil.release(nettyProxyMsg);
+        }
+    }
+
+}

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/socks/advanced/ServerHandleSocksReportServerProxyClientConnectionSuccessTypeAdvanced.java

@@ -8,6 +8,7 @@ import io.netty.handler.codec.socksx.v5.Socks5CommandRequestDecoder;
 import io.netty.handler.codec.socksx.v5.Socks5CommandStatus;
 import lombok.extern.slf4j.Slf4j;
 import org.framework.lazy.cloud.network.heartbeat.common.NettyTransferChannelContext;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelTypeAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyChannelContext;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.socks.server.AbstractHandleSocksReportServerProxyClientConnectionSuccessTypeAdvanced;
@@ -20,6 +21,7 @@ import org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.socks.handl
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.context.annotation.Role;
 import org.springframework.stereotype.Component;
+import org.wu.framework.spring.utils.SpringContextHolder;
 
 
 @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
@@ -49,6 +51,8 @@ public class ServerHandleSocksReportServerProxyClientConnectionSuccessTypeAdvanc
         ChannelAttributeKeyUtils.buildVisitorId(transferChannel,visitorId);
         ChannelAttributeKeyUtils.buildClientId(transferChannel,clientId);
 
+
+        ChannelTypeAdapter channelTypeAdapter = SpringContextHolder.getBean(ChannelTypeAdapter.class);
         // 根据传输通道获取代理通道
         Channel nextChannel = ChannelAttributeKeyUtils.getNextChannel(transferChannel);
 
@@ -58,7 +62,7 @@ public class ServerHandleSocksReportServerProxyClientConnectionSuccessTypeAdvanc
         // TODO bug fix
         nextChannel.pipeline().addLast(new TransferDecoder(Integer.MAX_VALUE, 1024 * 1024*10));
         // 请求数据开始上报
-        nextChannel.pipeline().addLast(new NettySocksServerProxyClientVisitorInboundHandler());
+        nextChannel.pipeline().addLast(new NettySocksServerProxyClientVisitorInboundHandler(channelTypeAdapter));
         DefaultSocks5CommandResponse commandResponse =
                 new DefaultSocks5CommandResponse(Socks5CommandStatus.SUCCESS, socks5AddressType);
         nextChannel.writeAndFlush(commandResponse);

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/socks/advanced/ServerHandleSocksReportServerProxyClientRequestTypeAdvanced.java

@@ -0,0 +1,61 @@
+package org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.socks.advanced;
+
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.util.ReferenceCountUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelProxyFlowAdapter;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyChannelContext;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.socks.server.AbstractHandleSocksReportServerProxyClientRequestTypeAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.socks.server.AbstractHandleSocksReportServerProxyClientResponseTypeAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.constant.ProxyMessageType;
+import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.socks.handler.NettySocksClientProxyServerRealHandler;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.context.annotation.Role;
+import org.springframework.stereotype.Component;
+import org.wu.framework.core.utils.ObjectUtils;
+
+
+@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+@Slf4j
+@Component
+public class ServerHandleSocksReportServerProxyClientRequestTypeAdvanced
+        extends AbstractHandleSocksReportServerProxyClientRequestTypeAdvanced<NettyProxyMsg> {
+    private final ChannelProxyFlowAdapter channelProxyFlowAdapter;
+
+    public ServerHandleSocksReportServerProxyClientRequestTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
+        this.channelProxyFlowAdapter = channelProxyFlowAdapter;
+    }
+
+    /**
+     * 处理当前数据
+     *
+     * @param nettyChannelContext 当前通道
+     * @param nettyProxyMsg       通道数据
+     * @see NettySocksClientProxyServerRealHandler
+     */
+    @Override
+    public void doHandler(NettyChannelContext nettyChannelContext, NettyProxyMsg nettyProxyMsg) {
+
+        Channel channel = nettyChannelContext.channel();
+
+
+        Channel nextChannel = ChannelAttributeKeyUtils.getNextChannel(channel);
+
+        if (nextChannel.isActive()) {
+            // 下发数据到服务端
+            NettyProxyMsg nettyProxyMsgRequest = new NettyProxyMsg();
+            nettyProxyMsgRequest.setType(ProxyMessageType.SOCKS_DISTRIBUTE_SERVER_PROXY_CLIENT_TRANSFER_REQUEST_);
+
+            nettyProxyMsgRequest.setData(nettyProxyMsg.getData());
+            nextChannel.writeAndFlush(nettyProxyMsgRequest);
+        } else {
+            log.info("释放内存");
+            ReferenceCountUtil.release(nettyProxyMsg);
+        }
+    }
+
+}

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/socks/advanced/ServerHandleSocksReportServerProxyClientResponseTypeAdvanced.java

@@ -4,6 +4,7 @@ package org.framework.lazy.cloud.network.heartbeat.server.netty.proxy.socks.adva
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
 import lombok.extern.slf4j.Slf4j;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelProxyFlowAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyChannelContext;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.proxy.socks.server.AbstractHandleSocksReportServerProxyClientResponseTypeAdvanced;
@@ -22,6 +23,11 @@ import org.wu.framework.core.utils.ObjectUtils;
 public class ServerHandleSocksReportServerProxyClientResponseTypeAdvanced
         extends AbstractHandleSocksReportServerProxyClientResponseTypeAdvanced<NettyProxyMsg> {
 
+    private final ChannelProxyFlowAdapter channelProxyFlowAdapter;
+
+    public ServerHandleSocksReportServerProxyClientResponseTypeAdvanced(ChannelProxyFlowAdapter channelProxyFlowAdapter) {
+        this.channelProxyFlowAdapter = channelProxyFlowAdapter;
+    }
 
     /**
      * 处理当前数据

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/socks/advanced/ServerHandleSocksReportServerProxyClientTypeAdvanced.java

@@ -38,6 +38,8 @@ public class ServerHandleSocksReportServerProxyClientTypeAdvanced
         String host = nettyProxyMsg.getTargetIpString();
         Integer port = Integer.parseInt(nettyProxyMsg.getTargetPortString());
 
+
+
         byte[] visitorId = nettyProxyMsg.getVisitorId();
         byte[] targetClientId = nettyProxyMsg.getClientId();
         Socks5AddressType socks5AddressType = nettySocketChannelContext.getSocks5AddressType();

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/socks/handler/NettySocksClientProxyServerRealHandler.java

@@ -8,6 +8,7 @@ import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.util.ReferenceCountUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.framework.lazy.cloud.network.heartbeat.common.NettyByteBuf;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelTypeAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
 import org.framework.lazy.cloud.network.heartbeat.common.constant.ProxyMessageType;
 import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
@@ -16,6 +17,11 @@ import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeK
 @Slf4j
 public class NettySocksClientProxyServerRealHandler extends SimpleChannelInboundHandler<NettyByteBuf> {
 
+    private final ChannelTypeAdapter channelTypeAdapter;
+
+    public NettySocksClientProxyServerRealHandler(ChannelTypeAdapter channelTypeAdapter) {
+        this.channelTypeAdapter = channelTypeAdapter;
+    }
 
 
     @Override
@@ -27,22 +33,16 @@ public class NettySocksClientProxyServerRealHandler extends SimpleChannelInbound
     public void channelRead0(ChannelHandlerContext ctx, NettyByteBuf nettyByteBuf) throws Exception {
         log.trace("开始写回客户端数据");
         // 结果下发
-        Channel channel = ctx.channel();
         byte[] bytes = nettyByteBuf.getData();
         log.debug("bytes.length:{}", bytes.length);
         log.debug("客户端socks代理服务端接收服务端真实服务数据:{}", new String(bytes));
-        Channel nextChannel = ChannelAttributeKeyUtils.getNextChannel(channel);
-        if (nextChannel.isActive()) {
-            // 将数据返回给客户端
-            NettyProxyMsg nettyProxyMsgResponse = new NettyProxyMsg();
-            nettyProxyMsgResponse.setType(ProxyMessageType.SOCKS_DISTRIBUTE_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_);
-            nettyProxyMsgResponse.setData(bytes);
 
-            nextChannel.writeAndFlush(nettyProxyMsgResponse);
+        // 将数据返回给客户端
-        } else {
+        NettyProxyMsg nettyProxyMsg = new NettyProxyMsg();
-            log.info("释放内存");
+        nettyProxyMsg.setType(ProxyMessageType.SOCKS_REPORT_CLIENT_PROXY_SERVER_TRANSFER_RESPONSE_);
-            ReferenceCountUtil.release(nettyByteBuf);
+        nettyProxyMsg.setData(bytes);
-        }
+
+        channelTypeAdapter.handler(ctx,nettyProxyMsg);
     }
 
     @Override

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/socks/handler/NettySocksServerProxyClientVisitorHandler.java

@@ -26,7 +26,6 @@ public class NettySocksServerProxyClientVisitorHandler extends SimpleChannelInbo
     @Override
     public void channelRead0(ChannelHandlerContext ctx, NettyProxyMsg nettyProxyMsg) throws Exception {
         channelTypeAdapter.handler(ctx, nettyProxyMsg);
-
     }
 
     @Override

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/proxy/socks/handler/NettySocksServerProxyClientVisitorInboundHandler.java

@@ -6,9 +6,9 @@ import io.netty.channel.Channel;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
-import io.netty.util.ReferenceCountUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.framework.lazy.cloud.network.heartbeat.common.NettyByteBuf;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelTypeAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.payload.NettyProxyMsg;
 import org.framework.lazy.cloud.network.heartbeat.common.constant.ProxyMessageType;
 import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
@@ -17,7 +17,11 @@ import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeK
 @Slf4j
 public class NettySocksServerProxyClientVisitorInboundHandler extends SimpleChannelInboundHandler<NettyByteBuf> {
 
+    private final ChannelTypeAdapter channelTypeAdapter;
 
+    public NettySocksServerProxyClientVisitorInboundHandler(ChannelTypeAdapter channelTypeAdapter) {
+        this.channelTypeAdapter = channelTypeAdapter;
+    }
 
 
     @Override
@@ -33,19 +37,13 @@ public class NettySocksServerProxyClientVisitorInboundHandler extends SimpleChan
         byte[] bytes = nettyByteBuf.getData();
         log.debug("bytes.length:{}",bytes.length);
         log.debug("服务端代理客户端,socks本地接收请求数据:{}", new String(bytes));
-        Channel nextChannel = ChannelAttributeKeyUtils.getNextChannel(channel);
 
-        if (nextChannel.isActive()) {
+        // 将数据返回给客户端
-            // 下发数据到服务端
+        NettyProxyMsg nettyProxyMsg = new NettyProxyMsg();
-            NettyProxyMsg nettyProxyMsg = new NettyProxyMsg();
+        nettyProxyMsg.setType(ProxyMessageType.SOCKS_REPORT_SERVER_PROXY_CLIENT_TRANSFER_REQUEST_);
-            nettyProxyMsg.setType(ProxyMessageType.SOCKS_DISTRIBUTE_SERVER_PROXY_CLIENT_TRANSFER_REQUEST_);
+        nettyProxyMsg.setData(bytes);
 
-            nettyProxyMsg.setData(nettyByteBuf.getData());
+        channelTypeAdapter.handler(ctx, nettyProxyMsg);
-            nextChannel.writeAndFlush(nettyProxyMsg);
-        } else {
-            log.info("释放内存");
-            ReferenceCountUtil.release(nettyByteBuf);
-        }
     }
 
     @Override