初始化
This commit is contained in:
qiminbao
@@ -0,0 +1,126 @@
|
||||
package com.tongran.agenttcp.server.handler;
|
||||
|
||||
import cn.hutool.cache.Cache;
|
||||
import cn.hutool.cache.CacheUtil;
|
||||
import cn.hutool.core.date.DateUnit;
|
||||
import com.alibaba.fastjson2.JSONObject;
|
||||
import com.tongran.agenttcp.server.annotation.AgentDispatcher;
|
||||
import com.tongran.agenttcp.server.basics.AgentDispatcherManager;
|
||||
import com.tongran.agenttcp.server.basics.AgentMsgHandler;
|
||||
import com.tongran.agenttcp.server.model.Message;
|
||||
import com.tongran.agenttcp.server.model.UpMsgResponse;
|
||||
import com.tongran.agenttcp.server.netty.core.session.SessionManager;
|
||||
import com.tongran.agenttcp.utils.AssertLog;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.channel.ChannelHandler;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.ChannelInboundHandlerAdapter;
|
||||
import io.netty.util.CharsetUtil;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
|
||||
@Component
|
||||
@ChannelHandler.Sharable
|
||||
public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
|
||||
|
||||
protected final SessionManager sessionManager;
|
||||
|
||||
@Resource
|
||||
private AgentDispatcherManager agentDispatcherManager;
|
||||
|
||||
public AgentDecoderHandler() {
|
||||
this.sessionManager = SessionManager.getInstance();
|
||||
}
|
||||
|
||||
// 用来临时保留没有处理过的请求报文
|
||||
private final Cache<String, String> lruCache = CacheUtil.newLRUCache(3000);
|
||||
|
||||
/**
|
||||
* 消息解码器
|
||||
*/
|
||||
@Override
|
||||
public void channelRead(ChannelHandlerContext ctx, Object msg) {
|
||||
// 检查是否为 ByteBuf（未解码的原始数据）
|
||||
if (msg instanceof ByteBuf) {
|
||||
ByteBuf byteBuf = (ByteBuf) msg;
|
||||
String messages = byteBuf.toString(CharsetUtil.UTF_8); // 指定字符集解码
|
||||
// AssertLog.info("<<[up1]:IP:{},[up-content]==>{}", sessionManager.client(ctx),messages);
|
||||
|
||||
String content = "";
|
||||
boolean isClear = false;
|
||||
String tempMsg = lruCache.get(sessionManager.client(ctx));
|
||||
tempMsg = tempMsg == null ? "" : tempMsg;
|
||||
int tmpMsgSize = tempMsg.length();
|
||||
|
||||
boolean startsWith = messages.startsWith("agent-tcp:");
|
||||
boolean endsWith = messages.endsWith("@tong-ran");
|
||||
String ms = messages.replaceAll("agent-tcp:","");
|
||||
String[] arr_msg = ms.split("@tong-ran");
|
||||
if(startsWith){
|
||||
//判定是否整包
|
||||
if(arr_msg.length == 1){
|
||||
//判定是否拆包
|
||||
if(endsWith){
|
||||
lruCache.remove(sessionManager.client(ctx));
|
||||
tmpMsgSize = 0;
|
||||
content = arr_msg[0]+"@tong-ran";
|
||||
}else{
|
||||
lruCache.remove(sessionManager.client(ctx));
|
||||
tmpMsgSize = 0;
|
||||
content = arr_msg[0];
|
||||
lruCache.put(sessionManager.client(ctx), content, DateUnit.SECOND.getMillis() * 3000);
|
||||
}
|
||||
}
|
||||
//判定是否粘包
|
||||
if(arr_msg.length > 1){
|
||||
if(!endsWith){
|
||||
lruCache.remove(sessionManager.client(ctx));
|
||||
tmpMsgSize = 0;
|
||||
content = arr_msg[0]+"@tong-ran";
|
||||
lruCache.put(sessionManager.client(ctx), arr_msg[1], DateUnit.SECOND.getMillis() * 3000);
|
||||
}else{
|
||||
lruCache.remove(sessionManager.client(ctx));
|
||||
tmpMsgSize = 0;
|
||||
content = String.join("@tong-ran", arr_msg);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (tmpMsgSize > 0) {
|
||||
content = tempMsg + messages;
|
||||
isClear = true;
|
||||
}
|
||||
AssertLog.info("<<[up]:IP:{},[up-content]==>{}", sessionManager.client(ctx),content);
|
||||
endsWith = content.endsWith("@tong-ran");
|
||||
//判断是否是整包
|
||||
if(endsWith){
|
||||
String[] arr = content.split("@tong-ran");
|
||||
for (String message : arr) {
|
||||
JSONObject jsonObject = JSONObject.parseObject(message);
|
||||
String dataType = jsonObject.getString("dataType");
|
||||
AgentMsgHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value);
|
||||
UpMsgResponse response = msgHandler.upHandle(message);
|
||||
AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}", response.getClientId(), dataType, message);
|
||||
Message agentMessage = Message.builder().build();
|
||||
agentMessage.setClientId(response.getClientId());
|
||||
agentMessage.setDataType(dataType);
|
||||
agentMessage.setData(response.getContent());
|
||||
ctx.fireChannelRead(agentMessage);//传递到下一个handler
|
||||
}
|
||||
}
|
||||
if (isClear) {
|
||||
lruCache.remove(sessionManager.client(ctx));
|
||||
}
|
||||
byteBuf.release(); // 释放 ByteBuf 资源（重要！）
|
||||
} else {
|
||||
System.out.println("Unexpected message type: " + msg.getClass());
|
||||
}
|
||||
}
|
||||
|
||||
//拆包
|
||||
public byte[] subByte(byte[] b, int off, int length) {
|
||||
byte[] bytes = new byte[length];
|
||||
System.arraycopy(b, off, bytes, 0, length);
|
||||
return bytes;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
package com.tongran.agenttcp.server.handler;
|
||||
|
||||
import com.tongran.agenttcp.server.model.Message;
|
||||
import com.tongran.agenttcp.server.netty.core.session.Session;
|
||||
import com.tongran.agenttcp.server.netty.core.session.SessionManager;
|
||||
import com.tongran.agenttcp.utils.AssertLog;
|
||||
import io.netty.channel.ChannelHandler;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.SimpleChannelInboundHandler;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
@Component
|
||||
@ChannelHandler.Sharable
|
||||
public class AgentDispatcherHandler extends SimpleChannelInboundHandler<Message> {
|
||||
|
||||
protected final SessionManager sessionManager;
|
||||
|
||||
public AgentDispatcherHandler() {
|
||||
this.sessionManager = SessionManager.getInstance();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void channelRead0(ChannelHandlerContext ctx, Message msg) {
|
||||
// AssertLog.info(">>>>>>[要处理的终端数据] {}", msg.toString());
|
||||
String clientId = msg.getClientId();
|
||||
if (StringUtils.isBlank(clientId)) {
|
||||
AssertLog.error("<<<<<<错误的信息from:{}", ctx.channel().remoteAddress());
|
||||
return;
|
||||
}
|
||||
Session session = Session.buildSession(ctx, clientId);
|
||||
sessionManager.put(clientId, session);
|
||||
if (StringUtils.isNotBlank(msg.getData())) {
|
||||
// AssertLog.info(">>>>>>[getSessionById的终端数据] {}", sessionManager.getSessionById(clientId));
|
||||
sessionManager.writeAndFlush(sessionManager.getSessionById(clientId).getChannel(), msg);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
package com.tongran.agenttcp.server.handler;
|
||||
|
||||
import com.alibaba.fastjson2.JSON;
|
||||
import com.tongran.agenttcp.server.model.Message;
|
||||
import com.tongran.agenttcp.server.netty.core.session.SessionManager;
|
||||
import com.tongran.agenttcp.utils.AssertLog;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.buffer.Unpooled;
|
||||
import io.netty.channel.ChannelHandler;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.ChannelOutboundHandlerAdapter;
|
||||
import io.netty.channel.ChannelPromise;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
@Component
|
||||
@ChannelHandler.Sharable
|
||||
public class AgentEncoderHandler extends ChannelOutboundHandlerAdapter {
|
||||
|
||||
protected final SessionManager sessionManager;
|
||||
|
||||
public AgentEncoderHandler() {
|
||||
this.sessionManager = SessionManager.getInstance();
|
||||
}
|
||||
|
||||
/**
|
||||
* 消息解码器
|
||||
*/
|
||||
@Override
|
||||
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
|
||||
if (msg instanceof Message) {
|
||||
Message entity = (Message) msg;
|
||||
if (StringUtils.isBlank(entity.getData())) {
|
||||
AssertLog.error(">>[down]:IP:{},errorContent:{}", sessionManager.client(ctx), msg);
|
||||
return;
|
||||
}
|
||||
AssertLog.info(">>[down]:IP:{},[content]==>{}", sessionManager.client(ctx), entity.getData());
|
||||
String json = JSON.toJSONString(entity);
|
||||
byte[] bytes = json.getBytes(StandardCharsets.UTF_8); // 显式指定 UTF-8
|
||||
// byte[] bytes = EscapeUtil.hexStringToByteArray(entity.getContent());
|
||||
ByteBuf buf = Unpooled.wrappedBuffer(bytes);
|
||||
ctx.write(buf, promise);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user