package com.fzzy.gateway.hx2023.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;


/**
 * 使用KAFKA推动返回信息实现，针对当前网关
 */
@Slf4j
@Component
public class KafkaDeviceReportService {


    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;


    public String publishWithTopic(String sendData, String topic) {
        //推送信息
        kafkaTemplate.send(topic, sendData).addCallback(success -> {
            // 消息发送到的topic
            // String topic = success.getRecordMetadata().topic();
            // 消息发送到的分区
            int partition = success.getRecordMetadata().partition();
            // 消息在分区内的offset
            long offset = success.getRecordMetadata().offset();

            log.info("---推送至KAFKA成功--:{}-{}-{}-{}", topic, partition, offset, sendData);
        }, failure -> {
            log.info("---推送至KAFKA失败--:{}-{}-{}-{}", topic, sendData);
            rePublishWithTopic(sendData, topic);
        });
        return "SUCCESS";
    }

    private void rePublishWithTopic(String sendData, String topic) {
        //推送信息
        kafkaTemplate.send(topic, sendData).addCallback(success -> {
            // 消息发送到的topic
            // String topic = success.getRecordMetadata().topic();
            // 消息发送到的分区
            int partition = success.getRecordMetadata().partition();
            // 消息在分区内的offset
            long offset = success.getRecordMetadata().offset();

            log.info("---推送至KAFKA成功--:{}-{}-{}-{}", topic, partition, offset, sendData);
        }, failure -> {
            log.info("---推送至KAFKA失败--:{}-{}-{}-{}", topic, sendData);
        });
    }


    @KafkaListener(topics = {"TOPIC_ZLJ_GRAIN_TEMPERATURE"})
    public void onMessage1(ConsumerRecord<?, ?> record) {
        //  log.debug("获取消费信息：" + record.topic() + "-" + record.partition() + "-" + record.value());
    }

}