package cn.com.taiji.dataService.utils;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Objects;

@Component
@Slf4j
public class StringProducer {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(String topic, String jsonStr) {
        kafkaTemplate.send(topic, jsonStr)
                .addCallback(success -> {
                    if (Objects.nonNull(success)) {
                        RecordMetadata metadata = success.getRecordMetadata();
                        log.info("生产者成功发送数据到kafka！topic: {}, partition: {}, offset: {}, json:{}",
                                metadata.topic(), metadata.partition(), metadata.offset(), jsonStr);
                    }
                }, failure -> log.error("发送数据失败，失败原因为: {}", failure.getMessage()));
    }
}
