/**
 * Copyright
 * All right reserved.
 * 项目名称:内蒙运维系统-数据ETL服务
 * 创建日期:2022/5/23
 */
package org.springblade.etl;


import lombok.extern.slf4j.Slf4j;
import org.glassfish.jersey.internal.guava.ThreadFactoryBuilder;
import org.springblade.core.tool.utils.Func;
import org.springblade.enums.EtlModeEnum;
import org.springblade.etl.service.*;
import org.springblade.modules.data.service.IWarnDataCacheService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.*;

/***
 * Date:2022/5/22
 * Title:  ETL模块
 * Description： 定时任务管理
 * @author swp
 * @version 1.0
 * Remark：认为有必要的其他信息
 */
@Slf4j
@Component
@EnableScheduling
public class EtlTaskManager {

    /**
     * 公共线程池
     **/
    private static ThreadFactory publicThreadFactory = new ThreadFactoryBuilder().setNameFormat("etl-thread-pool-%d").build();
    private static ExecutorService publicThreadPool = new ThreadPoolExecutor(1, 4,
            0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(1024), publicThreadFactory, new ThreadPoolExecutor.AbortPolicy());


    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private RedisTemplate redisTemplate;

    @Resource
    private IEtlRainDataService etlRainDataService;

    @Resource
    private IEtlRiverDataService etlRiverDataService;

    @Resource
    private IEtlRsvrDataService etlRsvrDataService;

    @Resource
    private IEtlGroundDataService etlGroundDataService;

    @Resource
    private IEtlMvAttStBaseService etlMvAttStBaseService;

    @Resource
    private IEtlWarningService etlWarningService;

    @Resource
    private IWarnDataCacheService warnDataService;

    @Value("${spring.mq-topic.topic-etl-rain}")
    private String topicRain;

    @Value("${spring.mq-topic.topic-etl-river}")
    private String topicRiver;

    @Value("${spring.mq-topic.topic-etl-rsvr}")
    private String topicRsvr;

    @Value("${spring.mq-topic.topic-etl-ground}")
    private String topicGround;

    @Value("${spring.mq-topic.topic-etl-warn}")
    private String topicWarn;

    public void submitRainDataEtlTask(int etlMode,String taskId) {
        EtlRainDataTask rainDataTask = new EtlRainDataTask(etlMvAttStBaseService, etlRainDataService, redisTemplate, kafkaTemplate, topicRain, etlMode,taskId);
        FutureTask<Integer> futureTask = new FutureTask<>(rainDataTask);
        publicThreadPool.execute(futureTask);
    }

    public void submitRiverDataEtlTask(int etlMode,String taskId) {
        EtlRiverDataTask riverDataTask = new EtlRiverDataTask(etlMvAttStBaseService, etlRiverDataService, redisTemplate, kafkaTemplate, topicRiver, etlMode,taskId);
        FutureTask<Integer> futureTask = new FutureTask<>(riverDataTask);
        publicThreadPool.execute(futureTask);
    }

    public void submitRsvrDataEtlTask(int etlMode,String taskId) {
        EtlRsvrDataTask rsvrDataTask = new EtlRsvrDataTask(etlMvAttStBaseService, etlRsvrDataService, redisTemplate, kafkaTemplate, topicRsvr, etlMode,taskId);
        FutureTask<Integer> futureTask = new FutureTask<>(rsvrDataTask);
        publicThreadPool.execute(futureTask);
    }

    public void submitWarnDataEtlTask(int etlMode,String taskId) {
        EtlWarnDataTask warnDataTask = new EtlWarnDataTask(etlMvAttStBaseService, etlWarningService,warnDataService,redisTemplate, kafkaTemplate, topicWarn, etlMode,taskId);
        FutureTask<Integer> futureTask = new FutureTask<>(warnDataTask);
        publicThreadPool.execute(futureTask);
    }

    public void submitGroundDataEtlTask(int etlMode,String taskId) {
        EtlGroundDataTask groundDataTask = new EtlGroundDataTask(etlMvAttStBaseService, etlGroundDataService, redisTemplate, kafkaTemplate, topicGround, etlMode,taskId);
        FutureTask<Integer> futureTask = new FutureTask<>(groundDataTask);
        publicThreadPool.execute(futureTask);
    }

    public void submitRtuDataPromptlyCollectTask(String rtuCode) {
        log.info("同步测站 {}", rtuCode);
    }

    @Scheduled(cron = "0 0/5 * * * * ")
    public void etlRainDataTask() {
        try {
            String id = Func.random(100);
            log.info("ETL-雨情实时数据同步任务开始执行,{}",id);
            this.submitRainDataEtlTask(EtlModeEnum.ETL_ONCE.getCode(),id);
        } catch (Exception e) {
            log.error("{}", e.getMessage());
        }
    }

    @Scheduled(cron = "10 0/5 * * * * ")
    public void etlRiverDataTask() {
        try {
            String id = Func.random(100);
            log.info("ETL-河道水情实时数据同步任务开始执行,{}",id);
            this.submitRiverDataEtlTask(EtlModeEnum.ETL_ONCE.getCode(),id);
        } catch (Exception e) {
            log.error("{}", e.getMessage());
        }
    }

    @Scheduled(cron = "20 0/5 * * * * ")
    public void etlRsvrDataTask() {
        try {
            String id = Func.random(100);
            log.info("ETL-水库水情实时数据同步任务开始执行,{}",id);
            this.submitRsvrDataEtlTask(EtlModeEnum.ETL_ONCE.getCode(),id);
        } catch (Exception e) {
            log.error("{}", e.getMessage());
        }
    }

    @Scheduled(cron = "30 0/5 * * * * ")
    public void etlWarnDataTask() {
        try {
            String id = Func.random(100);
            log.info("ETL-预警实时数据同步任务开始执行,{}",id);
            this.submitWarnDataEtlTask(EtlModeEnum.ETL_ONCE.getCode(),id);
        } catch (Exception e) {
            log.error("{}", e.getMessage());
        }
    }

    @Scheduled(cron = "40 0/10 * * * * ")
    public void etlGroundDataTask() {
        try {
            String id = Func.random(100);
            log.info("ETL-墒情实时数据同步任务开始执行,{}",id);
            this.submitGroundDataEtlTask(EtlModeEnum.ETL_ONCE.getCode(),id);
        } catch (Exception e) {
            log.error("{}", e.getMessage());
        }
    }

}
