package com.jay.monitor.data.server.store;

import com.cat.file.message.MessageManagerProcess;
import com.cat.file.message.internal.DefaultMessageTree;
import com.jay.monitor.data.core.enums.MsgTypeEnums;
import com.jay.monitor.data.core.model.serializable.base.BaseDTO;
import com.jay.monitor.data.core.model.serializable.base.ErrorInfoData;
import com.jay.monitor.data.core.model.serializable.base.JsonTextData;
import com.jay.monitor.data.server.config.props.MonitorProperties;
import com.jay.monitor.data.server.costs.ServerConstants;
import com.jay.monitor.data.server.enums.DataSplitType;
import com.jay.monitor.data.server.enums.StoreType;
import com.jay.monitor.data.server.factory.thread.TaskRunnable;
import com.jay.monitor.data.server.report.RealtimeAlertProcess;
import com.jay.monitor.data.server.store.mysql.MysqlDDLProcess;
import com.jay.monitor.data.server.utils.DataUtils;
import com.jay.monitor.data.server.utils.MonitorPropertiesUtil;
import org.apache.commons.beanutils.BeanUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;

/**
 * @Module 执行器
 * @Description 抽象存储
 * @Author liukaixiong
 * @Date 2020/10/29 13:33
 */
public abstract class AbstractStoreProcess<T extends BaseDTO, E> implements TaskRunnable, InitializingBean {
    private Logger logger = LoggerFactory.getLogger(getClass());

    private LongAdder counter = new LongAdder();

    private int queueSize = 5000;

    private int batchSize = 500;

    private volatile boolean isConsumer = true;

    private BlockingQueue<T> queue = new ArrayBlockingQueue<>(queueSize);

    @Autowired
    private MysqlDDLProcess ddlProcess;

    @Autowired
    private MonitorProperties monitorProperties;

    @Autowired
    private RealtimeAlertProcess realtimeAlertProcess;

    @Autowired
    protected MessageManagerProcess messageManagerProcess;

    public void setQueueSize(int queueSize) {
        this.queueSize = queueSize;
    }

    public int getQueueSize() {
        return queueSize;
    }

    public int getBatchSize() {
        return batchSize;
    }

    public void setBatchSize(int batchSize) {
        this.batchSize = batchSize;
    }

    /**
     * 存储类型
     *
     * @return
     */
    public abstract StoreType storeType();

    /**
     * 消息类型
     *
     * @return
     */
    public abstract MsgTypeEnums msgTypeEnums();

    /**
     * 根据信息编号查询异常内容信息
     *
     * @param date
     * @param id
     * @return
     * @throws Exception
     */
    public abstract String searchErrorById(String date, String id) throws Exception;

    /**
     * 根据信息编号查询请求内容信息
     *
     * @param date
     * @param id
     * @return
     * @throws Exception
     */
    public abstract String searchContentById(String date, String id) throws Exception;

    /**
     * 查询分页数据
     *
     * @param date
     * @param pageSize
     * @param entity
     * @return
     * @throws Exception
     */
    public abstract List<E> searchDataByList(String date, Integer pageSize, E entity) throws Exception;

    /**
     * 加速消费
     *
     * @param request
     */
    public abstract List<E> speedUpConsumerList(List<T> request);

    public Object store(T request) throws Exception {

        if (!this.isConsumer) {
            logger.debug("当前消费标记:" + this.isConsumer + " 不处理消息!");
            return null;
        }

        storeBefore(request);

        boolean add = getQueue().offer(request);

        if (!add) {
            counter.decrement();
            if (counter.intValue() == 0 || counter.intValue() > queueSize) {
                logger.warn(getClass() + " 队列消费满了，装不下去了");
                counter.reset();
            }
        }

        return add;
    }

    /**
     * 存储之前触发
     *
     * @param request
     */
    protected void storeBefore(T request) {

    }

    /**
     * 存储之后触发
     *
     * @param resultList
     * @param success
     */
    protected void storeAfter(List<T> requestList, List<E> resultList, boolean success) throws Exception {
        // 记录一些大文本数据
        for (int i = 0; i < resultList.size(); i++) {
            T request = requestList.get(i);
            E result = resultList.get(i);
            String id = BeanUtils.getProperty(result, "id");
            if (request instanceof JsonTextData) {
                DefaultMessageTree messageTree = DataUtils.getJsonContentMessageTree(request, Integer.valueOf(id));
                if (messageTree != null) {
                    request.getExtMap().put(ServerConstants.JSON_LOG_ID, messageTree.getFormatMessageId());
                    this.messageManagerProcess.insert(messageTree);
                }
            }

            if (request instanceof ErrorInfoData) {

                DefaultMessageTree messageTree = DataUtils.getErrorMessageTree(request, Integer.valueOf(id));
                if (messageTree != null) {
                    request.getExtMap().put(ServerConstants.ERROR_LOG_ID, messageTree.getFormatMessageId());
                    this.messageManagerProcess.insert(messageTree);
                }
            }
        }

        // 告警触发
        realtimeAlertProcess.process(requestList);
    }


    public void speedUpConsumer(BlockingQueue<T> queue, int rate) throws Exception {
        List<T> requestList = new ArrayList<>();

        for (int i = 0; i < rate; i++) {
            T request = queue.poll();
            requestList.add(request);
        }

        List<E> resultList = speedUpConsumerList(requestList);

        storeAfter(requestList, resultList, true);
    }

    protected BlockingQueue<T> getQueue() {
        return queue;
    }

    @Override
    public void run() {
        logger.info("开启异步消费机...");
        int defaultSize = getQueueSize();
        int batchSize = getBatchSize();
        int minBatch = batchSize / 10;
        int thresholdValue = (int) (defaultSize * 0.8);
        long startHour = TimeUnit.MILLISECONDS.toHours(System.currentTimeMillis());
        while (isConsumer) {
            try {
                int size = this.queue.size();
                if (size > 0) {
                    if (size > thresholdValue) {
                        speedUpConsumer(this.queue, defaultSize / 2);
                    } else if (size > batchSize) {
                        speedUpConsumer(this.queue, batchSize);
                    } else if (size > minBatch) {
                        speedUpConsumer(this.queue, size);
                    } else {
                        // 如果数量不多的情况下
                        speedUpConsumer(this.queue, 1);
                    }
                } else {
                    long currentHour = TimeUnit.MILLISECONDS.toHours(System.currentTimeMillis());
                    if (currentHour > startHour) {
                        logger.warn(" current queue size : " + size + "\t hour : " + currentHour);
                        startHour = currentHour;
                    }
                }
            } catch (Throwable e) {
                logger.error("异步队列消费失败", e);
                sleep(5);
            } finally {
                sleep(5);
            }
        }
        stop();
        logger.warn(getThreadName() + " stop!!!!");
    }

    @Override
    public void stop() {
        this.isConsumer = false;
    }

    private void sleep(int ms) {
        try {
            Thread.sleep(ms);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void startConsumerThread() {
        String threadName = getThreadName();
        logger.info("开启消费线程:" + threadName);
        this.isConsumer = true;
        Thread thread = new Thread(this, threadName);
        thread.setUncaughtExceptionHandler((currentThread, throwable) -> {
            logger.error("极端异常出现未捕获的情况:" + currentThread.getName(), throwable);
        });
        thread.start();
    }

    @Override
    public String getThreadName() {
        String threadName = getClass().getSimpleName() + "-consumer-thread";
        return threadName;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        startConsumerThread();
    }

    /**
     * 获取分区编号
     *
     * @return
     */
    protected String getPartitionId() {
        // 从配置文件中获取对应的规则
        Map<String, DataSplitType> tableRuleMap = MonitorPropertiesUtil.getMysqlTableRuleMap(monitorProperties);

        if (tableRuleMap != null) {
            DataSplitType dataSplitType = tableRuleMap.get(tableName());
            if (dataSplitType != null) {
                return ddlProcess.getDataId(tableName(), dataSplitType);
            }
        }
        return null;
    }

    /**
     * 处理的表名
     *
     * @return
     */
    public abstract String tableName();
}
