package com.twjitm.core.common.kafka;

import com.alibaba.fastjson.JSON;
import com.twjitm.core.common.config.global.GlobalConstants;
import com.twjitm.core.common.config.global.KafkaConfig;
import com.twjitm.core.common.service.IService;
import com.twjitm.core.spring.SpringServiceManager;
import com.twjitm.threads.thread.NettyThreadNameFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.util.concurrent.*;

/**
 * kafka 消息提供者
 *
 * @author twjitm - [Created on 2018-09-04 16:17]
 * @company https://github.com/twjitm/
 * @jdk java version "1.8.0_77"
 */
@Service
public class NettyKafkaProducerListener implements IService {
    private Logger logger = LoggerFactory.getLogger(NettyKafkaProducerListener.class);
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 消息处理线程池
     */
    private volatile ExecutorService executorService;

    private boolean run = true;
    /**
     * 消息队列
     */
    private BlockingQueue<AbstractKafkaPushTask> queue;

    /**
     * 将消息发送给kafka中心
     *
     * @param abstractKafkaPushTask
     */
    private void sendMessage(AbstractKafkaPushTask abstractKafkaPushTask) {
        String type = abstractKafkaPushTask.getTaskType().getTypeName();
        String value = JSON.toJSONString(abstractKafkaPushTask.getValue());
        //本身send方法就是一个异步执行方法
        ListenableFuture<SendResult<String, String>> result =
                kafkaTemplate.sendDefault(type, value);
        /**
         * 添加回调监听
         */
        result.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                logger.error("KAFKA MESSAGE SEND FAIL", throwable);
            }

            @Override
            public void onSuccess(SendResult<String, String> kafkaTaskTypeObjectSendResult) {
                logger.info("KAFKA MESSAGE SEND SUCCESS");
            }
        });


    }

    /**
     * 将任务存放到队列中
     *
     * @param task
     */
    public void put(AbstractKafkaPushTask task) {
        try {
            queue.put(task);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public String getId() {
        return NettyKafkaProducerListener.class.getSimpleName();
    }

    @Override
    public void startup() throws Exception {
        run = true;
        queue = new LinkedBlockingQueue<>();
        NettyThreadNameFactory factory = new NettyThreadNameFactory(GlobalConstants.Thread.GAME_KAFKA_TASK_EXECUTOR);
        //开启线程数量
        KafkaConfig kafkaConfig = SpringServiceManager.getSpringLoadService().getNettyGameServiceConfigService().getKafkaConfig();
        ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(kafkaConfig.getCorePoolSize(), kafkaConfig.getMaximumPoolSize(), kafkaConfig.getKeepAliveTime(), TimeUnit.SECONDS, new LinkedBlockingQueue<>(), factory);
        executorService = poolExecutor;
        for (int i = 0; i < 4; i++) {
            executorService.execute(new Worker());
        }
    }

    @Override
    public void shutdown() throws Exception {
        logger.info("STOP KAFKA EXECUTOR");
        run = false;
        executorService.shutdown();
    }

    private class Worker implements Runnable {

        @Override
        public void run() {
            try {
                while (run) {
                    sendMessage(queue.take());
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }


}
