package com.muyu.processing.consumer;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.nacos.shaded.com.google.common.collect.Lists;
import com.muyu.common.core.constant.KafkaConstants;
import com.muyu.domain.Fence;
import com.muyu.domain.Vehicle;
import com.muyu.domain.WarnRule;
import com.muyu.domain.WarnStrategy;
import com.muyu.processing.interfaces.EventInterface;
import com.muyu.processing.utils.CacheUtil;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.Collection;
import java.util.Map;

/**
 * kafka消费者
 * @Author：杨鹏
 * @Package：com.muyu.processing.consumer
 * @Project：car-cloud-server
 * @name：KafkaConsumerService
 * @Date：2024/9/28 22:08
 */
@Log4j2
@Component
public class KafkaConsumerService implements InitializingBean {

    @Resource
    private KafkaConsumer kafkaConsumer;

    @Resource
    private CacheUtil cacheUtil;

//    @Resource
//    private EventInterface eventInterface;

    @Override
    public void afterPropertiesSet() throws Exception {

        Thread thread = new Thread(() -> {
            log.info("启动线程监听Topic: {}", "zeshi");
            ThreadUtil.sleep(1000);
            Collection<String> topics = Lists.newArrayList("zeshi");
            kafkaConsumer.subscribe(topics);
            while (true) {
                ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord consumerRecord : consumerRecords) {
                    //从ConsumerRecord中获取消费数据
                    String originalMsg = (String) consumerRecord.value();
                    log.info("从Kafka中消费的原始数据: " + originalMsg);
                    //把消费数据转换为JSON对象
                    JSONObject jsonObject = JSON.parseObject(originalMsg);
                    log.info("消费数据转换为JSON对象: " + jsonObject);
                    log.info("消费数据转换为JSON对象: " + jsonObject.toString());

                    String value = jsonObject.toString();
                    String vin = value.substring(0, 11);
                    Map<String, Object> map = (Map<String, Object>) cacheUtil.get(vin);
                    WarnRule warnRule = (WarnRule) map.get("warnRule");
                    WarnStrategy warnStrategy = (WarnStrategy) map.get("warnStrategy");
                    Vehicle vehicle = (Vehicle) map.get("vehicle");
                    Object breakdown = map.get("breakdown");
                    Fence fence = (Fence) map.get("fence");
//                    eventInterface.handle(jsonObject);
                }
            }
        });
        thread.start();

    }
}
