package com.dy.rtuMw.server.tasks; import com.dy.common.mw.core.CoreTask; import com.dy.common.queue.Node; import com.dy.rtuMw.server.mqtt.MqttComResultCache; import com.dy.rtuMw.server.mqtt.MqttComResultNode; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; /** * 对RTU上行数据进行业务处理 */ public class MqttComResultConstantTask extends CoreTask { private static final Logger log = LogManager.getLogger(MqttComResultConstantTask.class.getName()); /** * 在单线程环境中运行 */ @Override public Integer execute() { try{ dealRtuComResult() ; }catch(Exception e){ log.error(e); } return MqttComResultCache.size()>0?0:1 ; } /** * 处理上行命令结果 */ public void dealRtuComResult() { Node first = MqttComResultCache.getFirstQueueNode() ; if(first != null){ Node last = MqttComResultCache.getLastQueueNode() ; while (last != null){ last = this.doDealRtuComResult(first, last); } } } /** * 处理缓存的上行数据节点 * @param first 第一个节点 * @param last 最后一个节点 */ private Node doDealRtuComResult(Node first, Node last){ if(last != null){ //在dealNode方法中,可能要把last从队列中移除,这时last.pre为空,所以提前把last.pre取出来 Node pre = last.pre ; dealNode(last) ; if(first != last){ return pre ; }else{ //停止 return null ; } }else{ return null ; } } /** * 处理一个节点 * @param node 节点 */ private void dealNode(Node node){ if(node != null && node.obj != null){ MqttComResultNode obj = (MqttComResultNode)node.obj ; obj.dealSelf() ; MqttComResultCache.removeNode(node); } } }