1、通信中间件重构MQTT相关下行命令的逻辑； 2、remote模块完善MQTT相关下行命令解除报警、搅拌启停、注肥启停、灌溉启停功能类；

liurunyu

2025-06-21 0c870ae5b2af348e19b10cc1a99f103f95c5a1cc

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/channel/tcp/DataDecoder.java

New file
@@ -0,0 +1,151 @@
package com.dy.common.mw.channel.tcp;

import com.dy.common.util.ByteUtil;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.mina.filter.codec.CumulativeProtocolDecoder;
import org.apache.mina.filter.codec.ProtocolDecoderOutput;
import org.apache.mina.core.buffer.* ;
import org.apache.mina.core.session.IoSession;


public class DataDecoder extends CumulativeProtocolDecoder {

   private static final Logger log = LogManager.getLogger(DataDecoder.class) ;

   protected PrefixedDataAvailableHandle pdaHandle ;
	
   public DataDecoder(PrefixedDataAvailableHandle pdaHandle) {
      this.pdaHandle = pdaHandle ;
   }

   /**
    * 解码
    * (non-Javadoc)
    * @see org.apache.mina.filter.codec.CumulativeProtocolDecoder#doDecode(
    *    org.apache.mina.core.session.IoSession, 
    *    org.apache.mina.core.buffer.IoBuffer, 
    *    org.apache.mina.filter.codec.ProtocolDecoderOutput)
    */
   @Override
   protected boolean doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) {
      String rtuAddr = (String)session.getAttribute(TcpIoSessionAttrIdIsRtuAddr.sessionArrRtuAddr) ;
      if(rtuAddr == null){
         //会话未被管理
         //会话管理器中不存在此会话，说明刚建立网络连接，
         return this.doDecode_onLineData(session, in, out) ;
      }else{
         //会话已被管理
         //会话管理器中存在此会话，说明已经上线了，
         return this.doDecode_data(session, in, out, rtuAddr) ;
      }
   }
	
   /**
    * 针对网络连接成功后第一包数据
    * @param session IO会话
    * @param in 输入Buffer
    * @param out 协议输出编码
    * @return 是否正好或粘包
    */
   @SuppressWarnings("unused")
   private boolean doDecode_onLineData(IoSession session, IoBuffer in, ProtocolDecoderOutput out){
      //有一些协议，负责发上线数据，但上线(或心跳)数据中无 Rtu地址，所以这样的数据只能放行，而且不能产生有身份(Rtu地址)的网络会话，即不能把session放入会话缓存中
      //有一些协议，Rtu负责发上线数据，上线数据中有 Rtu地址，所以这样的数据放行后，能产生有身份(Rtu地址)的网络会话，即能把session放入会话缓存中
      //不论何种情形，上线数据的数据量不会很大，一般不会产生断包，所以这里只进行简单断包检查。
      PrefixedDataAvailableStatus dataStatus = this.pdaHandle.forOnLine(session, in) ;

      if(dataStatus.protocolName != null && dataStatus.protocolVersion != null){
         session.setAttribute(TcpIoSessionAttrIdIsRtuAddr.sessionArrProtocolName, dataStatus.protocolName) ;
         session.setAttribute(TcpIoSessionAttrIdIsRtuAddr.sessionArrProtocolVersion, dataStatus.protocolVersion) ;
      }

      if(dataStatus.isCompleted() || dataStatus.isAdjoined()){
         //正好或粘包
         this.nextDeal(in, dataStatus.getDataLen(), out) ;
         return true;
      }else if(dataStatus.isRubbish()){
         //垃圾数据
         this.dealRubbishData(in, dataStatus.getDataLen()) ;
         return true ;
      }else{
         //断包
         return false ;
      }
   }
	

   /**
    * 针对网络连接成功后非第一包数据
    * @param session IO会话
    * @param in 输入Buffer
    * @param out 协议输出编码
    * @param rtuAddr 控制器地址
    * @return 是否正好或粘包
    */
   @SuppressWarnings("unused")
   private boolean doDecode_data(IoSession session, IoBuffer in, ProtocolDecoderOutput out, String rtuAddr) {
      //非上线数据，可能会出现断包或粘包现象
      PrefixedDataAvailableStatus dataStatus = this.pdaHandle.forUpData(session, in) ;
      if(dataStatus == null){
         //不可能发生
         log.error("严重错误，Rtu (RTU" + rtuAddr + ")上行数据完整性检查时，返回的对象为空。") ;
         this.nextDeal(in, null, out) ;
         return true;
      }else{
         if(dataStatus.isBreaked()){
            //断包了
            return false ;
         }else if(dataStatus.isCompleted() || dataStatus.isAdjoined()){
            //本包数据已经全部接收，或可能粘有下包数据
            this.nextDeal(in, dataStatus.getDataLen(), out) ;
            if(dataStatus.isAdjoined()){
               //说明粘包了，还有数据，需要对这些数据再次执行doDecode_方法.
               return this.doDecode_data(session, in, out, rtuAddr) ;//加上递归
            }else if(dataStatus.isCompleted()){
               //数据不断不粘
               return true;
            }else{
               //不存在此种情况
               return true ;
            }
         }else if(dataStatus.isRubbish()){
            //垃圾数据
            this.dealRubbishData(in, dataStatus.getDataLen()) ;
            return true ;
         }else{
            //不存在此种情况
            return true ;
         }
      }
   }	
    
   /**
    * 后续处理数据
    * @param in IObuffer
    * @param length 长度
    * @param out 协议编码输出
    */
   private void nextDeal(IoBuffer in, Integer length, ProtocolDecoderOutput out){
      if(length == null){
         length = in.limit() ;
      }
      if(length > 0){
         byte[] data = new byte[length];
         in.get(data);//get一个字节，相应position向后移动一个字节
         out.write(data);
      }
   }
	
   /**
    * 垃圾数据处理
    * @param in IObuffer
    * @param length 长度
    */
   @SuppressWarnings("unused")
   private void dealRubbishData(IoBuffer in, Integer length){
      byte[] data = new byte[in.limit()];
      in.get(data);//get一个字节，相应position向后移动一个字节
      log.error("抛弃垃圾数据：" + ByteUtil.bytes2Hex(data, true));
   }
}