package com.muyu.data.processing.kafka;


import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.nacos.shaded.com.google.common.collect.Lists;
import com.muyu.common.kafka.constants.KafkaConstants;
import com.muyu.data.processing.domain.BasicData;
import com.muyu.data.processing.service.DataProcessingService;
import com.muyu.data.processing.strategy.core.StartStrategy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
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: 胡杨
 * @Name: KafkaConsumerService
 * @Description: kafka消费者
 * @CreatedDate: 2024/9/27 上午9:27
 * @FilePath: com.muyu.data.processing.kafka
 */

@Slf4j
@Component
public class KafkaConsumerService implements InitializingBean {
    @Resource
    private KafkaConsumer kafkaConsumer;
    @Resource
    private StartStrategy startStrategy;

    @Override
    public void afterPropertiesSet() {
        new Thread(() -> {
            log.info("启动线程监听Topic: {}", KafkaConstants.KafkaTopic);
            Collection<String> topics = Lists.newArrayList(KafkaConstants.KafkaTopic);
            kafkaConsumer.subscribe(topics);
            while (true) {
                try {
                    ThreadUtil.sleep(1000);
                    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对象
                        List<BasicData> dataList = JSONUtil.toList(originalMsg, BasicData.class);
                        log.info("从Kafka中消费的实体数据: " + dataList);
                        // 执行策略
                        startStrategy.applyStrategy(getDataMap(dataList));
                    }
                }catch (Exception e) {
                    log.error("kafka执行异常：" + e.getMessage());
                }
            }
        }).start();
    }

    private HashMap<String, BasicData> getDataMap(List<BasicData> dataList) {
        HashMap<String, BasicData> basicDataHashMap = new HashMap<>();
        dataList.forEach(data -> basicDataHashMap.put(data.getKey(), data));
        return basicDataHashMap;
    }
}
