package com.elab.core.kafka.sender;

import com.alibaba.fastjson.JSON;
import com.elab.core.kafka.config.KafkaConsumerConfig;
import junit.framework.TestCase;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;

/**
 * @Module TODO
 * @Description TODO
 * @Author liukaixiong
 * @Date 2020/10/28 13:49
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = {KafkaConsumerConfig.class})
public class KafkaProducerImplTest extends TestCase {
    private Logger logger = LoggerFactory.getLogger(getClass());
//    @Autowired
//    private KafkaProducer<String, String> kafkaProducer;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    private final String defaultTopic = "xcx_behavior_topic_test";

    @Test
    public void testTemplate() throws ExecutionException, InterruptedException {
        Map<String, Object> map = new HashMap<>();
        map.put("est", "123123");
        kafkaTemplate.setDefaultTopic(defaultTopic);
        ListenableFuture<SendResult<String, String>> send = kafkaTemplate.send(defaultTopic, JSON.toJSONString(map));

        SendResult<String, String> stringStringSendResult = send.get();
        RecordMetadata recordMetadata = stringStringSendResult.getRecordMetadata();
        Assert.assertTrue(recordMetadata.hasOffset());
    }

//    @Test
//    public void testSend() throws Exception {
//        String value = "abc";
//        ProducerRecord<String, String> record = new ProducerRecord<String, String>(defaultTopic, value);
//
//        Future<RecordMetadata> send = kafkaProducer.send(record);
//
//        RecordMetadata recordMetadata = send.get();
//        logger.info(" 结果: " + JSON.toJSONString(recordMetadata));
//        Assert.assertTrue(recordMetadata.hasOffset());
//    }
}