diff --git a/pom.xml b/pom.xml index 46b1682..701f994 100644 --- a/pom.xml +++ b/pom.xml @@ -92,6 +92,15 @@ org.springframework.boot spring-boot-maven-plugin + + + org.apache.maven.plugins + maven-surefire-plugin + ${maven-surefire-plugin.version} + + true + + diff --git a/src/main/java/com/tongran/agent/server/controller/TcpController.java b/src/main/java/com/tongran/agent/server/controller/TcpController.java index 26e7291..04233ec 100644 --- a/src/main/java/com/tongran/agent/server/controller/TcpController.java +++ b/src/main/java/com/tongran/agent/server/controller/TcpController.java @@ -1,6 +1,6 @@ package com.tongran.agent.server.controller; -import com.alibaba.fastjson2.JSON; +import com.alibaba.fastjson2.JSONObject; import com.tongran.agent.server.netty.model.Message; import org.springframework.validation.annotation.Validated; import org.springframework.web.bind.annotation.PostMapping; @@ -15,17 +15,13 @@ public class TcpController { @Resource private TcpService tcpService; - @PostMapping("/create") - public void create(@Validated @RequestBody Message dto) { - tcpService.create(dto.getClientId()); - } - @PostMapping("/send") public void sendMessage(@Validated @RequestBody Message dto) { - String json = JSON.toJSONString(dto); -// ScriptPolicy policy = JSON.parseObject(dto.getData(), ScriptPolicy.class); -// System.out.println(policy.getFiles().get(0).getFileData()); - tcpService.sendMessage(json); + JSONObject jsonObject = new JSONObject(); + jsonObject.put("clientId", dto.getClientId()); + jsonObject.put("dataType", dto.getDataType()); + jsonObject.put("data", dto.getData()); + tcpService.sendMessage(jsonObject.toString()); } diff --git a/src/main/java/com/tongran/agent/server/controller/TcpService.java b/src/main/java/com/tongran/agent/server/controller/TcpService.java index 4f442e6..be24582 100644 --- a/src/main/java/com/tongran/agent/server/controller/TcpService.java +++ b/src/main/java/com/tongran/agent/server/controller/TcpService.java @@ -10,6 +10,8 @@ import com.tongran.agent.server.netty.annotation.AgentDispatcher; import com.tongran.agent.server.netty.basics.AgentDispatcherManager; import com.tongran.agent.server.netty.basics.AgentHandler; import com.tongran.agent.server.netty.model.Message; +import com.tongran.agent.server.rockermq.AgentProducer; +import com.tongran.agent.server.rockermq.model.MqMsg; import com.tongran.agent.server.utils.AssertLog; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; @@ -29,6 +31,9 @@ public class TcpService { @Resource private MultiTargetNettyClient client; + @Resource + private AgentProducer agentProducer; + public TcpService() { this.sessionManager = SessionManager.getInstance(); } @@ -60,67 +65,58 @@ public class TcpService { AssertLog.info("consumer==> received down message: {}", message); try { Message dto = JSON.parseObject(message, Message.class); - if(Objects.nonNull(dto)){ - if(StringUtils.equals(dto.getDataType(), MsgEnum.注册.getValue())){ - JSONObject jsonObject = JSONObject.parseObject(dto.getData()); - String clientIp = ""; - int port = 0; - long timestamp = 0; - if(jsonObject.containsKey("clientIp")){ - clientIp = jsonObject.getString("clientIp"); + if(StringUtils.equals(dto.getDataType(), MsgEnum.注册.getValue())){ + JSONObject jsonObject = JSONObject.parseObject(dto.getData()); + String clientIp = ""; + int clientPort = 0; + String switchBoard = ""; + long timestamp = 0; + if(jsonObject.containsKey("clientIp")){ + clientIp = jsonObject.getString("clientIp"); + } + if(jsonObject.containsKey("clientPort")){ + clientPort = jsonObject.getInteger("clientPort"); + } + if(jsonObject.containsKey("switchBoard")){ + switchBoard = jsonObject.getString("switchBoard"); + } + if(jsonObject.containsKey("timestamp")){ + timestamp = jsonObject.getLongValue("timestamp"); + } + if(StringUtils.isBlank(clientIp) || clientPort == 0){ + //返回提醒ip 端口 空 + JSONObject json = new JSONObject(); + json.put("clientId",dto.getClientId()); + json.put("dataType",MsgEnum.注册应答.getValue()); + JSONObject dataJson = new JSONObject(); + dataJson.put("resCode",0); + dataJson.put("resMag","客户端IP、端口不能为空"); + dataJson.put("timestamp", System.currentTimeMillis()); + json.put("data",dataJson.toString()); + MqMsg mqMsg = MqMsg.builder().topic("tongran_agent_down").content(json.toString()).build(); + agentProducer.asyncSend(mqMsg); + return; + } + boolean success = client.createConnection(dto.getClientId(), clientIp, clientPort, 5); + if(success){ + //连接成功 + // 连接建立后可以发送登录认证消息 + JSONObject object = new JSONObject(); + object.put("timestamp", timestamp); + if(StringUtils.isNotBlank(switchBoard)){ + object.put("switchBoard",switchBoard); } - if(jsonObject.containsKey("port")){ - port = jsonObject.getInteger("port"); - } - if(jsonObject.containsKey("timestamp")){ - timestamp = jsonObject.getLongValue("timestamp"); - } - if(StringUtils.isBlank(clientIp) || port == 0){ - //返回提醒ip 端口 空 - return; - } - boolean success = client.createConnection(dto.getClientId(), clientIp, port, 5); - if(success){ - String community = ""; - String ip = ""; - int switchPort = 0; - String oids = ""; - String params = ""; - if(jsonObject.containsKey("switchBoard")){ - String switchBoard = jsonObject.getString("switchBoard"); - jsonObject = JSONObject.parseObject(switchBoard); - community = jsonObject.getString("community"); - ip = jsonObject.getString("ip"); - switchPort = jsonObject.getInteger("port"); - oids = jsonObject.getString("oids"); - params = jsonObject.getString("params"); - } - //连接成功 - // 连接建立后可以发送登录认证消息 - JSONObject object = new JSONObject(); - object.put("clientId",dto.getClientId()); - object.put("timestamp",timestamp); - JSONObject switchBoard = new JSONObject(); - switchBoard.put("community", community); - switchBoard.put("ip", ip); - switchBoard.put("port", switchPort); - switchBoard.put("oids", oids); - switchBoard.put("params", params); - object.put("switchBoard",switchBoard.toString()); - Message msg = Message.builder().clientId(dto.getClientId()).dataType(MsgEnum.注册.getValue()).data(object.toString()).build(); - // 将对象转为 JSON 字符串 - client.sendMessages(dto.getClientId(), msg); - }else{ - //连接失败 - } - } else { - AgentHandler msgHandler = agentDispatcherManager.getHandler(dto.getDataType() + "&" + AgentDispatcher.VersionEnum.V1.value); - if (ObjectUtil.isNotEmpty(msgHandler)) { - String msg = msgHandler.downHandle(dto.getData(), dto.getClientId()); - Message chargeMessage = Message.builder().clientId(dto.getClientId()).dataType(dto.getDataType()).data(msg).build(); - if (Objects.nonNull(sessionManager.getSessionById(dto.getClientId()))) { - sessionManager.writeAndFlush(sessionManager.getSessionById(dto.getClientId()).getChannel(), chargeMessage); - } + Message msg = Message.builder().clientId(dto.getClientId()).dataType(MsgEnum.注册.getValue()).data(object.toString()).build(); + // 将对象转为 JSON 字符串 + client.sendMessages(dto.getClientId(), msg); + } + } else { + AgentHandler msgHandler = agentDispatcherManager.getHandler(dto.getDataType() + "&" + AgentDispatcher.VersionEnum.V1.value); + if (ObjectUtil.isNotEmpty(msgHandler)) { + String msg = msgHandler.downHandle(dto.getData(), dto.getClientId()); + Message chargeMessage = Message.builder().clientId(dto.getClientId()).dataType(dto.getDataType()).data(msg).build(); + if (Objects.nonNull(sessionManager.getSessionById(dto.getClientId()))) { + sessionManager.writeAndFlush(sessionManager.getSessionById(dto.getClientId()).getChannel(), chargeMessage); } } } diff --git a/src/main/java/com/tongran/agent/server/core/enums/MsgEnum.java b/src/main/java/com/tongran/agent/server/core/enums/MsgEnum.java index ff63566..b35e04f 100644 --- a/src/main/java/com/tongran/agent/server/core/enums/MsgEnum.java +++ b/src/main/java/com/tongran/agent/server/core/enums/MsgEnum.java @@ -22,6 +22,8 @@ public enum MsgEnum { 心跳上报("HEARTBEAT"), + 心跳上报应答("HEARTBEAT_RSP"), + CPU上报("CPU"), 磁盘上报("DISK"), diff --git a/src/main/java/com/tongran/agent/server/netty/enpoint/AgentEndpoint.java b/src/main/java/com/tongran/agent/server/netty/enpoint/AgentEndpoint.java index ae086c6..2b2b854 100644 --- a/src/main/java/com/tongran/agent/server/netty/enpoint/AgentEndpoint.java +++ b/src/main/java/com/tongran/agent/server/netty/enpoint/AgentEndpoint.java @@ -81,7 +81,9 @@ public class AgentEndpoint { jsonObject.put("data", data); MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build(); agentProducer.asyncSend(mqMsg); - return null; + JSONObject json = new JSONObject(); + json.put("resCode",1); + return UpMsgResponse.builder().clientId(clientId).dataType(MsgEnum.心跳上报应答.getValue()).content(json.toString()).build(); } } diff --git a/src/main/java/com/tongran/agent/server/netty/handler/AgentDecoderHandler.java b/src/main/java/com/tongran/agent/server/netty/handler/AgentDecoderHandler.java index c0f76bc..46905ed 100644 --- a/src/main/java/com/tongran/agent/server/netty/handler/AgentDecoderHandler.java +++ b/src/main/java/com/tongran/agent/server/netty/handler/AgentDecoderHandler.java @@ -121,8 +121,8 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter { if(Objects.nonNull(agentDispatcherManager)){ AgentHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value); if (ObjectUtil.isNotEmpty(msgHandler)) { - UpMsgResponse response = msgHandler.upHandle(data, clientId ); AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}", clientId, dataType, data); + UpMsgResponse response = msgHandler.upHandle(data, clientId ); if(Objects.nonNull(response)){ Message agentMessage = Message.builder().build(); agentMessage.setClientId(clientId); diff --git a/src/main/resources/application-dev.yml b/src/main/resources/application-dev.yml index 143403c..3da7cc2 100644 --- a/src/main/resources/application-dev.yml +++ b/src/main/resources/application-dev.yml @@ -1,7 +1,7 @@ server: - port: 9018 + port: 7012 servlet: - context-path: /agent-server + context-path: /tr-agent-server # 接口文档配置 knife4j: @@ -12,9 +12,13 @@ knife4j: logging: file: # path: /usr/local/tongran/logs - path: /data/agent-server/logs + path: /data/tr-agent-server/logs # path: D:/job/agent-logs/logs +rocketmq: + name-server: 172.16.15.103:9876 + enabled: true + netty: server: # host: 172.16.15.103 diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 166f7c3..3e7f3cb 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -5,7 +5,7 @@ spring: pathmatch: matching-strategy: ant_path_matcher application: - name: agent-server + name: tr-agent-server version: 1.0 web: resources: