package com.muyu.data.consumer;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.nacos.shaded.com.google.common.collect.Lists;
import com.muyu.common.kafka.constant.KafkaConstants;
import com.muyu.data.core.StartStrategy;
import com.muyu.data.domain.Information;
import jakarta.annotation.Resource;
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 java.time.Duration;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;

/**
 * kafka监听
 * @author WangXIn
 */
@Component
@Log4j2
public class MessageConsumer implements InitializingBean {

    @Resource
    private KafkaConsumer kafkaConsumer;
    @Resource
    private StartStrategy startStrategy;

    @Override
    public void afterPropertiesSet() throws Exception {
        Thread thread = new Thread(() -> {
            log.info("启动线程监听Topic: {}", KafkaConstants.KafkaTopic);
            ThreadUtil.sleep(1000);
            Collection<String> topics = Lists.newArrayList(KafkaConstants.KafkaTopic);
            kafkaConsumer.subscribe(topics);
            while (true) {
                System.out.println("开始消费数据,等待中...");
                ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord consumerRecord : consumerRecords) {
                    //1.从ConsumerRecord中获取消费数据
                    String originalMsg = (String) consumerRecord.value();
                    log.info("从Kafka中消费的原始数据: " + originalMsg);
                    //2.把消费数据转换为DTO对象
                    JSONObject jsonObject = JSONObject.parseObject(originalMsg);
                    Information information = jsonObject.getObject("key", Information.class);
                    log.info("从Kafka中消费的实体数据: " + information);
                    // 执行策略
                    startStrategy.applyStrategy(information);
                }
            }
        });
        thread.start();
    }
}
