初始化

2025-09-10 15:36:49 +08:00
commit 4330ffdff1
75 changed files with 6674 additions and 0 deletions
@@ -0,0 +1,141 @@
+package com.tongran.agent.client.netty.handler;
+
+import cn.hutool.cache.Cache;
+import cn.hutool.cache.CacheUtil;
+import cn.hutool.core.date.DateUnit;
+import cn.hutool.core.util.ObjectUtil;
+import com.alibaba.fastjson2.JSONObject;
+import com.tongran.agent.client.core.session.SessionManager;
+import com.tongran.agent.client.netty.annotation.AgentDispatcher;
+import com.tongran.agent.client.netty.basics.AgentDispatcherManager;
+import com.tongran.agent.client.netty.basics.AgentHandler;
+import com.tongran.agent.client.netty.model.Message;
+import com.tongran.agent.client.netty.model.UpMsgResponse;
+import com.tongran.agent.client.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;
+import java.util.Objects;
+
+@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-server:");
+			boolean endsWith = messages.endsWith("@tong-ran");
+			String ms = messages.replaceAll("agent-server:","");
+			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;
+				endsWith = content.endsWith("@tong-ran");
+				if(endsWith){
+					isClear = true;
+				}else{
+					lruCache.remove(sessionManager.client(ctx));
+					tmpMsgSize = 0;
+					lruCache.put(sessionManager.client(ctx), content, DateUnit.SECOND.getMillis() * 3000);
+				}
+
+			}
+			AssertLog.info("<<[up2]: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 clientId = jsonObject.getString("clientId");
+					String dataType = jsonObject.getString("dataType");
+					AgentHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value);
+					if (ObjectUtil.isNotEmpty(msgHandler)) {
+						UpMsgResponse response = msgHandler.upHandle(message, clientId );
+						AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}", response.getClientId(), dataType, message);
+						if(Objects.nonNull(response)){
+							Message agentMessage = Message.builder().build();
+							agentMessage.setClientId(response.getClientId());
+							agentMessage.setDataType(response.getDataType());
+							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.agent.client.netty.handler;
+
+import com.tongran.agent.client.core.session.Session;
+import com.tongran.agent.client.core.session.SessionManager;
+import com.tongran.agent.client.netty.model.Message;
+import com.tongran.agent.client.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,49 @@
+package com.tongran.agent.client.netty.handler;
+
+import com.alibaba.fastjson2.JSON;
+import com.tongran.agent.client.core.session.SessionManager;
+import com.tongran.agent.client.netty.model.Message;
+import com.tongran.agent.client.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);
+            String json = "agent-client:"+ JSON.toJSONString(entity)+"@tong-ran";
+            byte[] bytes = json.getBytes(StandardCharsets.UTF_8); // 显式指定 UTF-8
+//            byte[] bytes = EscapeUtil.hexStringToByteArray(entity.getContent());
+            ByteBuf buf = Unpooled.wrappedBuffer(bytes);
+            ctx.write(buf, promise);
+        }
+    }
+
+}
@@ -0,0 +1,78 @@
+package com.tongran.agent.client.netty.handler;
+
+import com.tongran.agent.client.core.config.GlobalConfig;
+import com.tongran.agent.client.core.session.SessionManager;
+import com.tongran.agent.client.scheduler.service.AppInitializer;
+import com.tongran.agent.client.utils.AssertLog;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.timeout.IdleState;
+import io.netty.handler.timeout.IdleStateEvent;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+import java.io.IOException;
+
+/**
+ * @Description: TCP消息适配器
+ */
+@Component
+@ChannelHandler.Sharable
+public class TCPListenHandler extends ChannelInboundHandlerAdapter {
+
+	@Resource
+	private AppInitializer appInitializer;
+
+	private final SessionManager sessionManager;
+
+	public TCPListenHandler() {
+		this.sessionManager = SessionManager.getInstance();
+	}
+
+	@Override
+	public void channelActive(ChannelHandlerContext ctx) {
+		AssertLog.info("<<<<<<[终端连接]{}", ctx.channel().remoteAddress());
+	}
+
+	@Override
+	public void channelInactive(ChannelHandlerContext ctx) {
+		AssertLog.info(">>>>>>[断开连接]{}", sessionManager.client(ctx));
+		sessionManager.remove(ctx.channel());
+		try {
+			GlobalConfig.isCollect = false;
+			appInitializer.run();
+		} catch (Exception e) {
+			e.printStackTrace();
+		}
+	}
+
+	@Override
+	public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) {
+		if (e instanceof IOException) {
+			AssertLog.info("<<<<<<[终端断开连接]{} {}", sessionManager.client(ctx), e.getMessage());
+		} else {
+			AssertLog.info(">>>>>>[消息处理异常]" + sessionManager.client(ctx), e);
+		}
+	}
+
+	@Override
+	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
+		if (evt instanceof IdleStateEvent) {
+			IdleStateEvent event = (IdleStateEvent) evt;
+			IdleState state = event.state();
+			if (state == IdleState.READER_IDLE || state == IdleState.WRITER_IDLE || state == IdleState.ALL_IDLE) {
+				AssertLog.warn(">>>>>>[终端心跳超时]{} {}", state, sessionManager.client(ctx));
+				sessionManager.remove(ctx.channel());
+				ctx.close();
+				try {
+					GlobalConfig.isCollect = false;
+					appInitializer.run();
+				} catch (Exception e) {
+					e.printStackTrace();
+				}
+			}
+		}
+	}
+
+}
@@ -0,0 +1,24 @@
+package com.tongran.agent.client.netty.handler;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.socket.DatagramPacket;
+import org.springframework.stereotype.Component;
+
+/**
+ * @Description: UDP消息适配器
+ */
+@Component
+@ChannelHandler.Sharable
+public class UDPListenHandler extends ChannelInboundHandlerAdapter {
+
+	@Override
+	public void channelRead(ChannelHandlerContext ctx, Object msg) {
+		DatagramPacket packet = (DatagramPacket) msg;
+		ByteBuf buf = packet.content();
+		buf.clear();// 暂未实现
+	}
+
+}