package com.base.orbit.thread;

import com.base.orbit.constant.EventBusConstants;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
public class ThreadPool {

    /*
     * 事件总线kafkaConsumer执行线程池
     */
    public static ThreadPoolTaskExecutor EVENT_BUS_KAFKA_CONSUMER_EXECUTOR;
    public static void createEventBusKafkaConsumerExecutor(int iotQty, int bizQty) {
        int totalQty = iotQty + bizQty;
        ThreadPool.EVENT_BUS_KAFKA_CONSUMER_EXECUTOR = ThreadPoolTaskExecutorBuilder.create()
                .executorName("事件总线kafkaConsumer执行线程池")
                // iot + biz
                .corePoolSize(totalQty)
                .maxPoolSize(totalQty)
                .queueCapacity(0)
                .keepAliveSeconds(60 * 60 * 24)
                .threadFactory(new ThreadFactoryBuilder()
                        .setNameFormat(EventBusConstants.KAFKA_CONSUMER_THREAD_NAME_PATTERN)
                        .setDaemon(false)
                        .setUncaughtExceptionHandler((t, e) -> log.error("Thread {} error!", t.getName(), e))
                        .build())
                .rejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy())
                .build();
    }

    /*
     * 事件总线持久化消费记录的线程池
     */
    public static ThreadPoolTaskExecutor EVENT_BUS_CONSUME_HISTORY_SAVER_EXECUTOR;
    // for unit test purpose
    public static void createEventBusConsumeHistorySaverExecutor() {
        ThreadPool.EVENT_BUS_CONSUME_HISTORY_SAVER_EXECUTOR = ThreadPoolTaskExecutorBuilder.create()
                .executorName("事件总线持久化消费记录的线程池")
                .corePoolSize(1)
                .maxPoolSize(1)
                .queueCapacity(0)
                .keepAliveSeconds(Integer.MAX_VALUE)
                .threadFactory(new ThreadFactoryBuilder()
                        .setNameFormat(EventBusConstants.EVENT_BUS_CONSUME_HISTORY_SAVER_THREAD_NAME_PATTERN)
                        .setDaemon(false)
                        .setUncaughtExceptionHandler((t, e) -> log.error("Thread {} error!", t.getName(), e))
                        .build())
                .rejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy())
                .build();
    }
    static {
        ThreadPool.createEventBusConsumeHistorySaverExecutor();
    }

    /**
     * 系统全局Timer管理
     */
    public static final ScheduledExecutorService timer = new ScheduledThreadPoolExecutor(2);

}
