package com.elab.core.kafka.sender;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.elab.log.utils.CatCrossProcess;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;

/**
 * @Module kafka生产者
 * @Description 负责产生消息的对象
 * @Author liukaixiong
 * @Date 2020/10/27 15:02
 */
public class KafkaProducerImpl<K, V> extends KafkaTemplate<K, V> {

    private Logger LOG = LoggerFactory.getLogger(KafkaProducerImpl.class);

    private final String defaultType = "kafka";

    public KafkaProducerImpl(ProducerFactory<K, V> producerFactory, boolean autoFlush) {
        super(producerFactory, autoFlush);
    }

    @Override
    protected ListenableFuture<SendResult<K, V>> doSend(ProducerRecord<K, V> record) {
        String topic = record.topic();
        ListenableFuture<SendResult<K, V>> recordMetadata = (ListenableFuture<SendResult<K, V>>) CatCrossProcess.createRemoteMQMsg(defaultType, topic, (msgContextMap) -> {
//            // 将上下文消息携带到请求中
            JSONObject valueObject = JSON.parseObject(record.value().toString());
            logger.debug("发送消息内容:" + record.value());
            msgContextMap.forEach((K, V) -> {
                valueObject.put(K, V);
            });
            ProducerRecord<K, V> producerRecord = new ProducerRecord<K, V>(topic, record.partition(), record.timestamp(), record.key(), (V) valueObject.toJSONString());
            try {
                ListenableFuture<SendResult<K, V>> listenableFuture = KafkaProducerImpl.super.doSend(producerRecord);
                logger.debug("消息推送成功!" + listenableFuture.get().toString());
                return listenableFuture;
            } catch (Exception e) {
                LOG.error("推送kafka失败", e);
            }
            return null;
        });
        return recordMetadata;
    }


}
