package com.elab.core.async.consumer;

import com.elab.core.async.TaskProcessAdaptor;
import com.elab.core.async.TaskStoreProcess;
import com.elab.core.async.store.TaskExecutorDecoration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Queue;

/**
 * 实时任务消费
 *
 * @author ： liukx
 * @time ： 2020/8/6 - 10:57
 */
public class RealTimeConsumer implements Runnable {

    private Logger logger = LoggerFactory.getLogger(RealTimeConsumer.class);

    private Queue<TaskExecutorDecoration> taskQueue;

    private TaskStoreProcess taskStoreProcess = new TaskProcessAdaptor();

    public RealTimeConsumer(Queue<TaskExecutorDecoration> taskQueue) {
        this.taskQueue = taskQueue;
    }

    public void setTaskStoreProcess(TaskStoreProcess taskStoreProcess) {
        this.taskStoreProcess = taskStoreProcess;
    }

    @Override
    public void run() {
        while (true) {
            TaskExecutorDecoration executor = this.taskQueue.poll();
            try {
                if (executor != null) {
                    executor.getExecutor().run();
                    taskStoreProcess.success(executor.getTaskStoreData());
                }
            } catch (Exception e) {
                logger.error("实时消费异步线程处理异常", e);
                taskStoreProcess.error(executor.getTaskStoreData());
                //如果异常了，要记录一些参数
            } finally {
                try {
                    Thread.sleep(5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
