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

liurunyu

2025-06-23 431bbfaddba9f39496359bc0c1e2dcaf2bb6fca2

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/forTcp/RtuStatusDealer.java

@@ -1,146 +1,162 @@
package com.dy.rtuMw.server.forTcp;

import com.dy.common.mw.protocol.rtuState.RtuStatus;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.dy.common.mw.protocol.rtuStatus.StCommandResult;
import com.dy.common.mw.protocol.rtuStatus.StDownData;
import com.dy.common.mw.protocol.rtuStatus.StOnOffLine;
import com.dy.common.mw.protocol.rtuStatus.StUpData;
import java.util.HashMap;
import java.util.Map;

public class RtuStatusDealer {
   private static Logger log = LogManager.getLogger(RtuStatusDealer.class.getName()) ;
	
	
   public static void onOffLine(RtuSessionStatus sta){
      StOnOffLine st = new StOnOffLine() ;
      st.rtuAddr = sta.rtuAddr ;
      st.onOff_trueFalse = sta.onTrueOffLine ;
      st.ip = sta.ip ;
      st.port = sta.port ;
		
//      if(member == null){
//         log.error("出错，未能得到处理RTU（地址为" + sta.rtuAddr + "）状态数据的数据中间件，可能数据中间件未启动或脱离的集群！");
//      }else{
//         log.info("通知数据中间件（" + member.toString() + "），RTU（地址为" + sta.rtuAddr + "）在线！");
//         new JgIoHandler().sendP2P(member, st);
//      }

   private static Map<String, RtuStatus> map = new HashMap<String, RtuStatus>() ;

   /**
     * 得到全部状态
     * @return
     */
   public static Map<String, RtuStatus> allStatus(){
      return map ;
   }
	
   public static void onLine(String rtuAddr, String ip, Integer port){
      StOnOffLine st = new StOnOffLine() ;
      st.rtuAddr = rtuAddr ;
      st.onOff_trueFalse = true ;
      st.ip = ip ;
      st.port = port ;
		
//      Address member = JgUnit.getInstance().getJgMembers().getDataMwMemberByRtuAddr(rtuAddr) ;
//      if(member == null){
//         log.error("出错，未能得到处理RTU（地址为" + rtuAddr + "）状态数据的数据中间件，可能数据中间件未启动或脱离的集群！");
//      }else{
//         log.info("通知数据中间件（" + member.toString() + "），RTU（地址为" + rtuAddr + "）在线！");
//         new JgIoHandler().sendP2P(member, st);
//      }
   /**
    * 得到部分状态
    * @return
    */
   public static Map<String, RtuStatus> someStatus(String[] rtuAddrArrGrp){
      synchronized (map){
         Map<String, RtuStatus> rsMap = new HashMap<>();
         for(String rtuAddr : rtuAddrArrGrp){
            RtuStatus status = map.get(rtuAddr) ;
            if(status != null){
               rsMap.put(rtuAddr, status) ;
            }
         }
         return rsMap ;
      }
   }
   /**
    * 得到一个RTU的状态
    * @return
    */
   public static RtuStatus oneStatus(String rtuAddr){
      return map.get(rtuAddr) ;
   }


   public static void onLine(String rtuAddr, String ip, Integer port, String protocolName, Short protocolVersion){
      RtuStatus vo = map.get(rtuAddr) ;
      if(vo == null) {
         vo = new RtuStatus();
         vo.onLine(rtuAddr, ip, port, protocolName, protocolVersion) ;
         map.put(rtuAddr, vo);
      }else {
         vo.onLine(rtuAddr, ip, port, protocolName, protocolVersion) ;
      }
   }
   
   public static void offLine(String rtuAddr){
      StOnOffLine st = new StOnOffLine() ;
      st.rtuAddr = rtuAddr ;
      st.onOff_trueFalse = false ;
		
//      Address member = JgUnit.getInstance().getJgMembers().getDataMwMemberByRtuAddr(rtuAddr) ;
//      if(member == null){
//         log.error("出错，未能得到处理RTU（地址为" + rtuAddr + "）状态数据的数据中间件，可能数据中间件未启动或脱离的集群！");
//      }else{
//         log.info("通知数据中间件（" + member.toString() + "），RTU（地址为" + rtuAddr + "）离线！");
//         new JgIoHandler().sendP2P(member, st);
//      }
      RtuStatus vo = map.get(rtuAddr) ;
      if(vo == null) {
         vo = new RtuStatus();
         vo.offLine();
         map.put(rtuAddr, vo);
      }else {
         vo.offLine();
      }
   }
   
   public static void upData(String rtuAddr, Integer bufferLen){
      StUpData st = new StUpData() ;
      st.rtuAddr = rtuAddr ;
      st.upBufferLen = bufferLen ;
      st.isReport = false ;
		
//      Address member = JgUnit.getInstance().getJgMembers().getDataMwMemberByRtuAddr(rtuAddr) ;
//      if(member == null){
//         log.error("出错，未能得到处理RTU（地址为" + rtuAddr + "）状态数据的数据中间件，可能数据中间件未启动或脱离的集群！");
//      }else{
//         log.info("通知数据中间件（" + member.toString() + "），RTU（地址为" + rtuAddr + "）上行数据！");
//         new JgIoHandler().sendP2P(member, st);
//      }
   public static void upData(String rtuAddr, String code, String codeName, Integer bufferLen){
      RtuStatus vo = map.get(rtuAddr) ;
      if(vo == null) {
         vo = new RtuStatus();
         vo.upData(code, codeName, bufferLen);
         map.put(rtuAddr, vo);
      }else {
         vo.upData(code, codeName, bufferLen);
      }
   }
   
   public static void upReport(String rtuAddr, Integer bufferLen){
      StUpData st = new StUpData() ;
      st.rtuAddr = rtuAddr ;
      st.upBufferLen = bufferLen ;
      st.isReport = true ;
		
//      Address member = JgUnit.getInstance().getJgMembers().getDataMwMemberByRtuAddr(rtuAddr) ;
//      if(member == null){
//         log.error("出错，未能得到处理RTU（地址为" + rtuAddr + "）状态数据的数据中间件，可能数据中间件未启动或脱离的集群！");
//      }else{
//         log.info("通知数据中间件（" + member.toString() + "），RTU（地址为" + rtuAddr + "）上报数据！");
//         new JgIoHandler().sendP2P(member, st);
//      }
   public static void upAutoReport(String rtuAddr, String code, String codeName, Integer bufferLen){
      RtuStatus vo = map.get(rtuAddr) ;
      if(vo == null) {
         vo = new RtuStatus();
         vo.upData(code, codeName, bufferLen);
         map.put(rtuAddr, vo);
      }else {
         vo.upData(code, codeName, bufferLen);
      }
   }
   
   public static void downData(String rtuAddr, Integer bufferLen){
      StDownData st = new StDownData() ;
      st.rtuAddr = rtuAddr ;
      st.downBufferLen = bufferLen ;
		
//      Address member = JgUnit.getInstance().getJgMembers().getDataMwMemberByRtuAddr(rtuAddr) ;
//      if(member == null){
//         log.error("出错，未能得到处理RTU（地址为" + rtuAddr + "）状态数据的数据中间件，可能数据中间件未启动或脱离的集群！");
//      }else{
//         log.info("通知数据中间件（" + member.toString() + "），RTU（地址为" + rtuAddr + "）下行数据！");
//         new JgIoHandler().sendP2P(member, st);
//      }
   public static void downData(String rtuAddr, String code, String codeName, Integer bufferLen){
      RtuStatus vo = map.get(rtuAddr) ;
      if(vo == null) {
         vo = new RtuStatus();
         vo.downData(code, codeName, bufferLen);
         map.put(rtuAddr, vo);
      }else {
         vo.downData(code, codeName, bufferLen);
      }
   }
   
   public static void commandSuccess(String rtuAddr){
      StCommandResult st = new StCommandResult() ;
      st.rtuAddr = rtuAddr ;
      st.isSuccess = true ;
		
//      Address member = JgUnit.getInstance().getJgMembers().getDataMwMemberByRtuAddr(rtuAddr) ;
//      if(member == null){
//         log.error("出错，未能得到处理RTU（地址为" + rtuAddr + "）状态数据的数据中间件，可能数据中间件未启动或脱离的集群！");
//      }else{
//         log.info("通知数据中间件（" + member.toString() + "），RTU（地址为" + rtuAddr + "）命令成功！");
//         new JgIoHandler().sendP2P(member, st);
//      }
   public static void commandSuccess(String rtuAddr, String code, String codeName){
      RtuStatus vo = map.get(rtuAddr) ;
      if(vo == null) {
         vo = new RtuStatus();
         vo.commandSuccess(code, codeName);
         map.put(rtuAddr, vo);
      }else {
         vo.commandSuccess(code, codeName);
      }
   }
   
   public static void commandFailure(String rtuAddr){
      StCommandResult st = new StCommandResult() ;
      st.rtuAddr = rtuAddr ;
      st.isSuccess = false ;
		
//      Address member = JgUnit.getInstance().getJgMembers().getDataMwMemberByRtuAddr(rtuAddr) ;
//      if(member == null){
//         log.error("出错，未能得到处理RTU（地址为" + rtuAddr + "）状态数据的数据中间件，可能数据中间件未启动或脱离的集群！");
//      }else{
//         log.info("通知数据中间件（" + member.toString() + "），RTU（地址为" + rtuAddr + "）命令失败！");
//         new JgIoHandler().sendP2P(member, st);
//      }
   public static void commandFailure(String rtuAddr, String code, String codeName){
      RtuStatus vo = map.get(rtuAddr) ;
      if(vo == null) {
         vo = new RtuStatus();
         vo.commandFailure(code, codeName);
         map.put(rtuAddr, vo);
      }else {
         vo.commandFailure(code, codeName);
      }
   }
	

   /**
    * 命令结果收到了，但未在缓存匹配上下行的命令，说明下行缓存命令因超时被清除，这说明有个命令由失败变成功了
    * @param rtuAddr
    */
   public static void commandFail2Success(String rtuAddr){
      StCommandResult st = new StCommandResult() ;
      st.rtuAddr = rtuAddr ;
      st.fail2Success = true ;
		
//      Address member = JgUnit.getInstance().getJgMembers().getDataMwMemberByRtuAddr(rtuAddr) ;
//      if(member == null){
//         log.error("出错，未能得到处理RTU（地址为" + rtuAddr + "）状态数据的数据中间件，可能数据中间件未启动或脱离的集群！");
//      }else{
//         log.info("通知数据中间件（" + member.toString() + "），RTU（地址为" + rtuAddr + "）命令由失败转成功！");
//         new JgIoHandler().sendP2P(member, st);
//      }
      //暂时无逻辑
   }

   /**
    * 收到了成功开阀报（开阀自报，开阀工作报，远程开阀（执行）应答等，都认为阀门是开启状态（这时上报的数据中阀门状态可能是关闭（因为打开阀门需要时间）））
    * @param rtuAddr
    */
   public static void valveOpen(String rtuAddr){
      RtuStatus vo = map.get(rtuAddr) ;
      if(vo == null) {
         vo = new RtuStatus();
         vo.valveOpen();
         map.put(rtuAddr, vo);
      }else {
         vo.valveOpen();
      }
   }

   /**
    * 收到了成功关阀报（关阀自报，远程关阀（执行）应答等，都认为阀门是关闭状态（这时上报的数据中阀门状态可能是开启（因为关闭阀门需要时间）））
    * @param rtuAddr
    */
   public static void valveClose(String rtuAddr){
      RtuStatus vo = map.get(rtuAddr) ;
      if(vo == null) {
         vo = new RtuStatus();
         vo.valveClose();
         map.put(rtuAddr, vo);
      }else {
         vo.valveClose();
      }
   }

}