package com.elab.core.kafka.consumer;

import com.alibaba.fastjson.JSON;
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.AcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.util.Assert;

import java.util.List;
import java.util.Map;

/**
 * @Module 监听者
 * @Description kafka的消息监听
 * @Author liukaixiong
 * @Date 2020/10/27 19:14
 */
public class KafkaMsgListener implements AcknowledgingMessageListener<String, String>, InitializingBean {

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

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

    @Override
    public void onMessage(ConsumerRecord<String, String> data, Acknowledgment acknowledgment) {
        Map<String, String> valueMap = JSON.parseObject(data.value(), Map.class);
        CatCrossProcess.buildRemoteMQMsg("kafka-" + data.topic(), data.partition() + "", valueMap, () -> {
            long startTime = System.currentTimeMillis();
            try {
                String topic = data.topic();
                this.messageListeners.forEach((listener) -> {
                    if (listener.subscribeTopic().equals(topic)) {
                        listener.onMessage(data, acknowledgment);
                    }
                });
                long requestTime = System.currentTimeMillis() - startTime;
                if (acknowledgment != null) {
                    sendMonitorData(data, 1, requestTime);
                } else {
                    sendMonitorData(data, -1, requestTime);
                }
            } catch (Exception e) {
                logger.error("kafka消费失败", e);
                sendMonitorData(data, -1, System.currentTimeMillis() - startTime, e);
                throw e;
            }
            return acknowledgment;
        });
    }

    private void sendMonitorData(ConsumerRecord<String, String> data, Integer status, Long time) {
        sendMonitorData(data, status, time, null);
    }

    private void sendMonitorData(ConsumerRecord<String, String> data, Integer status, Long time, Throwable e) {
        MQDataDTO mqDataDTO = MonitorUtils.builderConsumerMqDataDTO(data.topic(), data.key(), data.partition() + "", data.offset(), data.value(), status, time, e);
        MonitorSendProducer.sendMsg(data.topic(), mqDataDTO);
    }

    @Override
    public void afterPropertiesSet() throws Exception {

    }
}
