package com.elab.core.async.pruducer;

import com.elab.core.async.consumer.TaskConsumer;
import com.elab.core.async.store.TaskExecutorQueue;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

public class TaskProducerTest {
    private Logger logger = LoggerFactory.getLogger(getClass());

    private TaskProducer producer;

    @Before
    public void init() {
        TaskExecutorQueue queue = new TaskExecutorQueue(50);
        TaskConsumer taskConsumer = new TaskConsumer(queue);
        taskConsumer.startRealTimeConsumer();
        taskConsumer.startSchedulingConsumer(5);
        producer = new TaskProducer(queue);
    }

    @Test
    public void addRealTimeQueue() throws IOException {
        for (int i = 0; i < 10; i++) {
            int finalI = i;
            System.out.println("==业务执行开始==" + i);
            producer.sendRealTimeQueue(() -> {
                System.out.println("-------操作异步开始----------" + finalI);
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("-----异步操作结束-----------" + finalI);
            });
            System.out.println("==业务执行完成==" + i);
        }
        System.in.read();
    }

    @Test
    public void addSchedulingList() throws IOException {
        for (int i = 0; i < 10; i++) {
            int finalI = i;
            System.out.println("==业务执行开始==" + i);
            producer.sendSchedulingList(() -> {
                logger.info("-------操作异步开始----------" + finalI);
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                logger.info("-----异步操作结束-----------" + finalI);
            });
            System.out.println("==业务执行完成==" + i);
        }
        System.in.read();
    }

    @Test
    public void testReduce() throws Exception {
        List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);

        list.parallelStream().forEach((data) -> {
            System.out.println(Thread.currentThread().getName() + "\t" + data.toString());
        });

//        Map<String, Object> reduce = producer.reduce(list, (intList) -> {
//            System.out.println("====>>>>开始处理" + JSON.toJSONString(intList));
//            Map<String, Object> map = new HashMap<>();
//            List<Integer> aList = new ArrayList<>();
//            List<Integer> bList = new ArrayList<>();
//
//            for (int i = 0; i < intList.size(); i++) {
//                Integer index = list.get(i);
//                if (index % 2 == 0) {
//                    aList.add(index);
//                } else {
//                    bList.add(index);
//                }
//            }
//
//            map.put("A", aList);
//            map.put("B", bList);
//            return map;
//        });
//        System.out.println("map : " + JSON.toJSONString(reduce));
        System.in.read();

    }

}
