package com.dy.common.mw.protocol; import java.util.Collection; import java.util.HashMap; import com.dy.common.mw.channel.tcp.TcpIoSessionAttrIdIsRtuAddr; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.apache.mina.core.buffer.IoBuffer; import com.dy.common.util.ByteUtil; import com.dy.common.mw.channel.tcp.PrefixedDataAvailableHandle; import com.dy.common.mw.channel.tcp.PrefixedDataAvailableStatus; import org.apache.mina.core.session.IoSession; public class PrefixedDataAvailableHandleImp implements PrefixedDataAvailableHandle { private static final Logger log = LogManager.getLogger(PrefixedDataAvailableHandleImp.class) ; /** * 在多线程环境中运行 * 分析上线数据(网络连接后第一包数据)是否可获得 * @param in IoBuffer * @return PrefixedDataAvailableStatus */ public PrefixedDataAvailableStatus forOnLine(IoSession ioSession, IoBuffer in){ int remain = in.remaining() ; if(remain == 0){ return new PrefixedDataAvailableStatus().breaked() ; }else{ try{ PrefixedDataAvailableStatus pds = null ; PrefixedDataAvailable pda = null ; //从会话缓存得到上次上行数据确定的协议 String protocolName = (String) ioSession.getAttribute(TcpIoSessionAttrIdIsRtuAddr.sessionArrProtocolName) ; Short protocolVersion = (Short) ioSession.getAttribute(TcpIoSessionAttrIdIsRtuAddr.sessionArrProtocolVersion) ; //由会话中的协议得到处理类 //利用缓存的协议，进而对不符合协议的上行数据也能用该协议处理了，例如RTU采用不同数据协议的嵌入式软件升级协议 //即RTU第一包数据确定下来协议，后面数据都用这个协议处理，那么数据可以变化了，可以不符合第一包数据遵守的协议了 Object[] objs = this.getClassObjAndAnnotationVo(protocolName, protocolVersion) ; if(objs != null && objs[0] != null && objs[1] != null){ pda = (PrefixedDataAvailable)objs[0] ; //处理完整性检查 pds = pda.forOnLine(ioSession, in, remain, ((AnnotationPrefixedDataAvailableVo)objs[1]).onLineDataMinLength, ((AnnotationPrefixedDataAvailableVo)objs[1]).errorMaxLength) ; //用完对象后，放回池中 PrefixedDataAvailablePool.freeInstance(objs[0].getClass(), pda); } if(pds == null){ //如果会话中缓存的协议未能正确处理 HashMap prefixedDataAvailableMap = ProtocolCache.getPrefixedDataAvailableMap() ; Collection set = prefixedDataAvailableMap.values() ; if(set.size() == 0){ throw new Exception("上线数据完整性检查时，得到的协议完整性检查类集合为空。") ; } int priority = ProtocolConstant.firstPriority ; while(true){ objs = this.getClassObjAndAnnotationVo(priority, set) ; pda = (PrefixedDataAvailable)objs[0] ; if(pda == null && priority == ProtocolConstant.firstPriority){ throw new Exception("上线数据完整性检查时，未得到优先级为" + priority + "上线数据完整性检查类！") ; }else if(pda == null){ //说明上线数据完整性检查类集合已经遍历完了。 break ; } //处理完整性检查 pds = pda.forOnLine(ioSession, in, remain, ((AnnotationPrefixedDataAvailableVo)objs[1]).onLineDataMinLength, ((AnnotationPrefixedDataAvailableVo)objs[1]).errorMaxLength) ; //用完对象后，放回池中 //2024-11-25下面一行进行了修改 //PrefixedDataAvailablePool.freeInstance(((AnnotationPrefixedDataAvailableVo)objs[1]).clazz, pda); PrefixedDataAvailablePool.freeInstance(objs[0].getClass(), pda); if(pds == null){ //说明不是对应的协议数据，需要另外的协议来处理上线 //循环继续 priority++ ; }else{ //停止循环，返回结果 break ; } } } if(pds == null){ //说明数据不属于任何协议，一般为Rtu数据出错，或网络攻击数据 byte[] preByte = new byte[remain]; in.get(preByte) ; in.position(0) ; log.error("上线第一包数据未找到对应的解析协议。数据是：" + ByteUtil.bytes2Hex(preByte, true)) ; //认为是完整包数据，后续读出，防止占用内存及死循环IO读数据 pds = new PrefixedDataAvailableStatus().rubbish(remain) ; } return pds ; }catch(Exception e){ //处理过程中发生异常，上行数据就是垃圾数据了，后续以垃圾数据处理，即把数据从IO中取出来，以防止死循环IO读数据 return new PrefixedDataAvailableStatus().rubbish(remain) ; } } } /** * 在多线程环境中运行 * 分析上行数据(网络连接后第二(包含)包以后数据)是否可获得 * @param in IoBuffer * @return PrefixedDataAvailableStatus */ public PrefixedDataAvailableStatus forUpData(IoSession ioSession, IoBuffer in){ int remain = in.remaining() ; if(remain == 0){ return new PrefixedDataAvailableStatus().breaked() ; }else{ try{ PrefixedDataAvailableStatus pds = null ; PrefixedDataAvailable pda = null ; //从会话缓存得到上次上行数据确定的协议 String protocolName = (String) ioSession.getAttribute(TcpIoSessionAttrIdIsRtuAddr.sessionArrProtocolName) ; Short protocolVersion = (Short) ioSession.getAttribute(TcpIoSessionAttrIdIsRtuAddr.sessionArrProtocolVersion) ; //由会话中的协议得到处理类 Object[] objs = this.getClassObjAndAnnotationVo(protocolName, protocolVersion) ; if(objs != null && objs[0] != null && objs[1] != null){ pda = (PrefixedDataAvailable)objs[0] ; //处理完整性检查 pds = pda.forOnLine(ioSession, in, remain, ((AnnotationPrefixedDataAvailableVo)objs[1]).onLineDataMinLength, ((AnnotationPrefixedDataAvailableVo)objs[1]).errorMaxLength) ; //用完对象后，放回池中 PrefixedDataAvailablePool.freeInstance(objs[0].getClass(), pda); } if(pds == null) { //如果会话中缓存的协议未能正确处理 HashMap prefixedDataAvailableMap = ProtocolCache.getPrefixedDataAvailableMap() ; Collection set = prefixedDataAvailableMap.values() ; if(set.size() == 0){ throw new Exception("上线数据完整性检查时，得到的协议集合为空。") ; } int priority = ProtocolConstant.firstPriority ; while(true){ objs = this.getClassObjAndAnnotationVo(priority, set) ; pda = (PrefixedDataAvailable)objs[0] ; if(pda == null && priority == ProtocolConstant.firstPriority){ throw new Exception("上线数据完整性检查时，未得到优先级为" + priority + "上线数据完整性检查类！") ; }else if(pda == null){ //说明上线数据完整性检查类集合已经遍历完了。 break ; } //处理完整性检查 pds = pda.forUpData(ioSession, in, remain, ((AnnotationPrefixedDataAvailableVo)objs[1]).headMinLength, ((AnnotationPrefixedDataAvailableVo)objs[1]).errorMaxLength) ; //用完对象后，放回池中 PrefixedDataAvailablePool.freeInstance(objs[0].getClass(), pda); if(pds == null){ //说明不是对应的协议数据，需要另外的协议来处理上线 //循环继续 priority++ ; }else{ //停止循环，返回结果 break ; } } } if(pds == null){ //说明数据不属于任何协议，一般为Rtu数据出错，或网络攻击数据 byte[] preByte = new byte[remain]; in.get(preByte) ; in.position(0) ; log.error("上行数据未找到对应的解析协议。数据是：" + ByteUtil.bytes2Hex(preByte, true)) ; //认为是完整包数据，后续读出，防止占用内存及死循环IO读数据 pds = new PrefixedDataAvailableStatus().rubbish(remain) ; } return pds ; }catch(Exception e){ //处理过程中发生异常，上行数据就是垃圾数据了，后续以垃圾数据处理，即把数据从IO中取出来，以防止死循环IO读数据 return new PrefixedDataAvailableStatus().rubbish(remain) ; } } } /** * 得到处理类对象 * @param prority 优先级 * @param set 集合 * @return Object[] * @throws Exception 异常 */ private Object[] getClassObjAndAnnotationVo(int prority, Collection set) throws Exception{ PrefixedDataAvailable obj = null ; AnnotationPrefixedDataAvailableVo rVo = null ; for(AnnotationPrefixedDataAvailableVo vo : set){ if(prority == vo.priority){ obj = PrefixedDataAvailablePool.getInstance(vo.clazz) ; rVo = vo ; break ; } } return new Object[]{obj, rVo} ; } /** * 得到处理类对象 * @param protocolName 通信协议名称 * @param protocolVersion 通信协议版本号 * @return Object[] * @throws Exception 异常 */ private Object[] getClassObjAndAnnotationVo(String protocolName, Short protocolVersion) throws Exception{ PrefixedDataAvailable obj = null ; AnnotationPrefixedDataAvailableVo rVo = null ; if(protocolName != null && !protocolName.trim().equals("") && protocolVersion != null){ rVo = ProtocolCache.getAnnotationPrefixedDataAvailable(protocolName, protocolVersion) ; obj = PrefixedDataAvailablePool.getInstance(rVo.clazz) ; } return new Object[]{obj, rVo} ; } }