进行ApiFox发送内部命令测试，MQTTX发布气象数据测试，修改测试中发现的bug，修改不完善的地方。

liurunyu

4 天以前 21c080aa9da3acd53e014e8f917b50a48cb791cb

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/ProtocolParserSdV1.java

@@ -3,9 +3,17 @@
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 com.dy.common.mw.protocol4Mqtt.status.DevRunSt;
import org.eclipse.paho.client.mqttv3.MqttMessage;


/**
 * @Author: liurunyu
@@ -13,44 +21,81 @@
 * @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"));
        return ms;
    public MqttSubMsgSdV1 parseSubMsg(MqttTopic subTopic, MqttMessage mqttMsg, MqttCallback callback) throws Exception {
        String strTxt = new String(mqttMsg.getPayload(), "UTF-8");
        if(!JSON.isValid(strTxt)){
            throw new Exception("接收到MQTT消息，协议" + subTopic.protocol + "，设备ID" + subTopic.devId + "，主题" + subTopic.longName() + "消息格式非json数据(" + strTxt + ")") ;
        }
        MqttSubMsgSdV1 msg = new MqttSubMsgSdV1(subTopic, strTxt);
        Vo4Up vo ;
        DevRunSt stVo ;
        switch (subTopic.topic) {
            case ProtocolConstantSdV1.SubTopicWeather -> {
                vo = JSON.parseObject(strTxt, WeatherVo.class);
                break;
            }
            case ProtocolConstantSdV1.SubTopicSoil -> {
                vo = JSON.parseObject(strTxt, SoilVo.class);
                break;
            }
            case ProtocolConstantSdV1.SubTopicManure -> {
                vo = JSON.parseObject(strTxt, ManureVo.class);
                break;
            }
            case ProtocolConstantSdV1.SubTopicState -> {
                //此处未完成，应该产生一些通信的info，供下面callback.notify(objs)通知出去
                vo = JSON.parseObject(strTxt, StateVo.class);
                stVo = new DevRunSt() ;
                stVo.id = msg.deviceId ;
                //stVo.stirRunning = true ; //搅拌运行 true是 false否
                //stVo.injectRunning = true ; //注肥运行 true是 false否
                //stVo.irrRunning = true ; //灌溉运行 true是 false否
                //stVo.alarm = true ; //报警 true是 false否
                break;
            }
            default -> {
                throw new Exception("接收到MQTT消息，协议" + subTopic.protocol + "，设备ID" + subTopic.devId + "，主题" + subTopic.topic + "消息解析逻辑未实现");
            }
        }
        msg.vo4Up = vo ;
        callback.callback(msg);
        callback.notify(null);//此处未完成
        return msg;
    }

    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("接收到MQTT命令，协议" + com.protocol + "版本" + com.protocolVersion + "功能码" + com.code + "构造器未实现") ;
            default -> {
                throw new Exception("接收到MQTT命令，协议" + com.protocol + "版本" + com.protocolVersion + "功能码" + com.code + "构造器未实现");
            }
        }
        return msg ;
@@ -76,10 +121,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 +137,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 +153,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 +169,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 +181,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 ;
    }

}