package com.fastbee.base.core.hanler;

import com.fastbee.common.core.protocol.Message;
import com.fastbee.base.session.Session;
import com.fastbee.base.util.VirtualList;
import com.fastbee.common.exception.ServiceException;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Method;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 异步批量处理报文
 * @author bill
 */
@Slf4j
public class AsyncBatchHandler extends BaseHandler{


   /*消息处理队列*/
   private final ConcurrentLinkedQueue<Message> queue;

   /*线程池*/
   private final ThreadPoolExecutor executor;

   private final int poolSize;

   private final int maxEventSize;

   private final int maxWait;

   private final int warningLines;


    public AsyncBatchHandler(Object target, Method targetMethod, String desc, int poolSize, int maxEventSize, int maxWait) {

        super(target, targetMethod, desc);
        Class<?>[] parameterTypes = targetMethod.getParameterTypes();
        if (parameterTypes.length >1){
            throw new ServiceException("参数列表过长");
        }
        if (!parameterTypes[0].isAssignableFrom(List.class)){
            throw new ServiceException("参数不是List类型");
        }

        this.poolSize = poolSize;
        this.maxEventSize = maxEventSize;
        this.maxWait = maxWait;
        this.warningLines = maxEventSize * poolSize * 50;

        this.queue = new ConcurrentLinkedQueue<>();
        this.executor = new ThreadPoolExecutor(this.poolSize,this.poolSize,1000L, TimeUnit.MILLISECONDS
                ,new LinkedBlockingQueue<>(500),new DefaultThreadFactory(targetMethod.getName()));

        for (int i = 0; i < poolSize; i++) {
            boolean start = i == 0;
            executor.execute(()->{
                try {
                    startInternal(start);
                }catch (Exception e){
                    log.error("线程池处理数据出错",e);
                }
            });
        }

    }

    @Override
    public <T extends Message> T invoke(T request, Session session) throws Exception {
        queue.offer(request);
        return null;
    }

    public void startInternal(boolean master) {
        Message[] array = new Message[maxEventSize];
        long logtime = 0;
        long starttime = 0;

        for (; ; ) {
            Message temp;
            int i = 0;
            while ((temp = queue.poll()) != null) {
                array[i++] = temp;
                if (i >= maxEventSize) {
                    break;
                }
            }

            if (i > 0) {
                starttime = System.currentTimeMillis();
                try {
                    targetMethod.invoke(targetObject, new VirtualList<>(array, i));
                } catch (Exception e) {
                    log.warn(targetMethod.getName(), e);
                }
                long time = System.currentTimeMillis() - starttime;
                if (time > 1000L) {
                    log.warn("线程池处理数据耗时:{}ms,共{}条记录", time, i);
                }
            }

            if (i < maxEventSize) {
                try {
                    for (int j = 0; j < i; j++) {
                        array[j] = null;
                    }
                    Thread.sleep(maxWait);
                } catch (InterruptedException e) {
                    log.error("sleep error!");
                }
            } else if (master) {
                if (logtime < starttime) {
                    logtime = starttime + 5000L;

                    int size = queue.size();
                    if (size > warningLines) {
                        log.warn("线程池队列已满, size:{}", size);
                    }
                }
            }
        }
    }
}