From 80a7ca7844d96ddcd3b6951c96c969894fb760e8 Mon Sep 17 00:00:00 2001 From: aoshiguchen <1052045476@qq.com> Date: Tue, 30 Aug 2022 16:00:26 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9C=8D=E5=8A=A1=E7=AB=AF=E4=BB=A3=E7=90=86?= =?UTF-8?q?=E9=80=BB=E8=BE=91=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../neutrino/proxy/core/ChannelAttribute.java | 62 ++++++ .../proxy/core/ServerChannelHandler.java | 12 +- .../server/proxy/core/UserChannelHandler.java | 25 ++- .../proxy/domain/CmdChannelAttachInfo.java | 55 +++++ .../proxy/domain/UserChannelAttachInfo.java | 37 ++++ .../handler/ProxyMessageAuthHandler.java | 8 +- .../handler/ProxyMessageConnectHandler.java | 9 +- .../ProxyMessageDisconnectHandler.java | 18 +- .../server/util/ProxyChannelManager.java | 200 ------------------ .../neutrino/proxy/server/util/ProxyUtil.java | 172 +++++++++++++++ 10 files changed, 373 insertions(+), 225 deletions(-) create mode 100644 neutrino-proxy-core/src/main/java/fun/asgc/neutrino/proxy/core/ChannelAttribute.java create mode 100644 neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/domain/CmdChannelAttachInfo.java create mode 100644 neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/domain/UserChannelAttachInfo.java delete mode 100644 neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/util/ProxyChannelManager.java diff --git a/neutrino-proxy-core/src/main/java/fun/asgc/neutrino/proxy/core/ChannelAttribute.java b/neutrino-proxy-core/src/main/java/fun/asgc/neutrino/proxy/core/ChannelAttribute.java new file mode 100644 index 00000000..8216cca0 --- /dev/null +++ b/neutrino-proxy-core/src/main/java/fun/asgc/neutrino/proxy/core/ChannelAttribute.java @@ -0,0 +1,62 @@ +/** + * Copyright (c) 2022 aoshiguchen + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ +package fun.asgc.neutrino.proxy.core; + +import io.netty.channel.Channel; +import io.netty.util.AttributeKey; + +import java.util.HashMap; + +/** + * + * @author: aoshiguchen + * @date: 2022/8/30 + */ +public class ChannelAttribute extends HashMap { + public static ChannelAttribute create() { + return new ChannelAttribute(); + } + + public static ChannelAttribute of(String k, Object v) { + ChannelAttribute channelAttr = create(); + channelAttr.put(k, v); + + return channelAttr; + } + + public ChannelAttribute set(String k, Object v) { + super.put(k, v); + return this; + } + + public T get(String key) { + return (T)super.get(key); + } + + public String getString(String k) { + return String.valueOf(this.get(k)); + } + + public Long getLong(String k) { + return Long.valueOf(String.valueOf(this.get(k))); + } +} diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/core/ServerChannelHandler.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/core/ServerChannelHandler.java index 8858a309..0ca45176 100644 --- a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/core/ServerChannelHandler.java +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/core/ServerChannelHandler.java @@ -27,7 +27,7 @@ import fun.asgc.neutrino.core.base.Dispatcher; import fun.asgc.neutrino.core.util.BeanManager; import fun.asgc.neutrino.core.util.LockUtil; import fun.asgc.neutrino.proxy.core.*; -import fun.asgc.neutrino.proxy.server.util.ProxyChannelManager; +import fun.asgc.neutrino.proxy.server.util.ProxyUtil; import io.netty.buffer.Unpooled; import io.netty.channel.*; @@ -74,16 +74,20 @@ public class ServerChannelHandler extends SimpleChannelInboundHandler { } else { byte[] bytes = new byte[buf.readableBytes()]; buf.readBytes(bytes); - String userId = ProxyChannelManager.getUserChannelUserId(userChannel); +// String userId = ProxyChannelManager.getUserChannelUserId(userChannel); + String userId = ProxyUtil.getUserChannelUserId(userChannel); proxyChannel.writeAndFlush(ProxyMessage.buildTransferMessage(userId, bytes)); } } @@ -73,7 +73,8 @@ public class UserChannelHandler extends SimpleChannelInboundHandler { public void channelActive(ChannelHandlerContext ctx) throws Exception { Channel userChannel = ctx.channel(); InetSocketAddress sa = (InetSocketAddress) userChannel.localAddress(); - Channel cmdChannel = ProxyChannelManager.getCmdChannel(sa.getPort()); +// Channel cmdChannel = ProxyChannelManager.getCmdChannel(sa.getPort()); + Channel cmdChannel = ProxyUtil.getCmdChannelByServerPort(sa.getPort()); if (cmdChannel == null) { // 该端口还没有代理客户端 @@ -83,7 +84,8 @@ public class UserChannelHandler extends SimpleChannelInboundHandler { String lanInfo = ProxyUtil.getClientLanInfoByServerPort(sa.getPort()); // 用户连接到代理服务器时，设置用户连接不可读，等待代理后端服务器连接成功后再改变为可读状态 userChannel.config().setOption(ChannelOption.AUTO_READ, false); - ProxyChannelManager.addUserChannelToCmdChannel(cmdChannel, userId, userChannel); +// ProxyChannelManager.addUserChannelToCmdChannel(cmdChannel, userId, userChannel); + ProxyUtil.addUserChannelToCmdChannel(cmdChannel, userId, userChannel); cmdChannel.writeAndFlush(ProxyMessage.buildConnectMessage(userId).setData(lanInfo.getBytes())); } @@ -96,7 +98,9 @@ public class UserChannelHandler extends SimpleChannelInboundHandler { // 通知代理客户端 Channel userChannel = ctx.channel(); InetSocketAddress sa = (InetSocketAddress) userChannel.localAddress(); - Channel cmdChannel = ProxyChannelManager.getCmdChannel(sa.getPort()); +// Channel cmdChannel = ProxyChannelManager.getCmdChannel(sa.getPort()); + Channel cmdChannel = ProxyUtil.getCmdChannelByServerPort(sa.getPort()); + if (cmdChannel == null) { // 该端口还没有代理客户端 @@ -104,8 +108,11 @@ public class UserChannelHandler extends SimpleChannelInboundHandler { } else { // 用户连接断开，从控制连接中移除 - String userId = ProxyChannelManager.getUserChannelUserId(userChannel); - ProxyChannelManager.removeUserChannelFromCmdChannel(cmdChannel, userId); +// String userId = ProxyChannelManager.getUserChannelUserId(userChannel); + String userId = ProxyUtil.getUserChannelUserId(userChannel); +// ProxyChannelManager.removeUserChannelFromCmdChannel(cmdChannel, userId); + ProxyUtil.removeUserChannelFromCmdChannel(cmdChannel, userId); + Channel proxyChannel = userChannel.attr(Constants.NEXT_CHANNEL).get(); if (proxyChannel != null && proxyChannel.isActive()) { proxyChannel.attr(Constants.NEXT_CHANNEL).remove(); @@ -127,7 +134,9 @@ public class UserChannelHandler extends SimpleChannelInboundHandler { // 通知代理客户端 Channel userChannel = ctx.channel(); InetSocketAddress sa = (InetSocketAddress) userChannel.localAddress(); - Channel cmdChannel = ProxyChannelManager.getCmdChannel(sa.getPort()); +// Channel cmdChannel = ProxyChannelManager.getCmdChannel(sa.getPort()); + Channel cmdChannel = ProxyUtil.getCmdChannelByServerPort(sa.getPort()); + if (cmdChannel == null) { // 该端口还没有代理客户端 diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/domain/CmdChannelAttachInfo.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/domain/CmdChannelAttachInfo.java new file mode 100644 index 00000000..18b9caa5 --- /dev/null +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/domain/CmdChannelAttachInfo.java @@ -0,0 +1,55 @@ +/** + * Copyright (c) 2022 aoshiguchen + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ +package fun.asgc.neutrino.proxy.server.proxy.domain; + +import io.netty.channel.Channel; +import lombok.Data; +import lombok.experimental.Accessors; + +import java.util.Map; +import java.util.Set; + +/** + * + * @author: aoshiguchen + * @date: 2022/8/30 + */ +@Accessors(chain = true) +@Data +public class CmdChannelAttachInfo { + /** + * 用户通道映射 + */ + private Map userChannelMap; + /** + * 客户端信息 + */ + private String clientLanInfo; + /** + * 服务端端口集合 + */ + private Set serverPorts; + /** + * licenseKey + */ + private String licenseKey; +} diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/domain/UserChannelAttachInfo.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/domain/UserChannelAttachInfo.java new file mode 100644 index 00000000..ba4a3e06 --- /dev/null +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/domain/UserChannelAttachInfo.java @@ -0,0 +1,37 @@ +/** + * Copyright (c) 2022 aoshiguchen + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ +package fun.asgc.neutrino.proxy.server.proxy.domain; + +import lombok.Data; +import lombok.experimental.Accessors; + +/** + * + * @author: aoshiguchen + * @date: 2022/8/30 + */ +@Accessors(chain = true) +@Data +public class UserChannelAttachInfo { + private String userId; + private String lanInfo; +} diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageAuthHandler.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageAuthHandler.java index a210f57d..94b1a50b 100644 --- a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageAuthHandler.java +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageAuthHandler.java @@ -40,7 +40,6 @@ import fun.asgc.neutrino.proxy.server.proxy.domain.ProxyMapping; import fun.asgc.neutrino.proxy.server.service.LicenseService; import fun.asgc.neutrino.proxy.server.service.PortMappingService; import fun.asgc.neutrino.proxy.server.service.UserService; -import fun.asgc.neutrino.proxy.server.util.ProxyChannelManager; import fun.asgc.neutrino.proxy.server.util.ProxyUtil; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.Channel; @@ -117,13 +116,16 @@ public class ProxyMessageAuthHandler implements ProxyMessageHandler { return; } - Channel channel = ProxyChannelManager.getCmdChannel(licenseKey); +// Channel channel = ProxyChannelManager.getCmdChannel(licenseKey); + Channel channel = ProxyUtil.getCmdChannelByLicenseKey(licenseKey); + if (channel != null) { ctx.channel().close(); return; } - ProxyChannelManager.addCmdChannel(ports, licenseKey, ctx.channel()); +// ProxyChannelManager.addCmdChannel(ports, licenseKey, ctx.channel()); + ProxyUtil.addCmdChannel(licenseKey, ctx.channel(), ports); startUserPortServer(ports); } diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageConnectHandler.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageConnectHandler.java index 5fd83fc1..6dd043a5 100644 --- a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageConnectHandler.java +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageConnectHandler.java @@ -29,7 +29,7 @@ import fun.asgc.neutrino.proxy.core.Constants; import fun.asgc.neutrino.proxy.core.ProxyDataTypeEnum; import fun.asgc.neutrino.proxy.core.ProxyMessage; import fun.asgc.neutrino.proxy.core.ProxyMessageHandler; -import fun.asgc.neutrino.proxy.server.util.ProxyChannelManager; +import fun.asgc.neutrino.proxy.server.util.ProxyUtil; import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelOption; @@ -58,13 +58,16 @@ public class ProxyMessageConnectHandler implements ProxyMessageHandler { return; } - Channel cmdChannel = ProxyChannelManager.getCmdChannel(tokens[1]); +// Channel cmdChannel = ProxyChannelManager.getCmdChannel(tokens[1]); + Channel cmdChannel = ProxyUtil.getCmdChannelByLicenseKey(tokens[1]); + if (cmdChannel == null) { ctx.channel().close(); return; } - Channel userChannel = ProxyChannelManager.getUserChannel(cmdChannel, tokens[0]); +// Channel userChannel = ProxyChannelManager.getUserChannel(cmdChannel, tokens[0]); + Channel userChannel = ProxyUtil.getUserChannel(cmdChannel, tokens[0]); if (userChannel != null) { ctx.channel().attr(Constants.USER_ID).set(tokens[0]); ctx.channel().attr(Constants.CLIENT_KEY).set(tokens[1]); diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageDisconnectHandler.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageDisconnectHandler.java index 5b28ba5c..edf93939 100644 --- a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageDisconnectHandler.java +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageDisconnectHandler.java @@ -29,7 +29,8 @@ import fun.asgc.neutrino.proxy.core.Constants; import fun.asgc.neutrino.proxy.core.ProxyDataTypeEnum; import fun.asgc.neutrino.proxy.core.ProxyMessage; import fun.asgc.neutrino.proxy.core.ProxyMessageHandler; -import fun.asgc.neutrino.proxy.server.util.ProxyChannelManager; +import fun.asgc.neutrino.proxy.server.proxy.domain.UserChannelAttachInfo; +import fun.asgc.neutrino.proxy.server.util.ProxyUtil; import io.netty.buffer.Unpooled; import io.netty.channel.Channel; import io.netty.channel.ChannelFutureListener; @@ -52,7 +53,8 @@ public class ProxyMessageDisconnectHandler implements ProxyMessageHandler { // 代理连接没有连上服务器由控制连接发送用户端断开连接消息 if (clientKey == null) { String userId = proxyMessage.getInfo(); - Channel userChannel = ProxyChannelManager.removeUserChannelFromCmdChannel(ctx.channel(), userId); +// Channel userChannel = ProxyChannelManager.removeUserChannelFromCmdChannel(ctx.channel(), userId); + Channel userChannel = ProxyUtil.removeUserChannelFromCmdChannel(ctx.channel(), userId); if (userChannel != null) { // 数据发送完成后再关闭连接，解决http1.0数据传输问题 userChannel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE); @@ -60,18 +62,20 @@ public class ProxyMessageDisconnectHandler implements ProxyMessageHandler { return; } - Channel cmdChannel = ProxyChannelManager.getCmdChannel(clientKey); +// Channel cmdChannel = ProxyChannelManager.getCmdChannel(clientKey); + Channel cmdChannel = ProxyUtil.getCmdChannelByLicenseKey(clientKey); if (cmdChannel == null) { return; } - Channel userChannel = ProxyChannelManager.removeUserChannelFromCmdChannel(cmdChannel, ctx.channel().attr(Constants.USER_ID).get()); +// Channel userChannel = ProxyChannelManager.removeUserChannelFromCmdChannel(cmdChannel, ctx.channel().attr(Constants.USER_ID).get()); + Channel userChannel = ProxyUtil.removeUserChannelFromCmdChannel(cmdChannel, ((UserChannelAttachInfo)ProxyUtil.getAttachInfo(ctx.channel())).getUserId()); if (userChannel != null) { // 数据发送完成后再关闭连接，解决http1.0数据传输问题 userChannel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE); - ctx.channel().attr(Constants.NEXT_CHANNEL).remove(); - ctx.channel().attr(Constants.CLIENT_KEY).remove(); - ctx.channel().attr(Constants.USER_ID).remove(); +// ctx.channel().attr(Constants.NEXT_CHANNEL).remove(); +// ctx.channel().attr(Constants.CLIENT_KEY).remove(); +// ctx.channel().attr(Constants.USER_ID).remove(); } } diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/util/ProxyChannelManager.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/util/ProxyChannelManager.java deleted file mode 100644 index a548797c..00000000 --- a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/util/ProxyChannelManager.java +++ /dev/null @@ -1,200 +0,0 @@ -/** - * Copyright (c) 2022 aoshiguchen - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package fun.asgc.neutrino.proxy.server.util; - -import fun.asgc.neutrino.proxy.core.Constants; -import io.netty.channel.Channel; -import io.netty.util.AttributeKey; - -import java.net.InetSocketAddress; -import java.util.Iterator; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; - -/** - * - * @author: aoshiguchen - * @date: 2022/6/16 - */ -public class ProxyChannelManager { - - private static final AttributeKey> USER_CHANNELS = AttributeKey.newInstance("user_channels"); - - private static final AttributeKey REQUEST_LAN_INFO = AttributeKey.newInstance("request_lan_info"); - - private static final AttributeKey> CHANNEL_PORT = AttributeKey.newInstance("channel_port"); - - private static final AttributeKey CHANNEL_CLIENT_KEY = AttributeKey.newInstance("channel_client_key"); - - private static Map portCmdChannelMapping = new ConcurrentHashMap(); - - private static Map cmdChannels = new ConcurrentHashMap(); - - /** - * 增加代理服务器端口与代理控制客户端连接的映射关系 - * - * @param ports - * @param channel - */ - public static void addCmdChannel(Set ports, String clientKey, Channel channel) { - if (ports == null) { - throw new IllegalArgumentException("port can not be null"); - } - - // 客户端（proxy-client）相对较少，这里同步的比较重 - // 保证服务器对外端口与客户端到服务器的连接关系在临界情况时调用removeChannel(Channel channel)时不出问题 - synchronized (portCmdChannelMapping) { - for (int port : ports) { - portCmdChannelMapping.put(port, channel); - } - } - - channel.attr(CHANNEL_PORT).set(ports); - channel.attr(CHANNEL_CLIENT_KEY).set(clientKey); - channel.attr(USER_CHANNELS).set(new ConcurrentHashMap<>()); - cmdChannels.put(clientKey, channel); - } - - /** - * 代理客户端连接断开后清除关系 - * - * @param channel - */ - public static void removeCmdChannel(Channel channel) { - if (channel.attr(CHANNEL_PORT).get() == null) { - return; - } - - String clientKey = channel.attr(CHANNEL_CLIENT_KEY).get(); - Channel channel0 = cmdChannels.remove(clientKey); - if (channel != channel0) { - cmdChannels.put(clientKey, channel); - } - - Set ports = channel.attr(CHANNEL_PORT).get(); - for (int port : ports) { - Channel proxyChannel = portCmdChannelMapping.remove(port); - if (proxyChannel == null) { - continue; - } - - // 在执行断连之前新的连接已经连上来了 - if (proxyChannel != channel) { - portCmdChannelMapping.put(port, proxyChannel); - } - } - - if (channel.isActive()) { - channel.close(); - } - - Map userChannels = getUserChannels(channel); - Iterator ite = userChannels.keySet().iterator(); - while (ite.hasNext()) { - Channel userChannel = userChannels.get(ite.next()); - if (userChannel.isActive()) { - userChannel.close(); - } - } - } - - public static Channel getCmdChannel(Integer port) { - return portCmdChannelMapping.get(port); - } - - public static Channel getCmdChannel(String clientKey) { - return cmdChannels.get(clientKey); - } - - /** - * 增加用户连接与代理客户端连接关系 - * - * @param userId - * @param userChannel - */ - public static void addUserChannelToCmdChannel(Channel cmdChannel, String userId, Channel userChannel) { - InetSocketAddress sa = (InetSocketAddress) userChannel.localAddress(); - String lanInfo = ProxyUtil.getClientLanInfoByServerPort(sa.getPort()); - userChannel.attr(Constants.USER_ID).set(userId); - userChannel.attr(REQUEST_LAN_INFO).set(lanInfo); - cmdChannel.attr(USER_CHANNELS).get().put(userId, userChannel); - } - - /** - * 删除用户连接与代理客户端连接关系 - * - * @param userId - * @return - */ - public static Channel removeUserChannelFromCmdChannel(Channel cmdChannel, String userId) { - if (cmdChannel.attr(USER_CHANNELS).get() == null) { - return null; - } - - synchronized (cmdChannel) { - return cmdChannel.attr(USER_CHANNELS).get().remove(userId); - } - } - - /** - * 根据代理客户端连接与用户编号获取用户连接 - * - * @param userId - * @return - */ - public static Channel getUserChannel(Channel cmdChannel, String userId) { - return cmdChannel.attr(USER_CHANNELS).get().get(userId); - } - - /** - * 获取用户编号 - * - * @param userChannel - * @return - */ - public static String getUserChannelUserId(Channel userChannel) { - return userChannel.attr(Constants.USER_ID).get(); - } - - /** - * 获取用户请求的内网IP端口信息 - * - * @param userChannel - * @return - */ - public static String getUserChannelRequestLanInfo(Channel userChannel) { - return userChannel.attr(REQUEST_LAN_INFO).get(); - } - - /** - * 获取代理控制客户端连接绑定的所有用户连接 - * - * @param cmdChannel - * @return - */ - public static Map getUserChannels(Channel cmdChannel) { - return cmdChannel.attr(USER_CHANNELS).get(); - } - -} diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/util/ProxyUtil.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/util/ProxyUtil.java index 54d9b1c8..68c5490c 100644 --- a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/util/ProxyUtil.java +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/util/ProxyUtil.java @@ -23,9 +23,17 @@ package fun.asgc.neutrino.proxy.server.util; import fun.asgc.neutrino.core.util.CollectionUtil; import fun.asgc.neutrino.core.util.StringUtil; +import fun.asgc.neutrino.proxy.core.ChannelAttribute; +import fun.asgc.neutrino.proxy.core.Constants; +import fun.asgc.neutrino.proxy.server.proxy.domain.CmdChannelAttachInfo; import fun.asgc.neutrino.proxy.server.proxy.domain.ProxyMapping; +import fun.asgc.neutrino.proxy.server.proxy.domain.UserChannelAttachInfo; +import io.netty.channel.Channel; +import io.netty.util.AttributeKey; +import java.net.InetSocketAddress; import java.util.*; +import java.util.concurrent.ConcurrentHashMap; /** * @@ -33,6 +41,7 @@ import java.util.*; * @date: 2022/8/30 */ public class ProxyUtil { + public static final AttributeKey CHANNEL_ATTR_KEY = AttributeKey.valueOf("netty.channel.attr"); /** * license -> 服务端口映射 */ @@ -41,6 +50,14 @@ public class ProxyUtil { * 代理信息映射 */ private static final Map proxyInfoMap = new HashMap<>(); + /** + * 服务端口 -> 指令通道映射 + */ + private static Map serverPortToCmdChannelMap = new ConcurrentHashMap<>(); + /** + * license -> 指令通道映射 + */ + private static Map licenseToCmdChannelMap = new ConcurrentHashMap<>(); /** * 初始化代理信息 @@ -78,4 +95,159 @@ public class ProxyUtil { public static String getClientLanInfoByServerPort(Integer serverPort) { return proxyInfoMap.get(serverPort); } + + /** + * 添加指令通道相关缓存信息 + * @param licenseKey licenseKey + * @param cmdChannel 指令通道 + * @param serverPorts 服务端端口集合 + */ + public static void addCmdChannel(String licenseKey, Channel cmdChannel, Set serverPorts) { + if (CollectionUtil.isEmpty(serverPorts)) { + return; + } + + // 客户端（proxy-client）相对较少，这里同步的比较重 TODO 后续优化 + // 保证服务器对外端口与客户端到服务器的连接关系在临界情况时调用removeChannel(Channel channel)时不出问题 + synchronized (serverPortToCmdChannelMap) { + for (int port : serverPorts) { + serverPortToCmdChannelMap.put(port, cmdChannel); + } + } + + setAttachInfo(cmdChannel, new CmdChannelAttachInfo() + .setServerPorts(serverPorts) + .setLicenseKey(licenseKey) + .setUserChannelMap(new HashMap<>(16))); + licenseToCmdChannelMap.put(licenseKey, cmdChannel); + } + + /** + * 删除指令通道相关缓存信息 + * @param cmdChannel 指令通道 + */ + public static void removeCmdChannel(Channel cmdChannel) { + if (null == cmdChannel || null == getAttachInfo(cmdChannel)) { + return; + } + CmdChannelAttachInfo cmdChannelAttachInfo = getAttachInfo(cmdChannel); + Channel channel0 = licenseToCmdChannelMap.remove(cmdChannelAttachInfo.getLicenseKey()); + if (cmdChannel != channel0) { + licenseToCmdChannelMap.put(cmdChannelAttachInfo.getLicenseKey(), cmdChannel); + } + + for (int port : cmdChannelAttachInfo.getServerPorts()) { + Channel proxyChannel = serverPortToCmdChannelMap.remove(port); + if (proxyChannel == null) { + continue; + } + + // 在执行断连之前新的连接已经连上来了 + if (proxyChannel != cmdChannel) { + serverPortToCmdChannelMap.put(port, proxyChannel); + } + } + + if (cmdChannel.isActive()) { + cmdChannel.close(); + } + + Map userChannels = cmdChannelAttachInfo.getUserChannelMap(); + Iterator ite = userChannels.keySet().iterator(); + while (ite.hasNext()) { + Channel userChannel = userChannels.get(ite.next()); + if (userChannel.isActive()) { + userChannel.close(); + } + } + } + + public static Channel getCmdChannelByServerPort(Integer serverPort) { + return serverPortToCmdChannelMap.get(serverPort); + } + + public static Channel getCmdChannelByLicenseKey(String licenseKey) { + return licenseToCmdChannelMap.get(licenseKey); + } + + /** + * 增加用户连接与代理客户端连接关系 + * + * @param userId + * @param userChannel + */ + public static void addUserChannelToCmdChannel(Channel cmdChannel, String userId, Channel userChannel) { + InetSocketAddress sa = (InetSocketAddress) userChannel.localAddress(); + String lanInfo = getClientLanInfoByServerPort(sa.getPort()); + setAttachInfo(userChannel, new UserChannelAttachInfo() + .setUserId(userId) + .setLanInfo(lanInfo) + ); + ((CmdChannelAttachInfo)getAttachInfo(cmdChannel)).getUserChannelMap().put(userId, userChannel); + } + + public static Channel removeUserChannelFromCmdChannel(Channel cmdChannel, String userId) { + if (null == getAttachInfo(cmdChannel) || null == ((CmdChannelAttachInfo)getAttachInfo(cmdChannel)).getUserChannelMap().get(userId)) { + return null; + } + + synchronized (cmdChannel) { + return ((CmdChannelAttachInfo)getAttachInfo(cmdChannel)).getUserChannelMap().remove(userId); + } + } + + /** + * 根据代理客户端连接与用户编号获取用户连接 + * + * @param userId + * @return + */ + public static Channel getUserChannel(Channel cmdChannel, String userId) { + if (null == cmdChannel || null == getAttachInfo(cmdChannel)) { + return null; + } + return ((CmdChannelAttachInfo)getAttachInfo(cmdChannel)).getUserChannelMap().get(userId); + } + + /** + * 获取用户编号 + * + * @param userChannel + * @return + */ + public static String getUserChannelUserId(Channel userChannel) { + if (null == userChannel || null == getAttachInfo(userChannel)) { + return null; + } + return ((UserChannelAttachInfo)getAttachInfo(userChannel)).getUserId(); + } + + /** + * 获取代理控制客户端连接绑定的所有用户连接 + * + * @param cmdChannel + * @return + */ + public static Map getUserChannels(Channel cmdChannel) { + if (null == cmdChannel || null == getAttachInfo(cmdChannel)) { + return null; + } + return ((CmdChannelAttachInfo)getAttachInfo(cmdChannel)).getUserChannelMap(); + } + + private static void setAttachInfo(Channel channel, Object obj) { + if (null == channel) { + return; + } + channel.attr(CHANNEL_ATTR_KEY).set(ChannelAttribute.create() + .set("attachInfo", obj) + ); + } + + public static T getAttachInfo(Channel channel) { + if (null == channel || null == channel.attr(CHANNEL_ATTR_KEY).get()) { + return null; + } + return channel.attr(CHANNEL_ATTR_KEY).get().get("attachInfo"); + } }