Merge branch 'master' of http://8.140.179.55:20000/r/pipIrr-SV

Administrator

2023-12-20 dd694b8a3d05e2d785413182e1c28b25a5b3f431

 pipIrr-platform/pipIrr-mw/pipIrr-mw-accept/src/main/java/com/dy/aceMw/server/tasks/RtuUpTask.java

@@ -14,19 +14,19 @@
import com.dy.common.mw.protocol.Driver;
import com.dy.common.mw.protocol.OnLine;
import com.dy.common.mw.protocol.OnLineHandle;
import com.dy.common.mw.protocol.ProtocolCach;
import com.dy.common.mw.protocol.ProtocolCache;
import com.dy.aceMw.server.ServerProperties;
import com.dy.aceMw.server.forTcp.RtuLogDealer;
import com.dy.aceMw.server.forTcp.RtuStatusDealer;
import com.dy.aceMw.server.forTcp.TcpSessionCach;
import com.dy.aceMw.server.forTcp.TcpSessionCache;
import com.dy.common.util.ByteUtil;

public class RtuUpTask extends CoreTask {
   
   private static Logger log = LogManager.getLogger(RtuUpTask.class.getName());
   private static final Logger log = LogManager.getLogger(RtuUpTask.class.getName());

   @Override
   public Integer excute() {
   public Integer execute() {
      Object[] os = (Object[])this.data ;
      IoSession session = (IoSession)os[0] ;
      byte[] upBuf = (byte[])os[1] ;
@@ -40,8 +40,8 @@

   /**
    * RTU上行数据
    * @param session
    * @param upBuf
    * @param session IO会话
    * @param upBuf 上行数据
    */
   private void upData(IoSession session, byte[] upBuf) throws Exception{
      if(upBuf == null){
@@ -63,7 +63,7 @@
         //说明刚建立网络连接，此数据应该是上线数据
         isOnLine = true ;
         String[] rtuAddrProtocolName = this.parseOnLine(session, upBuf) ;
         if(rtuAddrProtocolName != null && rtuAddrProtocolName.length == 2){
         if(rtuAddrProtocolName.length == 2){
            //解析上线数据成功，并解析出RTU地址及通信协议名称
            rtuAddr = rtuAddrProtocolName[0] ;
            protocolName = rtuAddrProtocolName[1] ;
@@ -72,7 +72,7 @@
               //设置session的属性ID
               TcpUnit.getInstance().setIoSessionArrId(session, rtuAddr);
               //缓存session
               TcpSessionCach.putNewTcpSession(rtuAddr, protocolName, session);
               TcpSessionCache.putNewTcpSession(rtuAddr, protocolName, session);
               
               log.info("RTU(地址：" + rtuAddr + ")上线了。") ; 
            }
@@ -81,11 +81,11 @@
      
      if(rtuAddr != null){
         if(protocolName == null){
            protocolName = TcpSessionCach.getTcpProtocolName(rtuAddr) ;
            protocolName = TcpSessionCache.getTcpProtocolName(rtuAddr) ;
         }
         
         //设置收到数据时刻
         TcpSessionCach.cachUpDataTime(rtuAddr);
         TcpSessionCache.cacheUpDataTime(rtuAddr);
         
         if(protocolName != null){
            //对上行数据进行处理
@@ -96,8 +96,8 @@
   
   /**
    * 解析上线数据
    * @param session
    * @param upBuf
    * @param session IO会话
    * @param upBuf 上行数据
    */
   private String[] parseOnLine(IoSession session, byte[] upBuf){
      String rtuAddr = null ;
@@ -135,45 +135,47 @@
   
   /**
    * 处理上行数据
    * @param session
    * @param rtuAddr
    * @param upBuf
    * @param upHex
    * @throws Exception 
    * @param session IO会话
    * @param rtuAddrAtHead 控制器地址头部
    * @param protocolName 协议名称
    * @param isOnLine 是否上线数据
    * @param upBuf 上行数据
    * @param upHex 上行数据
    * @throws Exception 异常
    */
   private void dealUpData(IoSession session, String rtuAddrAtHead, String protocolName, boolean isOnLine, byte[] upBuf, String upHex) throws Exception{
      Driver dri = ProtocolCach.getDriver(protocolName) ;
      Driver dri = ProtocolCache.getDriver(protocolName) ;
      if(dri == null){
         log.error("严重错误，未能得到协议" + protocolName + "驱动类实例！");
      }else{
         MidResult[] midRses = dri.parseData(ServerProperties.isLowPower, rtuAddrAtHead, upBuf, upHex, new DriverParserDataCallback(){
         MidResult[] midRs = dri.parseData(ServerProperties.isLowPower, rtuAddrAtHead, upBuf, upHex, new DriverParserDataCallback(){
            @Override
            public void callback(String rtuAddrAtHead, String meterNoAtHead, String code, String upHex, Boolean reportOrResponse_trueOrFalse, boolean parseFail, String rtuAddrInData, String meterNoInData) {
            public void callback(String rtuAddrAtHead, String code, String upHex, Boolean reportOrResponse_trueOrFalse, boolean parseFail, String rtuAddrInData) {
               //更新终端状态
               if(rtuAddrInData != null && !rtuAddrInData.equals(rtuAddrAtHead)){
                  //数据头中的RTU地址与数据中的RTU地址不一致，更换成数据中的RTU地址
                  TcpSessionCach.changeRtuAddr(rtuAddrAtHead, rtuAddrInData, protocolName, session);
                  TcpSessionCache.changeRtuAddr(rtuAddrAtHead, rtuAddrInData, protocolName, session);
                  session.setAttribute(TcpIoSessionAttrIdIsRtuAddr.sessionArrIdKey, rtuAddrInData) ;
               }
					
               String rtuAddr = null ;

               String rtuAddr ;
               if(rtuAddrInData != null){
                  rtuAddr = rtuAddrInData  ; 
                  rtuAddr = rtuAddrInData  ;
               }else{
                  rtuAddr = rtuAddrAtHead ;
               }
					

               InetSocketAddress sa = (InetSocketAddress)session.getRemoteAddress() ;
               if(isOnLine){
                  //上线了
                  RtuStatusDealer.onLine(rtuAddr, sa.getAddress().getHostAddress(), sa.getPort());
               }
               if(reportOrResponse_trueOrFalse != null && reportOrResponse_trueOrFalse.booleanValue()){
               if(reportOrResponse_trueOrFalse != null && reportOrResponse_trueOrFalse){
                  RtuStatusDealer.upReport(rtuAddr, upBuf.length) ;
               }else{
                  RtuStatusDealer.upData(rtuAddr, upBuf.length) ;
               }
					

               //记录日志
               if(parseFail){
                  RtuLogDealer.log(rtuAddr, (isOnLine?"上线数据 ":"上行数据 ") + "" + code + "：" + upHex + "（解析失败）");
@@ -182,8 +184,8 @@
               }
            }
         }) ;
         if(midRses != null){
            for(MidResult rs : midRses){
         if(midRs != null){
            for(MidResult rs : midRs){
               rs.action();
            }
         }