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

新增在线状态更新逻辑

Browse Source
This commit is contained in:
aoshiguchen
2022-09-04 16:25:25 +08:00
parent 05f6b48920
commit 33b8870d0b
14 changed files with 95 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/core/ClientChannelHandler.java
+19
View File
@@ -29,6 +29,7 @@ import fun.asgc.neutrino.core.util.LockUtil;
import fun.asgc.neutrino.proxy.client.util.ProxyUtil;
import fun.asgc.neutrino.proxy.core.*;
import io.netty.channel.*;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
/**
@@ -97,4 +98,22 @@ public class ClientChannelHandler extends SimpleChannelInboundHandler<ProxyMessa
cause.printStackTrace();
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if(evt instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent)evt;
switch (event.state()) {
case READER_IDLE:
// 读超时，断开连接
log.info("读超时");
ctx.channel().close();
break;
case WRITER_IDLE:
ctx.channel().writeAndFlush(ProxyMessage.buildHeartbeatMessage());
break;
case ALL_IDLE:
break;
}
}
}
}
neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/core/ProxyClientRunner.java
+2 -1
View File
@@ -42,6 +42,7 @@ import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import javax.net.ssl.SSLContext;
@@ -103,7 +104,7 @@ public class ProxyClientRunner implements ApplicationRunner {
proxyConfig.getProtocol().getLengthFieldOffset(), proxyConfig.getProtocol().getLengthFieldLength(),
proxyConfig.getProtocol().getLengthAdjustment(), proxyConfig.getProtocol().getInitialBytesToStrip()));
ch.pipeline().addLast(new ProxyMessageEncoder());
ch.pipeline().addLast(new IdleCheckHandler(proxyConfig.getProtocol().getReadIdleTime(), proxyConfig.getProtocol().getWriteIdleTime(), proxyConfig.getProtocol().getAllIdleTimeSeconds()));
ch.pipeline().addLast(new IdleStateHandler(proxyConfig.getProtocol().getReadIdleTime(), proxyConfig.getProtocol().getWriteIdleTime(), proxyConfig.getProtocol().getAllIdleTimeSeconds()));
ch.pipeline().addLast(new ClientChannelHandler());
}
});
neutrino-proxy-client/src/main/resources/application.yml
+4 -4
View File
@@ -9,12 +9,12 @@ neutrino:
length-field-length: 4
initial-bytes-to-strip: 0
length-adjustment: 0
read-idle-time: 60
write-idle-time: 30
read-idle-time: 40
write-idle-time: 8
all-idle-time-seconds: 0
client:
key-store-password: 123456
jks-path: classpath:/test.jks
server-ip: localhost
server-port: 9000
ssl-enable: false
server-port: 9002
ssl-enable: true
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/dal/LicenseMapper.java
+3
View File
@@ -68,6 +68,9 @@ public interface LicenseMapper extends SqlMapper {
@Update("update `license` set enable = :enable, update_time = :updateTime where id = :id")
void updateEnableStatus(@Param("id") Integer id, @Param("enable") Integer enable, @Param("updateTime") Date updateTime);
@Update("update `license` set is_online = :isOnline, update_time = :updateTime where id = :id")
void updateOnlineStatus(@Param("id") Integer id, @Param("isOnline") Integer isOnline, @Param("updateTime") Date updateTime);
@Update("update `license` set key = :key,update_time = :updateTime where id = :id")
void reset(@Param("id") Integer id, @Param("key") String key, @Param("updateTime") Date updateTime);
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/dal/PortMappingMapper.java
+9 -2
View File
@@ -34,6 +34,7 @@ import fun.asgc.neutrino.proxy.server.controller.req.PortMappingListReq;
import fun.asgc.neutrino.proxy.server.controller.res.PortMappingListRes;
import fun.asgc.neutrino.proxy.server.dal.entity.PortMappingDO;
import java.util.Date;
import java.util.List;
import java.util.Set;
@@ -57,8 +58,8 @@ public interface PortMappingMapper extends SqlMapper {
@Select("select * from port_mapping where id = ?")
PortMappingDO findById(Integer id);
@Update("update `port_mapping` set enable = :enable where id = :id")
void updateEnableStatus(@Param("id") Integer id, @Param("enable") Integer enable);
@Update("update `port_mapping` set enable = :enable,update_time = :updateTime where id = :id")
void updateEnableStatus(@Param("id") Integer id, @Param("enable") Integer enable, @Param("updateTime") Date updateTime);
@Delete("delete from `port_mapping` where id = ?")
void delete(Integer id);
@@ -72,4 +73,10 @@ public interface PortMappingMapper extends SqlMapper {
@ResultType(PortMappingDO.class)
@Select("select * from port_mapping where license_id = ? and enable = 1")
List<PortMappingDO> findEnableListByLicenseId(Integer licenseId);
@Update("update `port_mapping` set is_online = :isOnline,update_time = :updateTime where license_id = :licenseId and server_port = :serverPort")
void updateOnlineStatus(@Param("licenseId") Integer licenseId, @Param("serverPort") Integer serverPort, @Param("isOnline") Integer isOnline, @Param("updateTime") Date updateTime);
@Update("update `port_mapping` set is_online = :isOnline,update_time = :updateTime where license_id = :licenseId")
void updateOnlineStatus(@Param("licenseId") Integer licenseId, @Param("isOnline") Integer isOnline, @Param("updateTime") Date updateTime);
}
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/core/ProxyServerRunner.java
+2 -2
View File
@@ -28,7 +28,6 @@ import fun.asgc.neutrino.core.annotation.Component;
import fun.asgc.neutrino.core.annotation.NonIntercept;
import fun.asgc.neutrino.core.context.ApplicationRunner;
import fun.asgc.neutrino.core.util.FileUtil;
import fun.asgc.neutrino.proxy.core.IdleCheckHandler;
import fun.asgc.neutrino.proxy.core.ProxyMessageDecoder;
import fun.asgc.neutrino.proxy.core.ProxyMessageEncoder;
import fun.asgc.neutrino.proxy.server.base.proxy.ProxyConfig;
@@ -39,6 +38,7 @@ import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import javax.net.ssl.*;
@@ -137,7 +137,7 @@ public class ProxyServerRunner implements ApplicationRunner {
proxyConfig.getProtocol().getLengthFieldOffset(), proxyConfig.getProtocol().getLengthFieldLength(),
proxyConfig.getProtocol().getLengthAdjustment(), proxyConfig.getProtocol().getInitialBytesToStrip()));
ch.pipeline().addLast(new ProxyMessageEncoder());
ch.pipeline().addLast(new IdleCheckHandler(proxyConfig.getProtocol().getReadIdleTime(), proxyConfig.getProtocol().getWriteIdleTime(), proxyConfig.getProtocol().getAllIdleTimeSeconds()));
ch.pipeline().addLast(new IdleStateHandler(proxyConfig.getProtocol().getReadIdleTime(), proxyConfig.getProtocol().getWriteIdleTime(), proxyConfig.getProtocol().getAllIdleTimeSeconds()));
ch.pipeline().addLast(new ServerChannelHandler());
}
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/core/ServerChannelHandler.java
+29 -1
View File
@@ -27,15 +27,20 @@ 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.proxy.domain.CmdChannelAttachInfo;
import fun.asgc.neutrino.proxy.server.service.ProxyMutualService;
import fun.asgc.neutrino.proxy.server.util.ProxyUtil;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
/**
*
* @author: aoshiguchen
* @date: 2022/6/16
*/
@Slf4j
public class ServerChannelHandler extends SimpleChannelInboundHandler<ProxyMessage> {
private static volatile Dispatcher<ChannelHandlerContext, ProxyMessage> dispatcher;
@@ -77,13 +82,17 @@ public class ServerChannelHandler extends SimpleChannelInboundHandler<ProxyMessa
Channel cmdChannel = ProxyUtil.getCmdChannelByLicenseId(licenseId);
if (cmdChannel != null) {
ProxyUtil.removeUserChannelFromCmdChannel(cmdChannel, visitorId);
ProxyUtil.removeVisitorChannelFromCmdChannel(cmdChannel, visitorId);
}
// 数据发送完成后再关闭连接，解决http1.0数据传输问题
userChannel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
userChannel.close();
} else {
CmdChannelAttachInfo cmdChannelAttachInfo = ProxyUtil.getAttachInfo(ctx.channel());
if (null != cmdChannelAttachInfo) {
BeanManager.getBean(ProxyMutualService.class).offline(cmdChannelAttachInfo);
}
ProxyUtil.removeCmdChannel(ctx.channel());
}
@@ -94,4 +103,23 @@ public class ServerChannelHandler extends SimpleChannelInboundHandler<ProxyMessa
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if(evt instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent)evt;
switch (event.state()) {
case READER_IDLE:
// 读超时，断开连接
log.info("读超时");
ctx.channel().close();
break;
case WRITER_IDLE:
log.info("写超时");
break;
case ALL_IDLE:
break;
}
}
}
}
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/core/VisitorChannelHandler.java
+1 -1
View File
@@ -105,7 +105,7 @@ public class VisitorChannelHandler extends SimpleChannelInboundHandler<ByteBuf>
// 用户连接断开，从控制连接中移除
String userId = ProxyUtil.getVisitorChannelUserId(userChannel);
ProxyUtil.removeUserChannelFromCmdChannel(cmdChannel, userId);
ProxyUtil.removeVisitorChannelFromCmdChannel(cmdChannel, userId);
Channel proxyChannel = userChannel.attr(Constants.NEXT_CHANNEL).get();
if (proxyChannel != null && proxyChannel.isActive()) {
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageAuthHandler.java
+1 -1
View File
@@ -143,9 +143,9 @@ public class ProxyMessageAuthHandler implements ProxyMessageHandler {
for (PortMappingDO portMapping : portMappingList) {
try {
proxyMutualService.bindServerPort(cmdChannelAttachInfo, portMapping.getServerPort());
bootstrap.bind(portMapping.getServerPort()).get();
log.info("绑定用户端口： {}", portMapping.getServerPort());
proxyMutualService.bindServerPort(cmdChannelAttachInfo, portMapping.getServerPort());
} catch (Exception ex) {
// BindException表示该端口已经绑定过
if (!(ex.getCause() instanceof BindException)) {
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageDisconnectHandler.java
+1 -2
View File
@@ -29,7 +29,6 @@ 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.proxy.domain.VisitorChannelAttachInfo;
import fun.asgc.neutrino.proxy.server.util.ProxyUtil;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
@@ -55,7 +54,7 @@ public class ProxyMessageDisconnectHandler implements ProxyMessageHandler {
}
// 代理连接没有连上服务器由控制连接发送用户端断开连接消息
String visitorId = proxyMessage.getInfo();
Channel userChannel = ProxyUtil.removeUserChannelFromCmdChannel(ctx.channel(), visitorId);
Channel userChannel = ProxyUtil.removeVisitorChannelFromCmdChannel(ctx.channel(), visitorId);
if (null != userChannel) {
// 数据发送完成后再关闭连接，解决http1.0数据传输问题
userChannel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/service/PortMappingService.java
+1 -1
View File
@@ -186,7 +186,7 @@ public class PortMappingService {
ParamCheckUtil.checkExpression(!licenseDO.getUserId().equals(1), ExceptionConstant.NO_PERMISSION_VISIT);
}
portMappingMapper.updateEnableStatus(req.getId(), req.getEnable());
portMappingMapper.updateEnableStatus(req.getId(), req.getEnable(), new Date());
return new PortMappingUpdateEnableStatusRes();
}
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/service/ProxyMutualService.java
+20 -1
View File
@@ -24,10 +24,14 @@ package fun.asgc.neutrino.proxy.server.service;
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.proxy.server.constant.OnlineStatusEnum;
import fun.asgc.neutrino.proxy.server.dal.LicenseMapper;
import fun.asgc.neutrino.proxy.server.dal.PortMappingMapper;
import fun.asgc.neutrino.proxy.server.proxy.domain.CmdChannelAttachInfo;
import lombok.extern.slf4j.Slf4j;
import java.util.Date;
/**
* 代理交互服务
* @author: aoshiguchen
@@ -39,6 +43,8 @@ import lombok.extern.slf4j.Slf4j;
public class ProxyMutualService {
@Autowired
private PortMappingMapper portMappingMapper;
@Autowired
private LicenseMapper licenseMapper;
/**
* 绑定服务端端口处理
@@ -46,8 +52,21 @@ public class ProxyMutualService {
* @param serverPort
*/
public void bindServerPort(CmdChannelAttachInfo attachInfo, Integer serverPort) {
// TODO
Date now = new Date();
portMappingMapper.updateOnlineStatus(attachInfo.getLicenseId(), serverPort, OnlineStatusEnum.ONLINE.getStatus(), now);
licenseMapper.updateOnlineStatus(attachInfo.getLicenseId(), OnlineStatusEnum.ONLINE.getStatus(), now);
log.info("绑定服务端端口 licenseId:{},ip:{},serverPort:{}", attachInfo.getLicenseId(), attachInfo.getIp(), serverPort);
}
/**
* 客户端下线
* @param attachInfo
*/
public void offline(CmdChannelAttachInfo attachInfo) {
Date now = new Date();
portMappingMapper.updateOnlineStatus(attachInfo.getLicenseId(), OnlineStatusEnum.OFFLINE.getStatus(), now);
licenseMapper.updateOnlineStatus(attachInfo.getLicenseId(), OnlineStatusEnum.OFFLINE.getStatus(), now);
log.info("客户端下线 licenseId:{},ip:{}", attachInfo.getLicenseId(), attachInfo.getIp());
}
}
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/util/ProxyUtil.java
+1 -1
View File
@@ -190,7 +190,7 @@ public class ProxyUtil {
}
}
public static Channel removeUserChannelFromCmdChannel(Channel cmdChannel, String visitorId) {
public static Channel removeVisitorChannelFromCmdChannel(Channel cmdChannel, String visitorId) {
if (null == getAttachInfo(cmdChannel) || null == ((CmdChannelAttachInfo)getAttachInfo(cmdChannel)).getVisitorChannelMap().get(visitorId)) {
return null;
}
neutrino-proxy-server/src/main/resources/application.yml
+2 -2
View File
@@ -17,8 +17,8 @@ neutrino:
length-field-length: 4
initial-bytes-to-strip: 0
length-adjustment: 0
read-idle-time: 60
write-idle-time: 40
read-idle-time: 40
write-idle-time: 10
all-idle-time-seconds: 0
server:
port: 9000