Commit 095c04a0 authored by Administrator's avatar Administrator

端口关闭

parent 2031ac45
package com.virjar.g4proxy_new.server; package com.virjar.g4proxy_new.server;
import com.google.common.base.Splitter; import com.google.common.base.Splitter;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets; import com.google.common.collect.Sets;
import java.util.Map;
import java.util.Set; import java.util.Set;
import java.util.concurrent.BlockingDeque; import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.LinkedBlockingDeque;
...@@ -20,6 +22,8 @@ public class AvailablePortResManager { ...@@ -20,6 +22,8 @@ public class AvailablePortResManager {
private static final String defaultSpaceConfig = "20000-25000"; private static final String defaultSpaceConfig = "20000-25000";
private static Map<String, Integer> allocatedResources = Maps.newConcurrentMap();
public static void init(String spaceConfig) { public static void init(String spaceConfig) {
if (inited.compareAndSet(false, true)) { if (inited.compareAndSet(false, true)) {
...@@ -48,9 +52,18 @@ public class AvailablePortResManager { ...@@ -48,9 +52,18 @@ public class AvailablePortResManager {
portQueue.addAll(availablePortResource); portQueue.addAll(availablePortResource);
} }
public static Integer allocate() { public synchronized static Integer allocate(String clientId) {
if (allocatedResources.containsKey(clientId)) {
Integer integer = allocatedResources.get(clientId);
if (integer != null) {
return integer;
}
}
init(defaultSpaceConfig); init(defaultSpaceConfig);
return portQueue.poll(); Integer ret = portQueue.poll();
allocatedResources.put(clientId, ret);
return ret;
} }
} }
...@@ -171,6 +171,7 @@ public class NatServerChannelHandler extends SimpleChannelInboundHandler<NatMess ...@@ -171,6 +171,7 @@ public class NatServerChannelHandler extends SimpleChannelInboundHandler<NatMess
ctx.close(); ctx.close();
return; return;
} }
log.info("the nat channel InActive ,close ");
clientManager.closeClient(clientId); clientManager.closeClient(clientId);
super.channelInactive(ctx); super.channelInactive(ctx);
} }
......
...@@ -25,7 +25,7 @@ public class ClientManager { ...@@ -25,7 +25,7 @@ public class ClientManager {
// this.userMappingBootstrap = userMappingBootstrap; // this.userMappingBootstrap = userMappingBootstrap;
} }
public boolean registerNewClient(String client, Channel channel) { public boolean registerNewClient(String client, Channel natChannel) {
NatClientImage natClientImage = natClientImageMap.get(client); NatClientImage natClientImage = natClientImageMap.get(client);
Integer mappingPort; Integer mappingPort;
if (natClientImage != null) { if (natClientImage != null) {
...@@ -35,9 +35,9 @@ public class ClientManager { ...@@ -35,9 +35,9 @@ public class ClientManager {
mappingPort = natClientImage.getMappingPort(); mappingPort = natClientImage.getMappingPort();
//需要关闭所有的连接,但是计数器不能清零。计数器清零可能导致紊乱 //需要关闭所有的连接,但是计数器不能清零。计数器清零可能导致紊乱
natClientImage.updateChannel(channel); natClientImage.updateChannel(natChannel);
} else { } else {
mappingPort = AvailablePortResManager.allocate(); mappingPort = AvailablePortResManager.allocate(client);
if (mappingPort == null) { if (mappingPort == null) {
log.error("failed to allocate port"); log.error("failed to allocate port");
return false; return false;
...@@ -53,10 +53,10 @@ public class ClientManager { ...@@ -53,10 +53,10 @@ public class ClientManager {
log.warn("bind mapping port:{} failed", mappingPort, channelFuture.cause()); log.warn("bind mapping port:{} failed", mappingPort, channelFuture.cause());
return false; return false;
} }
natClientImage = new NatClientImage(client, mappingPort, channel); natClientImage = new NatClientImage(client, mappingPort, natChannel, channelFuture.channel());
} }
channel.attr(Constant.NAT_CHANNEL_CLIENT_KEY).set(client); natChannel.attr(Constant.NAT_CHANNEL_CLIENT_KEY).set(client);
clientPortBiMap.put(client, mappingPort); clientPortBiMap.put(client, mappingPort);
natClientImageMap.put(client, natClientImage); natClientImageMap.put(client, natClientImage);
log.info("client :{} register success,with port:{}", client, mappingPort); log.info("client :{} register success,with port:{}", client, mappingPort);
...@@ -77,12 +77,15 @@ public class ClientManager { ...@@ -77,12 +77,15 @@ public class ClientManager {
public void closeClient(String clientId) { public void closeClient(String clientId) {
log.info("close client :{}", clientId);
NatClientImage client = natClientImageMap.remove(clientId); NatClientImage client = natClientImageMap.remove(clientId);
if (client == null) { if (client == null) {
log.error("now client registered for clientId:{}", clientId); log.error("no client registered for clientId:{}", clientId);
return; return;
} }
client.closeAllUserChannel(); client.closeAllUserChannel();
client.getNatChannel().close(); client.getNatChannel().close();
//这样,停止监听对应的代理端口,防止代理请求再次打进来
client.getUserMappingServerChannel().close();
} }
} }
...@@ -20,12 +20,16 @@ public class NatClientImage { ...@@ -20,12 +20,16 @@ public class NatClientImage {
@Getter @Getter
private Channel natChannel; private Channel natChannel;
@Getter
private Channel userMappingServerChannel;
private Map<Long, Channel> userMappingChannels = Maps.newConcurrentMap(); private Map<Long, Channel> userMappingChannels = Maps.newConcurrentMap();
public NatClientImage(String clientId, int mappingPort, Channel natChannel) { public NatClientImage(String clientId, int mappingPort, Channel natChannel, Channel userMappingServerChannel) {
this.clientId = clientId; this.clientId = clientId;
this.mappingPort = mappingPort; this.mappingPort = mappingPort;
this.natChannel = natChannel; this.natChannel = natChannel;
this.userMappingServerChannel = userMappingServerChannel;
} }
public void updateChannel(Channel channel) { public void updateChannel(Channel channel) {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment