采用模拟软件MQTTX进行模拟测试，发现并修改MQTT相关水肥机上行数据与下行命令逻辑bug.

liurunyu

2025-06-23 431bbfaddba9f39496359bc0c1e2dcaf2bb6fca2

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/local/CommandInnerDeaLer.java

@@ -3,7 +3,14 @@
import com.dy.common.mw.UnitCallbackInterface;
import com.dy.common.mw.channel.tcp.TcpUnit;
import com.dy.common.mw.protocol.Command;
import com.dy.common.mw.protocol.rtuState.RtuStatus;
import com.dy.rtuMw.server.local.localProtocol.*;
import com.dy.rtuMw.server.mqtt.DevStatus;
import com.dy.rtuMw.server.mqtt.DevStatusDealer;
import com.dy.rtuMw.server.mqtt.MqttUnit;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author liurunyu
@@ -18,29 +25,109 @@
     * @return
     */
    public Command deal(Command com) throws Exception{
        Command rCom ;
        String code = com.getCode() ;
        if(code.equals(CodeLocal.clock)){
            return this.clock(com) ;
        }else if(code.equals(CodeLocal.onAllLine)){
            return this.onAllLine(com) ;
        }else if(code.equals(CodeLocal.onPartLine)){
            return this.onPartLine(com) ;
        }else if(code.equals(CodeLocal.onLineStatistics)){
            return this.onLineStateStatistics(com) ;
        }else if(code.equals(CodeLocal.allProtocols)){
            return this.allProtocols(com) ;
        }else if(code.equals(CodeLocal.stopTcpSv)){
            return this.stopTcpSv(com) ;
        }else if(code.equals(CodeLocal.recoverTcpSv)){
            return this.recoverTcpSv(com) ;
        }else if(code.equals(CodeLocal.mwState)){
            return this.mwInfo(com) ;
        switch (code) {
            case CodeLocal.clock -> {
                rCom = this.clock(com);
                break;
            }
            case CodeLocal.mwState -> {
                rCom = this.mwInfo(com);
                break;
            }

            ////////////////////////////////////////////
            //
            // 以下是相关基于TCP连接的RTU设备的内部命令
            //
            ////////////////////////////////////////////
            case CodeLocal.onAllLine -> {
                rCom = this.onAllLine(com);
                break;
            }
            case CodeLocal.onPartLine -> {
                rCom = this.onPartLine(com);
                break;
            }
            case CodeLocal.onLineStatistics -> {
                rCom = this.onLineStateStatistics(com);
                break;
            }
            case CodeLocal.allRtuStates -> {
                rCom = this.allRtuStates(com);
                break;
            }
            case CodeLocal.partRtuStates -> {
                rCom = this.someRtuStates(com);
                break;
            }
            case CodeLocal.oneRtuStates -> {
                rCom = this.oneRtuStates(com);
                break;
            }
            case CodeLocal.allProtocols -> {
                rCom = this.allProtocols(com);
                break;
            }
            case CodeLocal.stopTcpSv -> {
                rCom = this.stopTcpSv(com);
                break;
            }
            case CodeLocal.recoverTcpSv -> {
                rCom = this.recoverTcpSv(com);
                break;
            }


            ////////////////////////////////////////////
            //
            // 以下是相关基于MQTT连接的设备的内部命令
            //
            ////////////////////////////////////////////
            case CodeLocal.onAllLineMqtt -> {
                rCom = this.onAllLineMqtt(com);
                break;
            }
            case CodeLocal.onPartLineMqtt -> {
                rCom = this.onPartLineMqtt(com);
                break;
            }
            case CodeLocal.onLineStatisticsMqtt -> {
                rCom = this.onLineStateStatisticsMqtt(com);
                break;
            }
            case CodeLocal.allRtuStatesMqtt -> {
                rCom = this.allRtuStatesMqtt(com);
                break;
            }
            case CodeLocal.partRtuStatesMqtt -> {
                rCom = this.someRtuStatesMqtt(com);
                break;
            }
            case CodeLocal.oneRtuStatesMqtt -> {
                rCom = this.oneRtuStatesMqtt(com);
                break;
            }
            case CodeLocal.stopMqttSv -> {
                rCom = this.stopMqttSv(com);
                break;
            }
            case CodeLocal.recoverMqttSv -> {
                rCom = this.recoverMqttSv(com);
                break;
            }
            default -> {
                rCom = ReturnCommand.errored("出错，收到内部命令的功能码不能识别！", com.getId(), com.getCode());
                break;
            }
        }
        return ReturnCommand.errored("出错，收到内部命令的功能码不能识别！", com.getId(), com.getCode()) ;
        return rCom ;
    }

    /**
     * 查询通信中间件时钟
     * @param  command
     * @throws Exception
     */
    private Command clock(Command command) throws Exception{
@@ -53,8 +140,8 @@
     * @throws Exception
     */
    private Command onAllLine(Command command) throws Exception{
        RtuOnLineVo ol = new RtuOnLineDeal().dealAll() ;
        return ReturnCommand.successed("查询所有RTU在线情况结果", command.getId(), command.getCode(), ol) ;
        HashMap<String, Boolean> map = new RtuOnLineDeal().dealAll() ;
        return ReturnCommand.successed("查询所有RTU在线情况结果", command.getId(), command.getCode(), map) ;
    }

    /**
@@ -64,8 +151,8 @@
    private Command onPartLine(Command command) throws Exception{
        if(command.param != null && command.param instanceof String && !command.param.equals("")){
            String[] rtuAddrGrp = ((String)command.param).split(",");
            RtuOnLineVo ol = new RtuOnLineDeal().dealPart(rtuAddrGrp) ;
            return ReturnCommand.successed("查询部分RTU在线情况结果", command.getId(), command.getCode(), ol) ;
            HashMap<String, Boolean> map = new RtuOnLineDeal().dealPart(rtuAddrGrp) ;
            return ReturnCommand.successed("查询部分RTU在线情况结果", command.getId(), command.getCode(), map) ;
        }else{
            return ReturnCommand.errored("出错，命令参数应该是所查询RTU的地址串",  command.getId(), command.getCode()) ;
        }
@@ -78,6 +165,43 @@
    private Command onLineStateStatistics(Command command) throws Exception{
        RtuOnLineStateStatisticsVo vo = new RtuOnLineStateStatisticsDeal().deal() ;
        return ReturnCommand.successed("查询所有RTU在线情况结果", command.getId(), command.getCode(), vo) ;
    }

    /**
     * 查询所有RTU状态
     * @throws Exception
     */
    private Command allRtuStates(Command command) throws Exception{
        Map<String, RtuStatus> map =  new RtuStatusDeal().dealAll() ;
        return ReturnCommand.successed("查询所有RTU状态结果", command.getId(), command.getCode(), map) ;
    }

    /**
     * 查询部分RTU状态
     * @throws Exception
     */
    private Command someRtuStates(Command command) throws Exception{
        if(command.param != null && command.param instanceof String && !command.param.equals("")){
            String[] rtuAddrGrp = ((String)command.param).split(",");
            Map<String, RtuStatus> map = new RtuStatusDeal().dealSome(rtuAddrGrp) ;
            return ReturnCommand.successed("查询部分RTU状态结果", command.getId(), command.getCode(), map) ;
        }else{
            return ReturnCommand.errored("出错，命令参数应该有所查询RTU的地址串",  command.getId(), command.getCode()) ;
        }
    }

    /**
     * 查询部分RTU状态
     * @throws Exception
     */
    private Command oneRtuStates(Command command) throws Exception{
        if(command.param != null && command.param instanceof String && !command.param.equals("")){
            String rtuAddr = (String)command.param;
            RtuStatus rtuStatus = new RtuStatusDeal().dealOne(rtuAddr) ;
            return ReturnCommand.successed("查询一个RTU状态结果", command.getId(), command.getCode(), rtuStatus) ;
        }else{
            return ReturnCommand.errored("出错，命令参数应该有所查询RTU的地址",  command.getId(), command.getCode()) ;
        }
    }

    /**
@@ -111,6 +235,99 @@
    }


    /**
     * 查询所有MQTT设备在线情况
     * @throws Exception
     */
    private Command onAllLineMqtt(Command command) throws Exception{
        HashMap<String, Boolean> map = DevStatusDealer.allOnLine() ;
        return ReturnCommand.successed("查询所有Mqtt设备在线情况结果", command.getId(), command.getCode(), map) ;
    }

    /**
     * 查询部分MQTT设备在线情况
     * @throws Exception
     */
    private Command onPartLineMqtt(Command command) throws Exception{
        if(command.param != null && command.param instanceof String && !command.param.equals("")){
            String[] devIds = ((String)command.param).split(",");
            HashMap<String, Boolean> map = DevStatusDealer.partOnLine(devIds) ;
            return ReturnCommand.successed("查询部分Mqtt设备在线情况结果", command.getId(), command.getCode(), map) ;
        }else{
            return ReturnCommand.errored("出错，命令参数应该有所查询Mqtt设备的地址串",  command.getId(), command.getCode()) ;
        }
    }

    /**
     * 统计MQTT设备在线与不在线情况
     * @throws Exception
     */
    private Command onLineStateStatisticsMqtt(Command command) throws Exception{
        RtuOnLineStateStatisticsVo vo = DevStatusDealer.statisticsOnLine() ;
        return ReturnCommand.successed("查询所有Mqtt设备在线情况结果", command.getId(), command.getCode(), vo) ;
    }

    /**
     * 查询所有MQTT设备状态
     * @throws Exception
     */
    private Command allRtuStatesMqtt(Command command) throws Exception{
        Map<String, DevStatus> map =  DevStatusDealer.allStatus() ;
        return ReturnCommand.successed("查询所有Mqtt设备状态结果", command.getId(), command.getCode(), map) ;
    }

    /**
     * 查询部分MQTT设备状态
     * @throws Exception
     */
    private Command someRtuStatesMqtt(Command command) throws Exception{
        if(command.param != null && command.param instanceof String && !command.param.equals("")){
            String[] devIds = ((String)command.param).split(",");
            Map<String, DevStatus> map = DevStatusDealer.someStatus(devIds) ;
            return ReturnCommand.successed("查询部分Mqtt设备状态结果", command.getId(), command.getCode(), map) ;
        }else{
            return ReturnCommand.errored("出错，命令参数应该有所查询Mqtt设备的地址串",  command.getId(), command.getCode()) ;
        }
    }

    /**
     * 查询部分MQTT设备状态
     * @throws Exception
     */
    private Command oneRtuStatesMqtt(Command command) throws Exception{
        if(command.param != null && command.param instanceof String && !command.param.equals("")){
            String devId = (String)command.param;
            DevStatus devStatus = DevStatusDealer.oneStatus(devId) ;
            return ReturnCommand.successed("查询一个Mqtt设备状态结果", command.getId(), command.getCode(), devStatus) ;
        }else{
            return ReturnCommand.errored("出错，命令参数应该有所查询Mqtt设备的地址",  command.getId(), command.getCode()) ;
        }
    }
    /**
     * 停止MQTT服务
     * @throws Exception
     */
    private Command stopMqttSv(Command command) throws Exception{
        MqttUnit.getInstance().stop(new UnitCallbackInterface(){
            public void call(Object obj) throws Exception {
            }
        });
        return ReturnCommand.successed("已经启动停止Mqtt服务", command.getId(), command.getCode(), null) ;
    }


    /**
     * 恢复MQTT服务
     * @throws Exception
     */
    private Command recoverMqttSv(Command command) throws Exception{
        MqttUnit.getInstance().recover(new UnitCallbackInterface(){
            public void call(Object obj) throws Exception {
            }
        });
        return ReturnCommand.successed("已经启动恢复Mqtt服务", command.getId(), command.getCode(), null) ;
    }


    /**
     * 查询通信中间件运行情况