package com.muyu.common.kafka.config;

import com.muyu.common.core.text.StrFormatter;
import com.muyu.common.kafka.constant.KafkaConfigConstants;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;

/**
 * @Author: wangXin
 * @Date 2024/9/28 20:32
 * @Description Kafka消费者配置
 */
@Configuration
public class KafkaConsumerConfig {

    /**
     * 服务端IP
     */
    @Value("${kafka.consumer.bootstrap-servers-ip}")
    private String bootstrapServersIP;

    /**
     * 服务端口号
     */
    @Value("${kafka.consumer.bootstrap-servers-port}")
    private String bootstrapServersPort;

    /**
     * 开启消费者偏移量
     */
    @Value("${kafka.consumer.enable-auto-commit}")
    private Boolean enableAutoCommit;

    /**
     * 自动提交时间间隔
     */
    @Value("${kafka.consumer.auto-commit-interval}")
    private Integer autoCommitInterval;

    /**
     * 自动重置偏移量
     */
    @Value("${kafka.consumer.auto-offset-reset}")
    private String autoOffsetReset;

    /**
     * 请求阻塞的最大时间
     */
    @Value("${kafka.consumer.fetch-max-wait}")
    private Integer fetchMaxWait;

    /**
     * 请求应答的最小字节数
     */
    @Value("${kafka.consumer.fetch-min-size}")
    private Integer fetchMinSize;

    /**
     * 心跳间隔时间
     */
    @Value("${kafka.consumer.heartbeat-interval}")
    private Integer heartbeatInterval;

    /**
     * 一次调用poll返回的最大记录条数
     */
    @Value("${kafka.consumer.max-poll-records}")
    private Integer maxPollRecords;

    /**
     * 指定消费组
     */
    @Value("${kafka.consumer.group-id}")
    private String groupId;

    /**
     * Kafka消费者初始化配置
     * @return Kafka消费者实例
     */
    @Bean
    public KafkaConsumer<String, String> kafkaConsumer() {
        HashMap<String, Object> configs = new HashMap<>();
        configs.put(KafkaConfigConstants.BOOTSTRAP_SERVERS,
                StrFormatter.format("{}:{}", bootstrapServersIP, bootstrapServersPort));
        configs.put(KafkaConfigConstants.ENABLE_AUTO_COMMIT, enableAutoCommit);
        configs.put(KafkaConfigConstants.AUTO_COMMIT_INTERVAL, autoCommitInterval);
        configs.put(KafkaConfigConstants.AUTO_OFFSET_RESET, autoOffsetReset);
        configs.put(KafkaConfigConstants.FETCH_MAX_WAIT, fetchMaxWait);
        configs.put(KafkaConfigConstants.FETCH_MIN_SIZE, fetchMinSize);
        configs.put(KafkaConfigConstants.HEARTBEAT_INTERVAL, heartbeatInterval);
        configs.put(KafkaConfigConstants.MAX_POLL_RECORDS, maxPollRecords);
        configs.put(KafkaConfigConstants.GROUP_ID, groupId);
        Deserializer<String> keyDeserializer = new StringDeserializer();
        Deserializer<String> valueDeserializer = new StringDeserializer();
        return new KafkaConsumer<>(configs, keyDeserializer, valueDeserializer);
    }
}
