package com.gyee.wisdom.dataadapter.websocket;

import com.gyee.wisdom.common.data.timeseries.TsData;
import com.gyee.wisdom.dao.timeseries.IDataChangeDao;
import com.gyee.wisdom.dataadapter.help.Constant;
import com.gyee.wisdom.dataadapter.websocket.backwork.PhotovoltaicInverterServer;
import com.gyee.wisdom.dataadapter.websocket.backwork.WindturbineInfoCache;
import com.gyee.wisdom.dataadapter.websocket.domain.MyConstant;
import com.gyee.wisdom.dataadapter.websocket.session.StompSessionService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;

/**
 * 数据发布服务
 * 1、周期性扫描实时库，监听topic关联数据的变化
 * 2、将变化的数据发布到对应的topic上
 */
@Slf4j
@Component
public class PublishServer {

    private int rtdbScanInterval = 100;  // 实时库论询时间间隔(ms)
    private int publishInterval = 500;   // 数据发布周期(ms)
    private boolean serverStarted = false;
    private boolean scanThreadFlag = false;
    private boolean publishThreadFlag = false;

    private final SimpMessagingTemplate messagingTemplate;

    private ConcurrentLinkedQueue<TopicMessage> dataQueue = new ConcurrentLinkedQueue<>();

    @Autowired
    private StompSessionService stompSessionService;

    @Autowired
    private IDataChangeDao dataChangeDao;

    @Autowired
    public PublishServer(SimpMessagingTemplate messagingTemplate) {
        this.messagingTemplate = messagingTemplate;
    }

    @Autowired
    private WebSocketTsDataController webSocketTsDataController;

    @Autowired
    private WindturbineInfoCache windturbineInfoCache;

    @Autowired
    private PhotovoltaicInverterServer PVServer;

    public void init(int scanInterval, int pubInterval) {
        rtdbScanInterval = scanInterval;
        publishInterval = pubInterval;
        Constant.d("===========[发布服务  init]===========");
    }

    public boolean start()
    {
        if (serverStarted)
        {
            return true;
        }

        Constant.d("===========[发布服务  Start]===========");
        try
        {
           /* scanThreadFlag = true;
            getScanRtdbThread().start();
            publishThreadFlag = true;
            getPublishThread().start();*/
            WindturbineInfoThread().start();
            InverterDataThread().start();

        }
        catch (Exception ex)
        {
            log.error(ex.getMessage());
        }
        return true;
    }

    public void stop() {
        scanThreadFlag = false;
        publishThreadFlag = false;
        PVServer.allowUpdatePVData=false;
        try {
            Thread.sleep(1000);
        } catch (Exception ex) {
        }
    }

    Thread getScanRtdbThread() {
        return new Thread(new Runnable() {
            public void run() {
                while (scanThreadFlag) {
                    try {
                        //log.info("rtdb scan task executing......");
                        Set<String> topics = stompSessionService.getSessionTopics();
                        for (String topic : topics) {
                            Map<String, TsData> dataMap = dataChangeDao.getTopicDataChange(topic);
                            if (dataQueue.offer(new TopicMessage(topic, dataMap)) == false) {
                                log.error("dataQueue.offer failed!");
                            }
                        }
                    } catch (Exception ex) {
                        log.error(ex.getMessage());
                    }
                    sleep(rtdbScanInterval);
                }
            }
        });
    }

    Thread getPublishThread() {
        return new Thread(new Runnable() {
            public void run() {
                while (publishThreadFlag) {
                    try {
                        //log.info("publish task executing......");
                        int cnt = dataQueue.size() < 10 ? dataQueue.size() : 10;
                        while (cnt-- > 0) {
                            TopicMessage topicData = dataQueue.poll();
                            if (topicData != null && topicData.getTsDataMap() != null && topicData.getTsDataMap().size() > 0) {
                                messagingTemplate.convertAndSend(topicData.getTopic(), topicData.getTsDataMap());
                                log.info("public message: topic=" + topicData.getTopic());
                            }
                        }
                    } catch (Exception ex) {
                        log.info(ex.getMessage());
                    }
                    sleep(publishInterval);
                }
            }
        });
    }

    //风机基本信息推荐线程
    Thread WindturbineInfoThread() {
        return new Thread(new Runnable() {
            @Override
            public void run() {
                Constant.d("-------    [风机 光伏] 数据推送开始     --------");
                while (true) {

                    try {
                        webSocketTsDataController.SendWindToAll(MyConstant.WINDTURBINE_INFO, windturbineInfoCache.getWindturbineDataMap());
                        webSocketTsDataController.SendPVToAll(MyConstant.Photovoltaic_INFO,PVServer.getInverterMap());


                        Thread .sleep(1500);
                    } catch (Exception ex) {
                        log.info(ex.getMessage());
                    }
                }
            }
        });
    }

    Thread InverterDataThread()
    {
        return new Thread(PVServer);
    }




    private void sleep(int milliseconds) {
        try {
            TimeUnit.MILLISECONDS.sleep(milliseconds);
        } catch (Exception ex) {
            log.info(ex.getMessage());
        }
    }

}