From cde1a99cda7b05d6f5ee4677a68db38f0233b89b Mon Sep 17 00:00:00 2001 From: aoshiguchen <1052045476@qq.com> Date: Wed, 31 Aug 2022 15:59:28 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BB=A3=E7=A0=81=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/aop/AsgcProxyGeneratorTest.java | 1 + .../client/core/ClientChannelHandler.java | 10 +- .../proxy/client/core/ProxyClientRunner.java | 4 +- .../client/core/RealServerChannelHandler.java | 8 +- .../handler/ProxyMessageConnectHandler.java | 8 +- .../ProxyMessageDisconnectHandler.java | 4 +- .../client/util/ClientChannelMannager.java | 139 ------------------ .../neutrino/proxy/client/util/ProxyUtil.java | 135 +++++++++++++++++ 8 files changed, 153 insertions(+), 156 deletions(-) delete mode 100644 neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/util/ClientChannelMannager.java create mode 100644 neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/util/ProxyUtil.java diff --git a/neutrino-core/src/test/java/fun/asgc/neutrino/core/aop/AsgcProxyGeneratorTest.java b/neutrino-core/src/test/java/fun/asgc/neutrino/core/aop/AsgcProxyGeneratorTest.java index 7102c1c7..d9c3da0f 100644 --- a/neutrino-core/src/test/java/fun/asgc/neutrino/core/aop/AsgcProxyGeneratorTest.java +++ b/neutrino-core/src/test/java/fun/asgc/neutrino/core/aop/AsgcProxyGeneratorTest.java @@ -56,6 +56,7 @@ public class AsgcProxyGeneratorTest { } public static class A implements Handler{ + @Override public void handle(String s, String s2) { System.out.println("hello"); } diff --git a/neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/core/ClientChannelHandler.java b/neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/core/ClientChannelHandler.java index 33abf8a9..12155b07 100644 --- a/neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/core/ClientChannelHandler.java +++ b/neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/core/ClientChannelHandler.java @@ -26,7 +26,7 @@ import fun.asgc.neutrino.core.base.DefaultDispatcher; 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.client.util.ClientChannelMannager; +import fun.asgc.neutrino.proxy.client.util.ProxyUtil; import fun.asgc.neutrino.proxy.core.*; import io.netty.channel.*; import lombok.extern.slf4j.Slf4j; @@ -75,10 +75,10 @@ public class ClientChannelHandler extends SimpleChannelInboundHandler USER_CHANNEL_WRITEABLE = AttributeKey.newInstance("user_channel_writeable"); - - private static final AttributeKey CLIENT_CHANNEL_WRITEABLE = AttributeKey.newInstance("client_channel_writeable"); - - private static final int MAX_POOL_SIZE = 100; - - private static Map realServerChannels = new ConcurrentHashMap(); - - private static ConcurrentLinkedQueue proxyChannelPool = new ConcurrentLinkedQueue(); - - private static volatile Channel cmdChannel; - - public static void borrowProxyChanel(Bootstrap bootstrap, final ProxyChannelBorrowListener borrowListener) { - Channel channel = proxyChannelPool.poll(); - if (channel != null) { - borrowListener.success(channel); - return; - } - - bootstrap.connect(ProxyConfig.instance.getClient().getServerIp(), ProxyConfig.instance.getClient().getServerPort()).addListener(new ChannelFutureListener() { - - @Override - public void operationComplete(ChannelFuture future) throws Exception { - if (future.isSuccess()) { - borrowListener.success(future.channel()); - } else { - borrowListener.error(future.cause()); - } - } - }); - } - - public static void returnProxyChanel(Channel proxyChanel) { - if (proxyChannelPool.size() > MAX_POOL_SIZE) { - proxyChanel.close(); - } else { - proxyChanel.config().setOption(ChannelOption.AUTO_READ, true); - proxyChanel.attr(Constants.NEXT_CHANNEL).remove(); - proxyChannelPool.offer(proxyChanel); - } - } - - public static void removeProxyChanel(Channel proxyChanel) { - proxyChannelPool.remove(proxyChanel); - } - - public static void setCmdChannel(Channel cmdChannel) { - ClientChannelMannager.cmdChannel = cmdChannel; - } - - public static Channel getCmdChannel() { - return cmdChannel; - } - - public static void setRealServerChannelUserId(Channel realServerChannel, String userId) { - realServerChannel.attr(Constants.USER_ID).set(userId); - } - - public static String getRealServerChannelUserId(Channel realServerChannel) { - return realServerChannel.attr(Constants.USER_ID).get(); - } - - public static Channel getRealServerChannel(String userId) { - return realServerChannels.get(userId); - } - - public static void addRealServerChannel(String userId, Channel realServerChannel) { - realServerChannels.put(userId, realServerChannel); - } - - public static Channel removeRealServerChannel(String userId) { - return realServerChannels.remove(userId); - } - - public static boolean isRealServerReadable(Channel realServerChannel) { - return realServerChannel.attr(CLIENT_CHANNEL_WRITEABLE).get() && realServerChannel.attr(USER_CHANNEL_WRITEABLE).get(); - } - - public static void clearRealServerChannels() { - Iterator> ite = realServerChannels.entrySet().iterator(); - while (ite.hasNext()) { - Channel realServerChannel = ite.next().getValue(); - if (realServerChannel.isActive()) { - realServerChannel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE); - } - } - - realServerChannels.clear(); - } -} diff --git a/neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/util/ProxyUtil.java b/neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/util/ProxyUtil.java new file mode 100644 index 00000000..5241a20a --- /dev/null +++ b/neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/util/ProxyUtil.java @@ -0,0 +1,135 @@ +/** + * 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.client.util; + +import fun.asgc.neutrino.proxy.client.config.ProxyConfig; +import fun.asgc.neutrino.proxy.client.core.ProxyChannelBorrowListener; +import fun.asgc.neutrino.proxy.core.Constants; +import io.netty.bootstrap.Bootstrap; +import io.netty.buffer.Unpooled; +import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelOption; +import io.netty.util.AttributeKey; + +import java.util.Iterator; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; + +/** + * + * @author: aoshiguchen + * @date: 2022/8/31 + */ +public class ProxyUtil { + private static final AttributeKey USER_CHANNEL_WRITEABLE = AttributeKey.newInstance("user_channel_writeable"); + + private static final AttributeKey CLIENT_CHANNEL_WRITEABLE = AttributeKey.newInstance("client_channel_writeable"); + + private static final int MAX_POOL_SIZE = 100; + + private static Map realServerChannels = new ConcurrentHashMap(); + + private static ConcurrentLinkedQueue proxyChannelPool = new ConcurrentLinkedQueue(); + + private static volatile Channel cmdChannel; + + public static void borrowProxyChanel(Bootstrap bootstrap, final ProxyChannelBorrowListener borrowListener) { + Channel channel = proxyChannelPool.poll(); + if (channel != null) { + borrowListener.success(channel); + return; + } + + bootstrap.connect(ProxyConfig.instance.getClient().getServerIp(), ProxyConfig.instance.getClient().getServerPort()).addListener(new ChannelFutureListener() { + + @Override + public void operationComplete(ChannelFuture future) throws Exception { + if (future.isSuccess()) { + borrowListener.success(future.channel()); + } else { + borrowListener.error(future.cause()); + } + } + }); + } + + public static void returnProxyChanel(Channel proxyChanel) { + if (proxyChannelPool.size() > MAX_POOL_SIZE) { + proxyChanel.close(); + } else { + proxyChanel.config().setOption(ChannelOption.AUTO_READ, true); + proxyChanel.attr(Constants.NEXT_CHANNEL).remove(); + proxyChannelPool.offer(proxyChanel); + } + } + + public static void removeProxyChanel(Channel proxyChanel) { + proxyChannelPool.remove(proxyChanel); + } + + public static void setCmdChannel(Channel cmdChannel) { + ProxyUtil.cmdChannel = cmdChannel; + } + + public static Channel getCmdChannel() { + return cmdChannel; + } + + public static void setRealServerChannelUserId(Channel realServerChannel, String userId) { + realServerChannel.attr(Constants.USER_ID).set(userId); + } + + public static String getRealServerChannelUserId(Channel realServerChannel) { + return realServerChannel.attr(Constants.USER_ID).get(); + } + + public static Channel getRealServerChannel(String userId) { + return realServerChannels.get(userId); + } + + public static void addRealServerChannel(String userId, Channel realServerChannel) { + realServerChannels.put(userId, realServerChannel); + } + + public static Channel removeRealServerChannel(String userId) { + return realServerChannels.remove(userId); + } + + public static boolean isRealServerReadable(Channel realServerChannel) { + return realServerChannel.attr(CLIENT_CHANNEL_WRITEABLE).get() && realServerChannel.attr(USER_CHANNEL_WRITEABLE).get(); + } + + public static void clearRealServerChannels() { + Iterator> ite = realServerChannels.entrySet().iterator(); + while (ite.hasNext()) { + Channel realServerChannel = ite.next().getValue(); + if (realServerChannel.isActive()) { + realServerChannel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE); + } + } + + realServerChannels.clear(); + } +}