发送消息到mq
This commit is contained in:
@@ -252,6 +252,30 @@ public class AgentEndpoint {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@AgentDispatcher(msgId = MsgEnum.容器包)
|
||||||
|
public class DockerHandler implements AgentMsgHandler {
|
||||||
|
@Override
|
||||||
|
public UpMsgResponse upHandle(String data, String clientId) {
|
||||||
|
JSONObject jsonObject = JSONObject.parseObject(data);
|
||||||
|
jsonObject.put("dataType",MsgEnum.容器包.getValue());
|
||||||
|
MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
|
||||||
|
agentProducer.asyncSend(mqMsg);
|
||||||
|
JSONObject json = new JSONObject();
|
||||||
|
json.put("resCode",1);
|
||||||
|
return UpMsgResponse.builder().clientId(clientId).content(json.toString()).build();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@AgentDispatcher(msgId = MsgEnum.容器包应答)
|
||||||
|
public class DockerRspHandler implements AgentMsgHandler {
|
||||||
|
@Override
|
||||||
|
public String downHandle(String data) {
|
||||||
|
StringBuilder res = new StringBuilder();
|
||||||
|
|
||||||
|
return res.toString();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
@AgentDispatcher(msgId = MsgEnum.交换机包)
|
@AgentDispatcher(msgId = MsgEnum.交换机包)
|
||||||
public class SwitchBoardHandler implements AgentMsgHandler {
|
public class SwitchBoardHandler implements AgentMsgHandler {
|
||||||
@Override
|
@Override
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package com.tongran.agenttcp.server.handler;
|
|||||||
import cn.hutool.cache.Cache;
|
import cn.hutool.cache.Cache;
|
||||||
import cn.hutool.cache.CacheUtil;
|
import cn.hutool.cache.CacheUtil;
|
||||||
import cn.hutool.core.date.DateUnit;
|
import cn.hutool.core.date.DateUnit;
|
||||||
|
import cn.hutool.core.util.ObjectUtil;
|
||||||
import com.alibaba.fastjson2.JSONObject;
|
import com.alibaba.fastjson2.JSONObject;
|
||||||
import com.tongran.agenttcp.server.annotation.AgentDispatcher;
|
import com.tongran.agenttcp.server.annotation.AgentDispatcher;
|
||||||
import com.tongran.agenttcp.server.basics.AgentDispatcherManager;
|
import com.tongran.agenttcp.server.basics.AgentDispatcherManager;
|
||||||
@@ -101,14 +102,16 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
|
|||||||
String clientId = jsonObject.getString("clientId");
|
String clientId = jsonObject.getString("clientId");
|
||||||
String dataType = jsonObject.getString("dataType");
|
String dataType = jsonObject.getString("dataType");
|
||||||
AgentMsgHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value);
|
AgentMsgHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value);
|
||||||
UpMsgResponse response = msgHandler.upHandle(message, clientId );
|
if (ObjectUtil.isNotEmpty(msgHandler)) {
|
||||||
AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}", response.getClientId(), dataType, message);
|
UpMsgResponse response = msgHandler.upHandle(message, clientId );
|
||||||
if(Objects.nonNull(response)){
|
AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}", response.getClientId(), dataType, message);
|
||||||
Message agentMessage = Message.builder().build();
|
if(Objects.nonNull(response)){
|
||||||
agentMessage.setClientId(response.getClientId());
|
Message agentMessage = Message.builder().build();
|
||||||
agentMessage.setDataType(dataType);
|
agentMessage.setClientId(response.getClientId());
|
||||||
agentMessage.setData(response.getContent());
|
agentMessage.setDataType(dataType);
|
||||||
ctx.fireChannelRead(agentMessage);//传递到下一个handler
|
agentMessage.setData(response.getContent());
|
||||||
|
ctx.fireChannelRead(agentMessage);//传递到下一个handler
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user