From 10a0b0ca34824307aa7d23b0ad6679b36bd57842 Mon Sep 17 00:00:00 2001
From: zhubaomin <zhubaomin>
Date: 星期二, 10 六月 2025 19:59:49 +0800
Subject: [PATCH] Merge branch 'master' of http://8.140.179.55:20000/r/pipIrr-SV
---
pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/ProtocolParserSdV1.java | 131 +++++++++++++++++++++++--------------------
1 files changed, 69 insertions(+), 62 deletions(-)
diff --git a/pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/ProtocolParserSdV1.java b/pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/ProtocolParserSdV1.java
index 1ca98b2..df625b1 100644
--- a/pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/ProtocolParserSdV1.java
+++ b/pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/ProtocolParserSdV1.java
@@ -3,9 +3,16 @@
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.dy.common.mw.protocol.Command;
-import com.dy.common.mw.protocol4Mqtt.pSdV1.downVos.ComCtrlVo;
+import com.dy.common.mw.protocol4Mqtt.MqttCallback;
+import com.dy.common.mw.protocol4Mqtt.MqttMsgParser;
+import com.dy.common.mw.protocol4Mqtt.MqttTopic;
+import com.dy.common.mw.protocol4Mqtt.Vo4Up;
+import com.dy.common.mw.protocol4Mqtt.pSdV1.comParam.ComCtrlVo;
+import com.dy.common.mw.protocol4Mqtt.pSdV1.downVos.FaultClearVo;
+import com.dy.common.mw.protocol4Mqtt.pSdV1.downVos.InjectStartVo;
+import com.dy.common.mw.protocol4Mqtt.pSdV1.downVos.StirStartVo;
+import com.dy.common.mw.protocol4Mqtt.pSdV1.upVos.*;
import org.eclipse.paho.client.mqttv3.MqttMessage;
-
/**
* @Author: liurunyu
@@ -13,44 +20,74 @@
* @Description
*/
public class ProtocolParserSdV1 {
- public MqttSubMsgSdV1 parseSubMsg(String deviceId, String topic, MqttMessage mqttMsg) throws Exception {
- MqttSubMsgSdV1 ms = new MqttSubMsgSdV1(deviceId, topic, new String(mqttMsg.getPayload(), "UTF-8"));
+ public MqttSubMsgSdV1 parseSubMsg(MqttTopic subTopic, MqttMessage mqttMsg, MqttCallback callback) throws Exception {
+ String msg = new String(mqttMsg.getPayload(), "UTF-8");
+ if(JSON.isValid(msg)){
+ throw new Exception("鎺ユ敹鍒癕QTT娑堟伅锛屽崗璁�" + subTopic.protocol + "锛岃澶嘔D" + subTopic.devId + "锛屼富棰�" + subTopic.topic + "娑堟伅鏍煎紡闈瀓son鏁版嵁(" + msg + ")") ;
+ }
+ MqttSubMsgSdV1 ms = new MqttSubMsgSdV1(subTopic, msg);
+ Vo4Up vo ;
+ switch (subTopic.topic) {
+ case ProtocolConstantSdV1.SubTopicWeather -> {
+ vo = JSON.parseObject(msg, WeatherVo.class);
+ break;
+ }
+ case ProtocolConstantSdV1.SubTopicSoil -> {
+ vo = JSON.parseObject(msg, SoilVo.class);
+ break;
+ }
+ case ProtocolConstantSdV1.SubTopicManure -> {
+ vo = JSON.parseObject(msg, ManureVo.class);
+ break;
+ }
+ case ProtocolConstantSdV1.SubTopicState -> {
+ //姝ゅ鏈畬鎴愶紝搴旇浜х敓涓�浜涢�氫俊鐨刬nfo锛屼緵涓嬮潰callback.notify(objs)閫氱煡鍑哄幓
+ vo = JSON.parseObject(msg, StateVo.class);
+ break;
+ }
+ default -> {
+ throw new Exception("鎺ユ敹鍒癕QTT娑堟伅锛屽崗璁�" + subTopic.protocol + "锛岃澶嘔D" + subTopic.devId + "锛屼富棰�" + subTopic.topic + "娑堟伅瑙ｆ瀽閫昏緫鏈疄鐜�");
+ }
+ }
+ ms.vo4Up = vo ;
+ callback.callback(ms);
+ callback.notify(null);//姝ゅ鏈畬鎴�
return ms;
}
public MqttPubMsgSdV1 createPubMsg(String orgTag, Command com) throws Exception {
- MqttPubMsgSdV1 msg = null ;
- switch (com.code){
- case CodeSdV1.cd_Fault:{
+ MqttPubMsgSdV1 msg ;
+ switch (com.code) {
+ case CodeSdV1.cd_Fault -> {
//鏁呴殰瑙ｉ櫎鍛戒护
this.checkParam(com);
this.checkRtnWebUrl(com);
- msg = this.createPubMsgOfFault(orgTag, com) ;
- break ;
+ msg = this.createPubMsgOfFault(orgTag, com);
+ break;
}
- case CodeSdV1.cd_Stir:{
+ case CodeSdV1.cd_Stir -> {
//鎼呮媽鍚仠鍛戒护
this.checkParam(com);
this.checkRtnWebUrl(com);
- msg = this.createPubMsgOfStir(orgTag, com) ;
- break ;
+ msg = this.createPubMsgOfStir(orgTag, com);
+ break;
}
- case CodeSdV1.cd_Inject:{
+ case CodeSdV1.cd_Inject -> {
//娉ㄨ偉鍚仠鍛戒护
this.checkParam(com);
this.checkRtnWebUrl(com);
- msg = this.createPubMsgOfInject(orgTag, com) ;
- break ;
+ msg = this.createPubMsgOfInject(orgTag, com);
+ break;
}
- case CodeSdV1.cd_Irr:{
+ case CodeSdV1.cd_Irr -> {
//鐏屾簤鍚仠鍛戒护
this.checkParam(com);
this.checkRtnWebUrl(com);
- msg = this.createPubMsgOfIrr(orgTag, com) ;
- break ;
+ msg = this.createPubMsgOfIrr(orgTag, com);
+ break;
}
- default:{
- throw new Exception("鎺ユ敹鍒癕QTT鍛戒护锛屽崗璁�" + com.protocol + "鐗堟湰" + com.protocolVersion + "鍔熻兘鐮�" + com.code + "鏋勯�犲櫒鏈疄鐜�") ;
+ default -> {
+ throw new Exception("鎺ユ敹鍒癕QTT鍛戒护锛屽崗璁�" + com.protocol + "鐗堟湰" + com.protocolVersion + "鍔熻兘鐮�" + com.code + "鏋勯�犲櫒鏈疄鐜�");
}
}
return msg ;
@@ -76,10 +113,9 @@
this.setPubMsgBase(com, msg);
msg.isCacheForOffLine = false ;
msg.hasResponse = true ;
- msg.address = 123 ;
- msg.value = "" + (cvo.isDo?1:0);
- msg.topic = createTopic(orgTag, com) ;
- msg.msg = "" ;
+ msg.cd = CodeSdV1.cd_Fault ;
+ msg.topic = MqttMsgParser.createPubTopic(new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicFault)) ;
+ msg.msg = JSON.toJSONString(new FaultClearVo(cvo.isDo)) ;
return msg ;
}
private MqttPubMsgSdV1 createPubMsgOfStir(String orgTag, Command com) throws Exception {
@@ -93,10 +129,9 @@
this.setPubMsgBase(com, msg);
msg.isCacheForOffLine = false ;
msg.hasResponse = true ;
- msg.address = 123 ;
- msg.value = "" + (cvo.isDo?1:0);
- msg.topic = createTopic(orgTag, com) ;
- msg.msg = "" ;
+ msg.cd = CodeSdV1.cd_Fault ;
+ msg.topic = MqttMsgParser.createPubTopic(new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicFault)) ;
+ msg.msg = JSON.toJSONString(new StirStartVo(cvo.isDo)) ;
return msg ;
}
private MqttPubMsgSdV1 createPubMsgOfInject(String orgTag, Command com) throws Exception {
@@ -110,10 +145,9 @@
this.setPubMsgBase(com, msg);
msg.isCacheForOffLine = false ;
msg.hasResponse = true ;
- msg.address = 123 ;
- msg.value = "" + (cvo.isDo?1:0);
- msg.topic = createTopic(orgTag, com) ;
- msg.msg = "" ;
+ msg.cd = CodeSdV1.cd_Fault ;
+ msg.topic = MqttMsgParser.createPubTopic(new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicFault)) ;
+ msg.msg = JSON.toJSONString(new InjectStartVo(cvo.isDo)) ;
return msg ;
}
private MqttPubMsgSdV1 createPubMsgOfIrr(String orgTag, Command com) throws Exception {
@@ -127,10 +161,9 @@
this.setPubMsgBase(com, msg);
msg.isCacheForOffLine = false ;
msg.hasResponse = true ;
- msg.address = 123 ;
- msg.value = "" + (cvo.isDo?1:0);
- msg.topic = createTopic(orgTag, com) ;
- msg.msg = "" ;
+ msg.cd = CodeSdV1.cd_Fault ;
+ msg.topic = MqttMsgParser.createPubTopic(new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicFault)) ;
+ msg.msg = JSON.toJSONString(new StirStartVo(cvo.isDo)) ;
return msg ;
}
@@ -140,31 +173,5 @@
msg.mqttResultSendWebUrl = com.rtuResultSendWebUrl ;
}
- private String createTopic(String orgTag, Command com){
- String topic = null ;
- switch (com.code){
- case CodeSdV1.cd_Fault:{
- topic = orgTag + "/" + com.protocol + com.protocolVersion + "/" + com.rtuAddr + "/control/m4" ;
- break ;
- }
- case CodeSdV1.cd_Stir:{
- topic = orgTag + "/" + com.protocol + com.protocolVersion + "/" + com.rtuAddr + "/control/m80" ;
- break ;
- }
- case CodeSdV1.cd_Inject:{
- topic = orgTag + "/" + com.protocol + com.protocolVersion + "/" + com.rtuAddr + "/control/m1" ;
- break ;
- }
- case CodeSdV1.cd_Irr:{
- topic = orgTag + "/" + com.protocol + com.protocolVersion + "/" + com.rtuAddr + "/control/m2" ;
- break ;
- }
- default:{
- topic = null ;
- break;
- }
- }
- return topic ;
- }
}
--
Gitblit v1.8.0