package com.elab.core.kafka.consumer;

import com.elab.log.utils.CatCrossProcess;
import com.jay.monitor.data.client.MonitorSendProducer;
import com.jay.monitor.data.client.utils.MonitorUtils;
import com.jay.monitor.data.core.model.serializable.MQDataDTO;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.kafka.listener.BatchAcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.util.Assert;

import java.util.List;

/**
 * @Module 监听器
 * @Description kafka批量消费监听器
 * @Author liukaixiong
 * @Date 2020/10/27 19:14
 */
public class BatchKafkaMsgListener implements BatchAcknowledgingMessageListener<String, String>, InitializingBean {

    private Logger logger = LoggerFactory.getLogger(getClass());
    private List<AbstractBatchKafkaConsumer> messageListeners;

    public BatchKafkaMsgListener(List<AbstractBatchKafkaConsumer> kafkaConsumer) {
        Assert.notNull(kafkaConsumer, "请实现一个AbstractKafkaConsumer的子类,并注册到ioc容器中");
        this.messageListeners = kafkaConsumer;
    }

    @Override
    public void onMessage(List<ConsumerRecord<String, String>> batchData, Acknowledgment acknowledgment) {
        if (batchData == null || batchData.size() == 0) {
            return;
        }
        // 默认不支持单个topic不同业务的情况
        ConsumerRecord<String, String> data = batchData.get(0);
        try {
            CatCrossProcess.processTransactionMsg("kafka-batch-" + data.topic(), data.partition() + "", (transaction) -> {
                long startTime = System.currentTimeMillis();
                try {
                    String topic = data.topic();
                    this.messageListeners.forEach((listener) -> {
                        if (listener.subscribeTopic().equals(topic)) {
                            listener.onMessage(batchData, acknowledgment);
                        }
                    });
                    long requestTime = System.currentTimeMillis() - startTime;
                    if (acknowledgment != null) {
                        sendMonitorData(batchData, 1, requestTime);
                    } else {
                        sendMonitorData(batchData, -1, requestTime);
                    }
                } catch (Exception e) {
                    transaction.setStatus(e);
                    logger.error("kafka消费失败", e);
                    sendMonitorData(batchData, -1, System.currentTimeMillis() - startTime);
                    throw e;
                }
                return null;
            });
        } catch (Exception e) {
            // 基本不会出现
            e.printStackTrace();
            logger.error("未知异常", e);
        }
    }


    private void sendMonitorData(List<ConsumerRecord<String, String>> batchData, Integer status, Long time) {
        for (int i = 0; i < batchData.size(); i++) {
            ConsumerRecord<String, String> data = batchData.get(i);
            MQDataDTO mqDataDTO = MonitorUtils.builderConsumerMqDataDTO(data.topic(), data.key(), data.partition() + "", data.offset(), data.value(), status, time);
            MonitorSendProducer.sendMsg(data.topic(), mqDataDTO);
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {

    }
}
