package com.yeejoin.amos.message.kafka;

import com.alibaba.fastjson.JSON;
import com.yeejoin.amos.message.utils.ClassToJsonUtil;
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.core.io.Resource;
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.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.Optional;

/**
 * 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_EVENT_ALARM = "romaSite/data/eventAlarm";

    @Autowired
    protected EmqKeeper emqKeeper;

    @Value("classpath:/json/commonMessage.json")
    private Resource commonMessage;

    /**
     * 消费单条消息,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数据
     * @param record record
     * @param ack ack
     */
    @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 object = JSONObject.fromObject(record.value());
                com.alibaba.fastjson.JSONObject jsonObj = ClassToJsonUtil.class2json(object, commonMessage, record.topic());
                emqKeeper.getMqttClient().publish(String.valueOf(jsonObj.get("mqTopic")), JSON.toJSONString(jsonObj).getBytes("UTF-8"), 0, false);
                log.info("接收到Roma消息对象: {}", object);
                ack.acknowledge();
            } catch (MqttException e) {
                log.error("解析数据失败,{}", e.getMessage());
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }
    }
//
//    /**
//     * 韶山换流对接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");
////                JSONObject object = JSONObject.fromObject(record.value());
////                String json = ClassToJsonUtil.class2json(object, commonMessage, MQTT_TOPIC_EVENT_ALARM);
////                emqKeeper.getMqttClient().publish(MQTT_TOPIC_EVENT_ALARM, json.getBytes(StandardCharsets.UTF_8), 0, false);
//
//                JSONObject object = JSONObject.fromObject(record.value());
//                com.alibaba.fastjson.JSONObject jsonObj = ClassToJsonUtil.class2json(object, commonMessage, record.topic());
//                emqKeeper.getMqttClient().publish(String.valueOf(jsonObj.get("mqTopic")), JSON.toJSONString(jsonObj).getBytes("UTF-8"), 0, false);
//                ack.acknowledge();
//            } catch (MqttException e) {
//                log.error("解析数据失败,{}", e.getMessage());
//            } catch (UnsupportedEncodingException e) {
//                e.printStackTrace();
//            }
//        }
//    }
//
//    /**
//     * 事件告警对接Kafka
//     * @param record record
//     * @param ack ack
//     *            groupId = kafkaConsumerGroup
//     *            该消息的消息格式为
//     *            {"data_class":"realdata","data_type":"alarm","op_type":"subscribe_emergency","condition":{"station_psr_id":"50edcb6c1b8a811030493c80a2014950ed9d4f59e8","station_name":"中州换流站","alarm_type":"yx_bw"},"data":[{"psrId":"D017020000000000000000999","astId":"D017020000000000000000999","equipType":"ASTType_0000111","eventType":"OtherSignal","alarmSource":"OWS","alarmLevel":"3","description":"2024-03-11 09:06:17::585 S2WCL12A E3.C01软水器再生结束信号 出现","dateTime":"2024-03-11 09:06:17.585"}]}
//     */
//
//    @KafkaListener(id = "kafkaConsumerEventAlarm", groupId = "kafkaConsumerGroupEventAlarm", topics = "#{'${queue.kafka.eventAlarm.topics}'.split(',')}", containerFactory = "kafkaRomaContainerFactory")
//    public void kafkaConsumerEventAlarm(ConsumerRecord<?, String> record, Acknowledgment ack) {
//        Optional<?> message = Optional.ofNullable(record.value());
//        if (message.isPresent()) {
//            try {
////                JSONObject messageObj = JSONObject.fromObject(record.value());
////                JSONArray dataArray = messageObj.getJSONArray("data");
////                JSONArray jsonArray = new JSONArray();
////                String timestamp = "";
////                for (Object obj : dataArray) {
////                    JSONObject finallyObj = new JSONObject();
////                    com.alibaba.fastjson.JSONObject detail = com.alibaba.fastjson.JSONObject.parseObject(com.alibaba.fastjson.JSONObject.toJSONString(obj));
////                    finallyObj.put("eventtextL1", detail.get("description"));
////                    finallyObj.put("pointId", detail.get("astId"));
////                    finallyObj.put("time", detail.get("dateTime"));
////                    jsonArray.add(finallyObj);
////                    timestamp = detail.get("dateTime").toString();
////                }
////                JSONObject jsonObjectMessage = new JSONObject();
////                jsonObjectMessage.put("warns", jsonArray);
////                jsonObjectMessage.put("timestamp", timestamp);
//
//                JSONObject object = JSONObject.fromObject(record.value());
//                com.alibaba.fastjson.JSONObject jsonObj = ClassToJsonUtil.class2json(object, commonMessage, record.topic());
//                emqKeeper.getMqttClient().publish(String.valueOf(jsonObj.get("mqTopic")), JSON.toJSONString(jsonObj).getBytes("UTF-8"), 0, false);
//                ack.acknowledge();
//            } catch (MqttException e) {
//                log.error("解析数据失败,{}", e.getMessage());
//            } catch (UnsupportedEncodingException e) {
//                e.printStackTrace();
//            }
//        }
//    }


/*    @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();
//    }
}