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.MqttPubMsgCache;
import com.dy.rtuMw.server.mqtt.MqttPubMsgNode;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * @Author: liurunyu
 * @Date: 2025/6/4 16:25
 * @Description
 */
public class MqttPubMessageConstantTask extends CoreTask {
    private static final Logger log = LogManager.getLogger(MqttPubMessageConstantTask.class.getName());

    /**
     * 在单线程环境中运行
     */
    @Override
    public Integer execute() {
        try{
            dealMqMsg() ;
        }catch(Exception e){
            log.error(e);
        }
        return MqttPubMsgCache.size()>0?0:1 ;
    }
    /**
     * 处理MQTT订阅的消息
     */
    public void dealMqMsg() {
        Node first = MqttPubMsgCache.getFirstQueueNode() ;
        if(first != null){
            Node last = MqttPubMsgCache.getLastQueueNode() ;
            while (last != null){
                last = this.doDealMqMsg(System.currentTimeMillis(), first, last);
            }
        }
    }

    /**
     * 处理缓存的上行数据节点
     * @param now 当前时刻
     * @param first 第一个节点
     * @param last 最后一个节点
     */
    private Node doDealMqMsg(Long now, Node first, Node last){
        if(last != null){
            //在dealNode方法中，可能要把last从队列中移除，这时last.pre为空，所以提前把last.pre取出来
            Node pre = last.pre ;
            dealNode(now, last) ;
            if(first != last){
                return pre ;
            }else{
                //停止
                return null ;
            }
        }else{
            return null ;
        }
    }


    /**
     * 处理一个节点
     * @param now 现在时刻
     * @param node 节点
     */
    private void dealNode(Long now, Node node){
        if(node != null && node.obj != null){
            MqttPubMsgNode obj = (MqttPubMsgNode)node.obj ;
            boolean removeNode = obj.dealSelf(now) ;
            if(removeNode){
                MqttPubMsgCache.removeNode(node);
            }
        }
    }

}