服务端代理逻辑优化
This commit is contained in:
aoshiguchen
@@ -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<String, Object> {
|
||||
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> 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)));
|
||||
}
|
||||
}
|
||||
+8
-4
@@ -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<ProxyMessa
|
||||
if (userChannel != null && userChannel.isActive()) {
|
||||
String clientKey = ctx.channel().attr(Constants.CLIENT_KEY).get();
|
||||
String userId = ctx.channel().attr(Constants.USER_ID).get();
|
||||
Channel cmdChannel = ProxyChannelManager.getCmdChannel(clientKey);
|
||||
// Channel cmdChannel = ProxyChannelManager.getCmdChannel(clientKey);
|
||||
Channel cmdChannel = ProxyUtil.getCmdChannelByLicenseKey(clientKey);
|
||||
|
||||
if (cmdChannel != null) {
|
||||
ProxyChannelManager.removeUserChannelFromCmdChannel(cmdChannel, userId);
|
||||
// ProxyChannelManager.removeUserChannelFromCmdChannel(cmdChannel, userId);
|
||||
ProxyUtil.removeUserChannelFromCmdChannel(cmdChannel, userId);
|
||||
}
|
||||
|
||||
// 数据发送完成后再关闭连接，解决http1.0数据传输问题
|
||||
userChannel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
|
||||
userChannel.close();
|
||||
} else {
|
||||
ProxyChannelManager.removeCmdChannel(ctx.channel());
|
||||
// ProxyChannelManager.removeCmdChannel(ctx.channel());
|
||||
ProxyUtil.removeCmdChannel(ctx.channel());
|
||||
}
|
||||
|
||||
super.channelInactive(ctx);
|
||||
|
||||
+17
-8
@@ -24,7 +24,6 @@ package fun.asgc.neutrino.proxy.server.proxy.core;
|
||||
|
||||
import fun.asgc.neutrino.proxy.core.Constants;
|
||||
import fun.asgc.neutrino.proxy.core.ProxyMessage;
|
||||
import fun.asgc.neutrino.proxy.server.util.ProxyChannelManager;
|
||||
import fun.asgc.neutrino.proxy.server.util.ProxyUtil;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.channel.Channel;
|
||||
@@ -64,7 +63,8 @@ public class UserChannelHandler extends SimpleChannelInboundHandler<ByteBuf> {
|
||||
} 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<ByteBuf> {
|
||||
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<ByteBuf> {
|
||||
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<ByteBuf> {
|
||||
// 通知代理客户端
|
||||
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<ByteBuf> {
|
||||
} 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<ByteBuf> {
|
||||
// 通知代理客户端
|
||||
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) {
|
||||
|
||||
// 该端口还没有代理客户端
|
||||
|
||||
+55
@@ -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<String, Channel> userChannelMap;
|
||||
/**
|
||||
* 客户端信息
|
||||
*/
|
||||
private String clientLanInfo;
|
||||
/**
|
||||
* 服务端端口集合
|
||||
*/
|
||||
private Set<Integer> serverPorts;
|
||||
/**
|
||||
* licenseKey
|
||||
*/
|
||||
private String licenseKey;
|
||||
}
|
||||
+37
@@ -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;
|
||||
}
|
||||
+5
-3
@@ -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);
|
||||
}
|
||||
|
||||
+6
-3
@@ -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]);
|
||||
|
||||
+11
-7
@@ -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();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
-200
@@ -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<Map<String, Channel>> USER_CHANNELS = AttributeKey.newInstance("user_channels");
|
||||
|
||||
private static final AttributeKey<String> REQUEST_LAN_INFO = AttributeKey.newInstance("request_lan_info");
|
||||
|
||||
private static final AttributeKey<Set<Integer>> CHANNEL_PORT = AttributeKey.newInstance("channel_port");
|
||||
|
||||
private static final AttributeKey<String> CHANNEL_CLIENT_KEY = AttributeKey.newInstance("channel_client_key");
|
||||
|
||||
private static Map<Integer, Channel> portCmdChannelMapping = new ConcurrentHashMap<Integer, Channel>();
|
||||
|
||||
private static Map<String, Channel> cmdChannels = new ConcurrentHashMap<String, Channel>();
|
||||
|
||||
/**
|
||||
* 增加代理服务器端口与代理控制客户端连接的映射关系
|
||||
*
|
||||
* @param ports
|
||||
* @param channel
|
||||
*/
|
||||
public static void addCmdChannel(Set<Integer> 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<Integer> 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<String, Channel> userChannels = getUserChannels(channel);
|
||||
Iterator<String> 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<String, Channel> getUserChannels(Channel cmdChannel) {
|
||||
return cmdChannel.attr(USER_CHANNELS).get();
|
||||
}
|
||||
|
||||
}
|
||||
+172
@@ -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<ChannelAttribute> CHANNEL_ATTR_KEY = AttributeKey.valueOf("netty.channel.attr");
|
||||
/**
|
||||
* license -> 服务端口映射
|
||||
*/
|
||||
@@ -41,6 +50,14 @@ public class ProxyUtil {
|
||||
* 代理信息映射
|
||||
*/
|
||||
private static final Map<Integer, String> proxyInfoMap = new HashMap<>();
|
||||
/**
|
||||
* 服务端口 -> 指令通道映射
|
||||
*/
|
||||
private static Map<Integer, Channel> serverPortToCmdChannelMap = new ConcurrentHashMap<>();
|
||||
/**
|
||||
* license -> 指令通道映射
|
||||
*/
|
||||
private static Map<String, Channel> 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<Integer> 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<String, Channel> userChannels = cmdChannelAttachInfo.getUserChannelMap();
|
||||
Iterator<String> 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<String, Channel> 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> 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");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user