yeyinlaoying/neutrino-proxy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

端口映射操作（增删改、启用、禁用）时，代理实时生效

Browse Source
This commit is contained in:
aoshiguchen
2023-02-05 17:31:49 +08:00
parent 37ee8021be
commit eecd5d9d06
10 changed files with 240 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/constant/EnableStatusEnum.java
+9
View File
@@ -24,6 +24,11 @@ package fun.asgc.neutrino.proxy.server.constant;
import lombok.AllArgsConstructor;
import lombok.Getter;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
/**
* 启用状态枚举
* @author: aoshiguchen
@@ -34,7 +39,11 @@ import lombok.Getter;
public enum EnableStatusEnum {
ENABLE(1, "启用"),
DISABLE(2, "禁用");
private static Map<Integer, EnableStatusEnum> CACHE = Stream.of(EnableStatusEnum.values()).collect(Collectors.toMap(EnableStatusEnum::getStatus, Function.identity()));
private Integer status;
private String desc;
public static EnableStatusEnum of(Integer status) {
return CACHE.get(status);
}
}
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/core/VisitorChannelHandler.java
+12 -8
View File
@@ -23,6 +23,7 @@
package fun.asgc.neutrino.proxy.server.proxy.core;
import fun.asgc.neutrino.core.util.BeanManager;
import fun.asgc.neutrino.core.util.StringUtil;
import fun.asgc.neutrino.proxy.core.Constants;
import fun.asgc.neutrino.proxy.core.ProxyMessage;
import fun.asgc.neutrino.proxy.server.proxy.domain.VisitorChannelAttachInfo;
@@ -87,10 +88,13 @@ public class VisitorChannelHandler extends SimpleChannelInboundHandler<ByteBuf>
} else {
String visitorId = newVisitorId();
String lanInfo = ProxyUtil.getClientLanInfoByServerPort(sa.getPort());
// 用户连接到代理服务器时，设置用户连接不可读，等待代理后端服务器连接成功后再改变为可读状态
visitorChannel.config().setOption(ChannelOption.AUTO_READ, false);
ProxyUtil.addVisitorChannelToCmdChannel(cmdChannel, visitorId, visitorChannel);
cmdChannel.writeAndFlush(ProxyMessage.buildConnectMessage(visitorId).setData(lanInfo.getBytes()));
if (!StringUtil.isEmpty(lanInfo)) {
// 用户连接到代理服务器时，设置用户连接不可读，等待代理后端服务器连接成功后再改变为可读状态
visitorChannel.config().setOption(ChannelOption.AUTO_READ, false);
ProxyUtil.addVisitorChannelToCmdChannel(cmdChannel, visitorId, visitorChannel, sa.getPort());
cmdChannel.writeAndFlush(ProxyMessage.buildConnectMessage(visitorId).setData(lanInfo.getBytes()));
}
}
super.channelActive(ctx);
@@ -133,17 +137,17 @@ public class VisitorChannelHandler extends SimpleChannelInboundHandler<ByteBuf>
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
// 通知代理客户端
Channel userChannel = ctx.channel();
InetSocketAddress sa = (InetSocketAddress) userChannel.localAddress();
Channel visitorChannel = ctx.channel();
InetSocketAddress sa = (InetSocketAddress) visitorChannel.localAddress();
Channel cmdChannel = ProxyUtil.getCmdChannelByServerPort(sa.getPort());
if (cmdChannel == null) {
// 该端口还没有代理客户端
ctx.channel().close();
} else {
Channel proxyChannel = userChannel.attr(Constants.NEXT_CHANNEL).get();
Channel proxyChannel = visitorChannel.attr(Constants.NEXT_CHANNEL).get();
if (proxyChannel != null) {
proxyChannel.config().setOption(ChannelOption.AUTO_READ, userChannel.isWritable());
proxyChannel.config().setOption(ChannelOption.AUTO_READ, visitorChannel.isWritable());
}
}
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/domain/ProxyMapping.java
+7 -3
View File
@@ -52,11 +52,15 @@ public class ProxyMapping {
return list;
}
for (PortMappingDO portMapping : portMappingList) {
list.add(new ProxyMapping()
.setServerPort(portMapping.getServerPort())
.setLanInfo(String.format("%s:%s", portMapping.getClientIp(), portMapping.getClientPort())));
list.add(build(portMapping));
}
return list;
}
public static ProxyMapping build(PortMappingDO portMappingDO) {
return new ProxyMapping()
.setServerPort(portMappingDO.getServerPort())
.setLanInfo(String.format("%s:%s", portMappingDO.getClientIp(), portMappingDO.getClientPort()));
}
}
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/domain/VisitorChannelAttachInfo.java
+1
View File
@@ -34,6 +34,7 @@ import lombok.experimental.Accessors;
public class VisitorChannelAttachInfo {
private String visitorId;
private String lanInfo;
private Integer serverPort;
/**
* licenseId
*/
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/service/LicenseService.java
+6 -1
View File
@@ -58,6 +58,8 @@ public class LicenseService {
private LicenseMapper licenseMapper;
@Autowired
private UserMapper userMapper;
@Autowired
private VisitorChannelService visitorChannelService;
public Page<LicenseListRes> page(PageQuery pageQuery, LicenseListReq req) {
Page<LicenseListRes> page = Page.create(pageQuery);
@@ -159,7 +161,8 @@ public class LicenseService {
*/
public LicenseUpdateEnableStatusRes updateEnableStatus(LicenseUpdateEnableStatusReq req) {
licenseMapper.updateEnableStatus(req.getId(), req.getEnable(), new Date());
// 更新VisitorChannel
visitorChannelService.updateVisitorChannelByLicenseId(req.getId());
return new LicenseUpdateEnableStatusRes();
}
@@ -169,6 +172,8 @@ public class LicenseService {
*/
public void delete(Integer id) {
licenseMapper.delete(id);
// 更新VisitorChannel
visitorChannelService.updateVisitorChannelByLicenseId(id);
}
/**
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/service/PortMappingService.java
+22
View File
@@ -68,6 +68,8 @@ public class PortMappingService {
private UserMapper userMapper;
@Autowired
private PortPoolMapper portPoolMapper;
@Autowired
private VisitorChannelService visitorChannelService;
public Page<PortMappingListRes> page(PageQuery pageQuery, PortMappingListReq req) {
Page<PortMappingListRes> page = Page.create(pageQuery);
@@ -126,6 +128,8 @@ public class PortMappingService {
portMappingDO.setCreateTime(now);
portMappingDO.setUpdateTime(now);
portMappingMapper.add(portMappingDO);
// 更新VisitorChannel
visitorChannelService.addVisitorChannelByPortMapping(portMappingDO);
return new PortMappingCreateRes();
}
@@ -140,6 +144,10 @@ public class PortMappingService {
ParamCheckUtil.checkNotNull(portPoolDO, ExceptionConstant.PORT_NOT_EXIST);
ParamCheckUtil.checkExpression(null == portMappingMapper.findByPort(req.getServerPort(), Sets.newHashSet(req.getId())), ExceptionConstant.PORT_CANNOT_REPEAT_MAPPING, req.getServerPort());
// 查询原端口映射
PortMappingDO oldPortMappingDO = portMappingMapper.findById(req.getId());
ParamCheckUtil.checkNotNull(oldPortMappingDO, ExceptionConstant.PORT_MAPPING_NOT_EXIST);
PortMappingDO portMappingDO = new PortMappingDO();
portMappingDO.setId(req.getId());
portMappingDO.setLicenseId(req.getLicenseId());
@@ -147,7 +155,10 @@ public class PortMappingService {
portMappingDO.setClientIp(req.getClientIp());
portMappingDO.setClientPort(req.getClientPort());
portMappingDO.setUpdateTime(new Date());
portMappingDO.setEnable(EnableStatusEnum.ENABLE.getStatus());
portMappingMapper.update(portMappingDO);
// 更新VisitorChannel
visitorChannelService.updateVisitorChannelByPortMapping(oldPortMappingDO, portMappingDO);
return new PortMappingUpdateRes();
}
@@ -193,6 +204,14 @@ public class PortMappingService {
portMappingMapper.updateEnableStatus(req.getId(), req.getEnable(), new Date());
// 更新VisitorChannel
portMappingDO.setEnable(req.getEnable());
if (EnableStatusEnum.ENABLE == EnableStatusEnum.of(req.getEnable())) {
visitorChannelService.addVisitorChannelByPortMapping(portMappingDO);
} else {
visitorChannelService.removeVisitorChannelByPortMapping(portMappingDO);
}
return new PortMappingUpdateEnableStatusRes();
}
@@ -207,6 +226,9 @@ public class PortMappingService {
}
portMappingMapper.delete(id);
// 更新VisitorChannel
visitorChannelService.removeVisitorChannelByPortMapping(portMappingDO);
}
/**
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/service/UserService.java
+6 -1
View File
@@ -62,6 +62,8 @@ public class UserService {
private UserTokenMapper userTokenMapper;
@Autowired
private UserLoginRecordMapper userLoginRecordMapper;
@Autowired
private VisitorChannelService visitorChannelService;
public LoginRes login(LoginReq req) {
UserDO userDO = userMapper.findByLoginName(req.getLoginName());
@@ -162,7 +164,8 @@ public class UserService {
public UserUpdateEnableStatusRes updateEnableStatus(UserUpdateEnableStatusReq req) {
userMapper.updateEnableStatus(req.getId(), req.getEnable(), new Date());
// 更新VisitorChannel
visitorChannelService.updateVisitorChannelByUserId(req.getId());
return new UserUpdateEnableStatusRes();
}
@@ -208,5 +211,7 @@ public class UserService {
public void delete(Integer id) {
userMapper.delete(id);
// 更新VisitorChannel
visitorChannelService.updateVisitorChannelByUserId(id);
}
}
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/service/VisitorChannelService.java
+105 -5
View File
@@ -21,11 +21,20 @@
*/
package fun.asgc.neutrino.proxy.server.service;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import fun.asgc.neutrino.core.annotation.Autowired;
import fun.asgc.neutrino.core.annotation.Component;
import fun.asgc.neutrino.core.annotation.NonIntercept;
import fun.asgc.neutrino.core.util.CollectionUtil;
import fun.asgc.neutrino.proxy.core.Constants;
import fun.asgc.neutrino.proxy.server.constant.EnableStatusEnum;
import fun.asgc.neutrino.proxy.server.dal.LicenseMapper;
import fun.asgc.neutrino.proxy.server.dal.PortMappingMapper;
import fun.asgc.neutrino.proxy.server.dal.UserMapper;
import fun.asgc.neutrino.proxy.server.dal.entity.LicenseDO;
import fun.asgc.neutrino.proxy.server.dal.entity.PortMappingDO;
import fun.asgc.neutrino.proxy.server.dal.entity.UserDO;
import fun.asgc.neutrino.proxy.server.proxy.core.BytesMetricsHandler;
import fun.asgc.neutrino.proxy.server.proxy.core.VisitorChannelHandler;
import fun.asgc.neutrino.proxy.server.proxy.domain.CmdChannelAttachInfo;
@@ -57,16 +66,20 @@ public class VisitorChannelService {
@Autowired("serverWorkerGroup")
private NioEventLoopGroup serverWorkerGroup;
@Autowired
private PortMappingService portMappingService;
@Autowired
private ProxyMutualService proxyMutualService;
@Autowired
private UserMapper userMapper;
@Autowired
private LicenseMapper licenseMapper;
@Autowired
private PortMappingMapper portMappingMapper;
/**
* 初始化
* @param licenseId
*/
public void initVisitorChannel(Integer licenseId, Channel cmdChannel) {
List<PortMappingDO> portMappingList = portMappingService.findEnableListByLicenseId(licenseId);
List<PortMappingDO> portMappingList = portMappingMapper.findEnableListByLicenseId(licenseId);
// 没有端口映射仍然保持连接
ProxyUtil.initProxyInfo(licenseId, ProxyMapping.buildList(portMappingList));
@@ -77,11 +90,98 @@ public class VisitorChannelService {
/**
* 更新
* 触发时机：新增端口映射、修改端口映射、删除端口映射、禁用端口映射、启用端口映射、禁用license、启用license、禁用用户、启用用户
* 触发时机：删除用户、禁用用户、启用用户 （新增、修改用户不涉及VisitorChannel的变更）
* @param userId
*/
public void updateVisitorChannelByUserId(Integer userId) {
if (null == userId) {
return;
}
UserDO userDO = userMapper.findById(userId);
if (null == userDO || EnableStatusEnum.ENABLE != EnableStatusEnum.of(userDO.getEnable())) {
// TODO
} else {
// TODO
}
}
/**
* 更新
* 触发时机：删除license、禁用license、启用license （新增、修改license不涉及VisitorChannel的变更）
* 重置licenseKey，不会立即影响已经连接成功的license，如果想要立即影响，请先进行禁用
* @param licenseId
*/
public void UpdateVisitorChannel(Integer licenseId) {
public void updateVisitorChannelByLicenseId(Integer licenseId) {
if (null == licenseId) {
return;
}
LicenseDO licenseDO = licenseMapper.findById(licenseId);
if (null == licenseDO || EnableStatusEnum.ENABLE != EnableStatusEnum.of(licenseDO.getEnable())) {
// TODO
} else {
// TODO
}
}
/**
* 更新
* 触发时机：修改端口映射
* @param oldPortMappingDO
* @param newPortMappingDO
*/
public void updateVisitorChannelByPortMapping(PortMappingDO oldPortMappingDO, PortMappingDO newPortMappingDO) {
if (null == oldPortMappingDO || null == newPortMappingDO) {
return;
}
removeVisitorChannelByPortMapping(oldPortMappingDO);
addVisitorChannelByPortMapping(newPortMappingDO);
}
/**
* 新增VisitorChannel
* 触发时机：新增端口映射、启用端口映射
* @param portMappingDO
*/
public void addVisitorChannelByPortMapping(PortMappingDO portMappingDO) {
if (null == portMappingDO) {
return;
}
Channel cmdChannel = ProxyUtil.getCmdChannelByLicenseId(portMappingDO.getLicenseId());
if (null == cmdChannel) {
// 如果不存在有效的cmdChannel，则无需更新VisitorChannel
return;
}
if (EnableStatusEnum.DISABLE != EnableStatusEnum.of(portMappingDO.getEnable())) {
// 未删除且未禁用，则开启代理
ProxyUtil.addProxyInfo(portMappingDO.getLicenseId(), ProxyMapping.build(portMappingDO));
ProxyUtil.addCmdChannel(portMappingDO.getLicenseId(), cmdChannel, Sets.newHashSet(portMappingDO.getServerPort()));
startUserPortServer(ProxyUtil.getAttachInfo(cmdChannel), Lists.newArrayList(portMappingDO));
}
}
/**
* 删除VisitorChannel
* 触发时机：删除端口映射、禁用端口映射
* @param portMappingDO
*/
public void removeVisitorChannelByPortMapping(PortMappingDO portMappingDO) {
if (null == portMappingDO) {
return;
}
Channel cmdChannel = ProxyUtil.getCmdChannelByLicenseId(portMappingDO.getLicenseId());
if (null == cmdChannel) {
// 如果不存在有效的cmdChannel，则无需更新VisitorChannel
return;
}
Channel visitorChannel = ProxyUtil.getVisitorChannelByServerPort(portMappingDO.getServerPort());
if (null != visitorChannel) {
Channel proxyChannel = visitorChannel.attr(Constants.NEXT_CHANNEL).get();
if (null != proxyChannel) {
proxyChannel.close();
}
visitorChannel.close();
}
ProxyUtil.removeProxyInfo(portMappingDO.getServerPort());
}
private void startUserPortServer(CmdChannelAttachInfo cmdChannelAttachInfo, List<PortMappingDO> portMappingList) {
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/util/ProxyUtil.java
+47 -7
View File
@@ -21,6 +21,7 @@
*/
package fun.asgc.neutrino.proxy.server.util;
import com.google.common.collect.Sets;
import fun.asgc.neutrino.core.util.ChannelUtil;
import fun.asgc.neutrino.core.util.CollectionUtil;
import fun.asgc.neutrino.proxy.core.ChannelAttribute;
@@ -50,7 +51,7 @@ public class ProxyUtil {
/**
* 代理信息映射
*/
private static final Map<Integer, String> proxyInfoMap = new HashMap<>();
private static final Map<Integer, String> proxyInfoMap = new ConcurrentHashMap<>();
/**
* 服务端口 -> 指令通道映射
*/
@@ -59,6 +60,10 @@ public class ProxyUtil {
* license -> 指令通道映射
*/
private static Map<Integer, Channel> licenseToCmdChannelMap = new ConcurrentHashMap<>();
/**
* 服务端口 -> 访问通道映射
*/
private static Map<Integer, Channel> serverPortToVisitorChannel = new ConcurrentHashMap<>();
/**
* cmdChannelAttachInfo.getUserChannelMap() 读写锁
@@ -72,6 +77,10 @@ public class ProxyUtil {
*/
public static void initProxyInfo(Integer licenseId, List<ProxyMapping> proxyMappingList) {
licenseToServerPortMap.put(licenseId, new HashSet<>());
addProxyInfo(licenseId, proxyMappingList);
}
public static void addProxyInfo(Integer licenseId, List<ProxyMapping> proxyMappingList) {
if (!CollectionUtil.isEmpty(proxyMappingList)) {
for (ProxyMapping proxyMapping : proxyMappingList) {
licenseToServerPortMap.get(licenseId).add(proxyMapping.getServerPort());
@@ -80,6 +89,18 @@ public class ProxyUtil {
}
}
public static void addProxyInfo(Integer licenseId, ProxyMapping proxyMapping) {
if (null == licenseId || null == proxyMapping) {
return;
}
licenseToServerPortMap.get(licenseId).add(proxyMapping.getServerPort());
proxyInfoMap.put(proxyMapping.getServerPort(), proxyMapping.getLanInfo());
}
public static void removeProxyInfo(Integer serverPort) {
proxyInfoMap.remove(serverPort);
}
/**
* 根据licenseId获取服务端端口集合
* @param licenseId licenseId
@@ -110,12 +131,20 @@ public class ProxyUtil {
serverPortToCmdChannelMap.put(port, cmdChannel);
}
}
CmdChannelAttachInfo cmdChannelAttachInfo = getAttachInfo(cmdChannel);
if (null == cmdChannelAttachInfo) {
cmdChannelAttachInfo = new CmdChannelAttachInfo()
.setIp(ChannelUtil.getIP(cmdChannel))
.setLicenseId(licenseId)
.setVisitorChannelMap(new HashMap<>(16))
.setServerPorts(Sets.newHashSet());
setAttachInfo(cmdChannel, cmdChannelAttachInfo);
}
if (!CollectionUtil.isEmpty(serverPorts)) {
cmdChannelAttachInfo.getServerPorts().addAll(serverPorts);
}
setAttachInfo(cmdChannel, new CmdChannelAttachInfo()
.setIp(ChannelUtil.getIP(cmdChannel))
.setServerPorts(serverPorts)
.setLicenseId(licenseId)
.setVisitorChannelMap(new HashMap<>(16)));
licenseToCmdChannelMap.put(licenseId, cmdChannel);
}
@@ -174,7 +203,7 @@ public class ProxyUtil {
* @param visitorId
* @param visitorChannel
*/
public static void addVisitorChannelToCmdChannel(Channel cmdChannel, String visitorId, Channel visitorChannel) {
public static void addVisitorChannelToCmdChannel(Channel cmdChannel, String visitorId, Channel visitorChannel, Integer serverPort) {
InetSocketAddress sa = (InetSocketAddress) visitorChannel.localAddress();
String lanInfo = getClientLanInfoByServerPort(sa.getPort());
CmdChannelAttachInfo cmdChannelAttachInfo = getAttachInfo(cmdChannel);
@@ -182,6 +211,7 @@ public class ProxyUtil {
setAttachInfo(visitorChannel, new VisitorChannelAttachInfo()
.setVisitorId(visitorId)
.setLanInfo(lanInfo)
.setServerPort(serverPort)
.setLicenseId(cmdChannelAttachInfo.getLicenseId())
.setIp(ChannelUtil.getIP(visitorChannel))
);
@@ -191,6 +221,7 @@ public class ProxyUtil {
} finally {
userChannelMapLock.writeLock().unlock();
}
serverPortToVisitorChannel.put(serverPort, visitorChannel);
}
public static Channel removeVisitorChannelFromCmdChannel(Channel cmdChannel, String visitorId) {
@@ -219,6 +250,15 @@ public class ProxyUtil {
return ((CmdChannelAttachInfo)getAttachInfo(cmdChannel)).getVisitorChannelMap().get(visitorId);
}
/**
* 根据服务端口获取访问通道
* @param serverPort
* @return
*/
public static Channel getVisitorChannelByServerPort(Integer serverPort) {
return serverPortToVisitorChannel.get(serverPort);
}
/**
* 获取访问者ID
*
todolist.MD
+25 -1
View File
@@ -6,7 +6,31 @@
- 弹框展示月度明细
- 弹框展示今日流量明细
- 首页图表📈
- 1、License在线数
- 2、端口映射在线数
- 3、今日流量（上行、下行）
- 4、历史流量（上行、下行）
- 点击1～4 切换列表
- 在线License列表
- 用户名
- License名称
- LicenseKey
- 在线端口映射列表
- 用户名
- License名称
- 服务端口
- 代理客户端
- 今日24小时流量列表（按小时到排序）
- 用户名
- License名称
- 时间
- 流量
- 历史流量列表（取最近12个月按月到排序）
- 用户名
- License名称
- 时间
- 流量
- 今日流量折线图（上行、下行、总流量,按分钟统计0～24小时）
# Bug
- windows环境下直接运行发布版的jar包，日志输出乱码
- 部份用户windows环境下启动客户端，扫描类个数为0个