package com.elab.spring.utils;

import com.elab.core.async.RealTaskExecutor;
import com.elab.core.async.SchedulingTaskExecutor;
import com.elab.core.async.model.TaskStoreData;
import com.elab.core.async.pruducer.ITaskProducer;
import com.elab.log.ext.CatSupplier;
import com.elab.spring.config.prop.ThreadPoolProperties;
import com.elab.spring.config.prop.ThreadProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;
import java.util.function.Supplier;

/**
 * @Module 工具类
 * @Description 线程执行工具类
 * @Author liukaixiong
 * @Date 2020/11/18 15:51
 */
public class ThreadProcessUtils implements ApplicationContextAware, InitializingBean {
    private static Logger logger = LoggerFactory.getLogger(ThreadProcessUtils.class);
    private ApplicationContext applicationContext;

    private static ITaskProducer taskProducer;

    private ThreadProperties threadProperties;

    private static ExecutorService executorService;

    private Long defaultTimeOut = 10000L;

    static ExecutorService getExecutorService() {
        return executorService;
    }

    /**
     * 设置线程池
     *
     * @param executorService
     */
    static void setExecutorService(ExecutorService executorService) {
        ThreadProcessUtils.executorService = executorService;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    public static boolean addRealTimeQueue(RealTaskExecutor taskExecutor, TaskStoreData taskStoreData) {
        if (taskProducer != null) {
            return taskProducer.sendRealTimeQueue(taskExecutor, taskStoreData);
        }
        logger.warn("加入队列失败，taskProducer 为空！");
        return false;
    }

    /**
     * 支持异步并串联日志
     *
     * @param supplier 提交任务
     * @param <U>
     * @return
     */
    public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
        return CompletableFuture.supplyAsync(new CatSupplier<>(supplier), getExecutorService());
    }

    /**
     * 并行任务执行
     *
     * @param supplier 任务列表
     * @param <U>
     * @return
     * @throws Exception
     */
    public static <U> CompletableFuture<U>[] supplyAsync(Supplier<U>... supplier) throws Exception {
        return supplyAsync(true, null, supplier);
    }

    /**
     * 多任务并行处理，最终按照提交顺序返回
     *
     * @param isBlockResult 是否阻塞返回结果
     * @param supplier      提交任务
     * @param <U>
     * @return 阻塞
     * @throws Exception
     */
    public static <U> CompletableFuture<U>[] supplyAsync(boolean isBlockResult, Long timeOutMs, Supplier<U>... supplier)
        throws Exception {
        CompletableFuture<U>[] result = new CompletableFuture[supplier.length];
        for (int i = 0; i < supplier.length; i++) {
            CompletableFuture<U> completableFuture = supplyAsync(supplier[i]);
            result[i] = completableFuture;
        }

        allOf(isBlockResult, timeOutMs, result);

        return result;
    }

    public static <U> Map<String, CompletableFuture<U>> supplyAsync(Map<String, Supplier<U>> taskGroupMap)
        throws Exception {
        return supplyAsync(true, taskGroupMap);
    }

    /**
     * 多任务并行处理,并按照key分组后返回结AuthMobileDataHandleConsumer果
     *
     * @param isBlockResult 是否阻塞返回结果集
     * @param taskGroupMap  任务分组map
     * @param <U>
     * @return
     * @throws Exception
     */
    public static <U> Map<String, CompletableFuture<U>> supplyAsync(boolean isBlockResult,
        Map<String, Supplier<U>> taskGroupMap) throws Exception {
        return supplyAsync(isBlockResult, taskGroupMap, null);
    }

    /**
     * 异步调用
     *
     * @param isBlockResult 是否阻塞直到结果获取完成
     * @param taskGroupMap  任务组
     * @param timeOutMs     等待获取时间
     * @param <U>
     * @return
     * @throws Exception
     */
    public static <U> Map<String, CompletableFuture<U>> supplyAsync(boolean isBlockResult,
        Map<String, Supplier<U>> taskGroupMap, Long timeOutMs) throws Exception {
        CompletableFuture<U>[] result = new CompletableFuture[taskGroupMap.size()];

        Map<String, CompletableFuture<U>> resultMap = new HashMap<>();
        int i = 0;
        for (Map.Entry<String, Supplier<U>> entry : taskGroupMap.entrySet()) {
            CompletableFuture<U> completableFuture = supplyAsync(entry.getValue());
            result[i] = completableFuture;
            resultMap.put(entry.getKey(), completableFuture);
            i++;
        }

        allOf(isBlockResult, timeOutMs, result);

        return resultMap;
    }

    private static <U> void allOf(boolean isBlockResult, Long timeOutMs, CompletableFuture<U>[] result)
        throws InterruptedException, ExecutionException, TimeoutException {
        if (isBlockResult) {
            CompletableFuture<Void> completableFuture = CompletableFuture.allOf(result);
            if (timeOutMs != null) {
                completableFuture.get(timeOutMs, TimeUnit.MILLISECONDS);
            } else {
                completableFuture.get();
            }
        }
    }

    public static void addSchedulingList(SchedulingTaskExecutor schedulingTaskExecutor) {
        if (taskProducer != null) {
            taskProducer.sendSchedulingList(schedulingTaskExecutor);
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        taskProducer = this.applicationContext.getBean(ITaskProducer.class);
        this.threadProperties = this.applicationContext.getBean(ThreadProperties.class);
        if (this.threadProperties != null && this.threadProperties.getThreadPool() != null) {
            ThreadPoolProperties threadPool = this.threadProperties.getThreadPool();
            ExecutorService executorService =
                new ThreadPoolExecutor(threadPool.getCorePoolSize(), threadPool.getMaximumPoolSize(),
                    threadPool.getKeepAliveTime(), TimeUnit.SECONDS, new ArrayBlockingQueue(threadPool.getQueueSize()));
            setExecutorService(executorService);
        } else {
            ThreadPoolProperties threadPool = new ThreadPoolProperties();
            ExecutorService executorService =
                new ThreadPoolExecutor(threadPool.getCorePoolSize(), threadPool.getMaximumPoolSize(),
                    threadPool.getKeepAliveTime(), TimeUnit.SECONDS, new ArrayBlockingQueue(threadPool.getQueueSize()));
            setExecutorService(executorService);
        }
    }
}
