MQTT设备命令日志功能实现

liurunyu

2025-06-25 0909caac96ddac21f007d6c7ba6b348de9d2981a

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol/p206V202404/PrefixedDataAvailableV202404.java

New file
@@ -0,0 +1,136 @@
package com.dy.common.mw.protocol.p206V202404;

import com.dy.common.mw.channel.tcp.PrefixedDataAvailableStatus;
import com.dy.common.mw.protocol.AnnotationPrefixedDataAvailable;
import com.dy.common.mw.protocol.PrefixedDataAvailable;
import com.dy.common.util.ByteUtil;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.mina.core.buffer.BufferDataException;
import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.session.IoSession;


@AnnotationPrefixedDataAvailable(
      protocolName = ProtocolConstantV206V202404.protocolName,
      protocolVersion = ProtocolConstantV206V202404.protocolVer,
      priority = ProtocolConstantV206V202404.priority,
      onLineDataMinLength = ProtocolConstantV206V202404.onLineDataMinLength,
      headMinLength = ProtocolConstantV206V202404.headMinLength)
@SuppressWarnings("unused")
public class PrefixedDataAvailableV202404 implements PrefixedDataAvailable {

   private static final Logger log = LogManager.getLogger(PrefixedDataAvailableV202404.class) ;
	
   private static final String protocolName = ProtocolConstantV206V202404.protocolName ;
   private static final short protocolVersion = ProtocolConstantV206V202404.protocolVer ;


   /**
    * 分析上线数据(网络连接后第一包数据)是否可获得
    * @param in IoBuffer
    * @param remain 一定时大于0的数据
    * @param minDataLength 最小数据长度
    * @param maxDataLength 最大数据长度，达到或超过此长度，认为是垃圾数据。例如：数据头部是正确的，但合法数据结尾总不出现，认为此数据垃圾数据
    * @return 不是本协议数据时返回空
    * @throws Exception 异常
    */
   public PrefixedDataAvailableStatus forOnLine(IoSession ioSession,
                                     IoBuffer in,
                                     int remain,
                                     int minDataLength, //对应ProtocolConstantV2_0.onLineDataMinLength
                                     int maxDataLength //对应ProtocolConstant.errorMaxLength
         ) throws Exception {
		
      int oldPosition = in.position() ;
      byte[] preByte = new byte[remain];
      in.get(preByte) ;
      //in.position(0) ;//错误用法，如果发生粘包数据，将会死循环
      in.position(oldPosition) ;

      return this.doForData(ioSession, preByte, minDataLength, maxDataLength) ;
   }

   /**
    * 分析上线后(网络连接后非第一包数据)是否可获得
    * @param in IoBuffer
    * @param remain  剩余
    * @param minDataLength 最小数据长度
    * @param maxDataLength 最大数据长度
    * @return PrefixedDataAvailableStatus
    * @throws Exception 异常
    */
   @Override
   public PrefixedDataAvailableStatus forUpData(IoSession ioSession,
                                     IoBuffer in,
                                     int remain,
                                     int minDataLength,
                                     int maxDataLength) throws Exception {
      int oldPosition = in.position() ;
      byte[] preByte = new byte[remain];
      in.get(preByte) ;
      //in.position(0) ;//错误用法，如果发生粘包数据，将会死循环
      in.position(oldPosition) ;
		
      return this.doForData(ioSession, preByte, minDataLength, maxDataLength) ;

    }
   /**
    * 进行判断
    * @param ioSession ioSession
    * @param preByte byte[]
    * @param minDataLength 最小数据长度
    * @param maxDataLength 最大数据长度
    * @return PrefixedDataAvailableStatus
    * @throws Exception 异常
    */
   private PrefixedDataAvailableStatus doForData(IoSession ioSession, byte[] preByte, int minDataLength, int maxDataLength) throws Exception{
      int remain = preByte.length ;
    	
        if (remain < minDataLength) {
           log.warn("基于协议（" + protocolName + "）判断数据头部发生断包现象。") ;
            return new PrefixedDataAvailableStatus().breaked() ;
        }
        
        if(remain >= maxDataLength){
         //超出了最大长度限制,例如：数据头部是正确的，但合法数据结尾总不出现，认为此数据垃圾数据
         return new PrefixedDataAvailableStatus().rubbish(remain) ;
      }

		
        CommonV202404 parseCommon = new CommonV202404();
      boolean isThisProtocolData = parseCommon.isThisProtocolHead(preByte) ;

      if(!isThisProtocolData){
         //不是本协议数据
         return null ;
      }

      Integer frameLen = parseCommon.parseFrameLen(preByte) ;

      if(frameLen == null){
         String headHex = ByteUtil.bytes2Hex(preByte, true) ;
         throw new BufferDataException("收到数据之帧前部：" + headHex + "，但严重错误，在进行断包与粘包检查时，未能得到数据帧的长度。");
      }
		
        if (frameLen <= 0 || frameLen > maxDataLength) {
         String headHex = ByteUtil.bytes2Hex(preByte, true) ;
            throw new BufferDataException("收到数据之帧前部：" + headHex + "，但严重错误，在进行断包与粘包检查时，数据帧的长度(" + frameLen + ")超出合法范围。");
        }
        
        if(remain == frameLen){
           //不断不粘
            return new PrefixedDataAvailableStatus().completed(frameLen, protocolName, protocolVersion) ;
        }else if(remain > frameLen){
         String headHex = ByteUtil.bytes2Hex(preByte, true) ;
           log.warn("收到数据之帧前部：" + headHex + "，但发生粘包现象。") ;
            return new PrefixedDataAvailableStatus().adjoined(frameLen, protocolName, protocolVersion) ;
        }else{
           // remain < dataLen
         String headHex = ByteUtil.bytes2Hex(preByte, true) ;
           log.warn("收到数据之帧前部：" + headHex + "，但发生断包现象。") ;
            return new PrefixedDataAvailableStatus().breaked() ;
        }
   }

}