package com.yeejoin.amos.message.kafka;

import lombok.extern.slf4j.Slf4j;
import net.sf.json.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import org.typroject.tyboot.component.emq.EmqKeeper;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import static com.yeejoin.amos.message.kafka.Constant.*;

/**
 * kafka 消费服务
 *
 * @author litw
 * @create 2022/11/1 10:06
 **/
@Slf4j
@Service
public class KafkaConsumerService {
    private static final String MQTT_TOPIC = "romaSite/data/transmit";
    private static final String MQTT_TOPIC_SHAOSHAN = "romaSite/data/shaoshan";

    private static final String PROVINCE_MQTT_TOPIC = "province/data/transport";
    @Autowired
    protected EmqKeeper emqKeeper;
    @Value("${system.zxj}")
    private boolean isZxj;

    /**
     * 批量消费kafka消息
     * Kafka消息转emq
     *
     * @param consumerRecords messages
     * @param ack             ack
     */
    @KafkaListener(id = "consumerSingle", groupId = "zhTestGroup", topics = "#{'${kafka.topics}'.split(',')}")
    public void listen1(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment ack) {
        try {
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value());
                if (kafkaMessage.isPresent()) {
                    JSONObject messageObj = JSONObject.fromObject(kafkaMessage.get());
                    if (messageObj.has(TOPIC)) {
                        emqKeeper.getMqttClient().publish(messageObj.optString(TOPIC), messageObj.getJSONObject(DATA).toString()
                                .getBytes(StandardCharsets.UTF_8), 0, false);
                    }
                    log.info("kafka消费zhTestGroup消息{}", messageObj);
                }
            }
        } catch (Exception e) {
            log.error("kafka失败，当前失败的批次: data:{}", consumerRecords);
        } finally {
            ack.acknowledge();
        }
    }

    /**
     * 批量消费kafka消息
     * 监听数据表变化kafka数据转发emq
     *
     * @param consumerRecords messages
     * @param ack             ack
     */
    @KafkaListener(id = "provinceMessage", groupId = "province", topics = "#{'${kafka.topics}'.split(',')}")
    public void listen(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment ack) {
        try {
            if (isZxj) {
                for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                    Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value());
                    if (kafkaMessage.isPresent()) {
                        JSONObject messageObj = JSONObject.fromObject(kafkaMessage.get());
                        String type = messageObj.optString(TYPE);
                        String table = messageObj.optString(TABLE);
                        if (Arrays.asList(INSERT, UPDATE).contains(type) && DBTableTypeEnum.have(table) != null) {
                            JSONObject data = (JSONObject) messageObj.getJSONArray(DATA).get(0);
                            data.put(DB_TYPE, type);
                            data.put(TABLE, table);
                            emqKeeper.getMqttClient().publish(PROVINCE_MQTT_TOPIC, data.toString().getBytes(StandardCharsets.UTF_8), 0, false);
                            log.info("kafka消费province消息{}", messageObj);
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("kafka失败，当前失败的批次: data:{}", consumerRecords);
        } finally {
            ack.acknowledge();
        }
    }

    /**
     * 消费单条消息,topics 可以监听多个topic，如：topics = {"topic1", "topic2"}
     *
     * @param message 消息
     *
     */
    @KafkaListener(id = "consumerSingle", idIsGroup = false, topics = "#{'${kafka.topics}'.split(',')}", concurrency = "2")
    public void consumerSingle(String message, Acknowledgment ack) {
        JSONObject messageObj = JSONObject.fromObject(message);
        try {
            String topic = messageObj.getString("topic");
            JSONObject data = messageObj.getJSONObject("data");
            emqKeeper.getMqttClient().publish(topic, data.toString().getBytes(StandardCharsets.UTF_8), 0, false);
            ack.acknowledge();
        } catch (MqttException e) {
            log.error("解析数据失败,{}", e.getMessage());
        }
    }

    /**
     * 转发苏州，绍兴换流站Kafka数据对emq
     *
     * @param record record
     * 绍兴，苏州换流站对接Kafka数据
     * @param record record
     */
    @KafkaListener(id = "kafkaRoma", groupId = "kafkaRoma", topics = "#{'${queue.kafka.topics}'.split(',')}", containerFactory = "kafkaRomaContainerFactory")
    public void kafkaListener(ConsumerRecord<?, String> record, Acknowledgment ack) {
            Optional<?> messages = Optional.ofNullable(record.value());
            if (messages.isPresent()) {
                try {
                    JSONObject messageObj = JSONObject.fromObject(record.value());
                    if (messageObj.getJSONObject(BODY).isEmpty()) {
                        messageObj.put(DATA_TYPE, STATE);
                    }
                    emqKeeper.getMqttClient().publish(MQTT_TOPIC, messageObj.toString().getBytes(StandardCharsets.UTF_8), 0, false);
                } catch (MqttException e) {
                    log.error("解析数据失败,{}", e.getMessage());
                } finally {
                    ack.acknowledge();
                }
            }
    }

    /**
     * 韶山换流对接Kafka
     * @param record record
     * @param ack ack
     */
    @KafkaListener(id = "kafkaConsumer", groupId = "kafkaConsumerGroup", topics = "#{'${queue.kafka.shaoshan.topics}'.split(',')}", containerFactory = "kafkaRomaContainerFactory")
    public void kafkaConsumer(ConsumerRecord<?, String> record, Acknowledgment ack) {
        Optional<?> message = Optional.ofNullable(record.value());
        if (message.isPresent()) {
            try {
                JSONObject messageObj = JSONObject.fromObject(record.value());
                JSONObject data = messageObj.getJSONObject("body");
                emqKeeper.getMqttClient().publish(MQTT_TOPIC_SHAOSHAN, data.toString().getBytes(StandardCharsets.UTF_8), 0, false);
                ack.acknowledge();
            } catch (MqttException e) {
                log.error("解析数据失败,{}", e.getMessage());
            }
        }
    }


    ///**
    // * 消费单条消息,topics 可以监听多个topic，如：topics = {"topic1", "topic2"}
    // *
    // * @param message 消息
    // */
    //@KafkaListener(id = "consumerSingle", groupId = "zhTestGroup", topics = "#{'${kafka.topics}'.split(',')}", concurrency = "2")
    //public void consumerSingle(String message, Acknowledgment ack) {
    //    JSONObject messageObj = JSONObject.fromObject(message);
    //    try {
    //        String topic = null;
    //        JSONObject data = null;
    //        if (messageObj.has("topic")) {
    //            topic = messageObj.getString("topic");
    //            data = messageObj.getJSONObject("data");
    //            emqKeeper.getMqttClient().publish(topic, data.toString().getBytes(StandardCharsets.UTF_8), 0, false);
    //            ack.acknowledge();
    //            log.info("消息转发成功" + messageObj);
    //        }
    //    } catch (Exception e) {
    //        log.error("消息转发失败" + e.getMessage(), e);
    //    }
    //}
    //

    ///**
    // * 省级消息转发
    // *
    // * @param message 省级消息
    // * @param ack     ack
    // */
    //@KafkaListener(id = "provinceMessage", groupId = "province", topics = "#{'${kafka.topics}'.split(',')}", concurrency = "2")
    //public void consumerSingle1(String message, Acknowledgment ack) {
    //    log.info("省级消息转发打印" + message);
    //    if(isZxj) {
    //        Optional<?> messages = Optional.ofNullable(message);
    //        if (messages.isPresent()) {
    //            try {
    //                JSONObject jsonObject = JSONObject.fromObject(message);
    //                String type = jsonObject.optString("type");
    //                String table = jsonObject.optString("table");
    //                if (Arrays.asList(Constant.INSERT, Constant.UPDATE).contains(type) && DBTableTypeEnum.have(table) != null) {
    //                    if (Arrays.asList("INSERT", "UPDATE").contains(type)) {
    //                        JSONArray array = jsonObject.getJSONArray("data");
    //                        JSONObject data = (JSONObject)array.get(0);
    //                        data.put("dbType", type);
    //                        data.put("table", table);
    //                        emqKeeper.getMqttClient().publish(PROVINCE_MQTT_TOPIC, data.toString().getBytes(StandardCharsets.UTF_8), 0, false);
    //                        log.info("省级消息: {}", data);
    //                    }
    //                }
    //            } catch (MqttException e) {
    //                log.error("消息转发失败" + e.getMessage(), e);
    //            }
    //            ack.acknowledge();
    //        }
    //    }
    //}


/*    @KafkaListener(id = "consumerBatch", topicPartitions = {
            @TopicPartition(topic = "hello-batch1", partitions = "0"),
            @TopicPartition(topic = "hello-batch2", partitionOffsets = @PartitionOffset(partition = "2", initialOffset = "4"))
    })*/
//    /**
//     * 批量消费消息
//     * @param messages
//     */
//    @KafkaListener(id = "consumerBatch", topics = "test-batch")
//    public void consumerBatch(List<ConsumerRecord<String, String>> messages) {
//        log.info("consumerBatch =====> messageSize: {}", messages.size());
//        log.info(messages.toString());
//    }

//    /**
//     * 指定消费异常处理器
//     * @param message
//     */
//    @KafkaListener(id = "consumerException", topics = "kafka-test-topic", errorHandler = "consumerAwareListenerErrorHandler")
//    public void consumerException(String message) {
//        throw new RuntimeException("consumer exception");
//    }
//
//    /**
//     * 验证ConsumerInterceptor
//     * @param message
//     */
//    @KafkaListener(id = "interceptor", topics = "consumer-interceptor")
//    public void consumerInterceptor(String message) {
//        log.info("consumerInterceptor ====> message: {}", message);
//    }
//
//
    //kafka的监听器，topic为"zhTest"，消费者组为"zhTestGroup"
    //@KafkaListener(topics = "test", groupId = "zhTestGroup")
    //public void listenZhugeGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
    //    String value = record.value();
    //    System.out.println(value);
    //    System.out.println(record);
    //    //手动提交offset
    //    ack.acknowledge();
    //}

//    //kafka的监听器，topic为"zhTest"，消费者组为"zhTestGroup"
//    @KafkaListener(topics = "test", groupId = "zhTestGroup")
//    public void listenZhugeGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
//        String value = record.value();
//        System.out.println(value);
//        System.out.println(record);
//        //手动提交offset
//        ack.acknowledge();
//    }
}
