package org.dromara.common.kafka.config.properties;

import lombok.Data;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.util.CollectionUtils;

import java.time.Duration;
import java.util.*;

/**
 * kafka 配置简化
 *
 * @author ym
 * @version 1.0.0
 * @date 2024-08-23 13:40
 */
@Data
@ConfigurationProperties(prefix = "kafka")
public class MyKafkaProperties {

    /**
     * kafka 服务器
     */
    private List<String> bootstrapServers = new ArrayList<>(Collections.singletonList("localhost:9092"));
    /**
     * 配置信息
     */
    private final Map<String, String> properties = new HashMap<>();
    /**
     * 消费者信息
     */
    private final Consumer consumer = new Consumer();
    /**
     * 生产者
     */
    private final Producer producer = new Producer();
    /**
     * 是否开启
     */
    private Boolean enabled;

    /**
     * 通用配置构建
     */
    public Map<String, Object> buildCommonProperties() {
        Map<String, Object> properties = new HashMap<>();
        if (this.bootstrapServers != null) {
            properties.put("bootstrap.servers", this.bootstrapServers);
        }

        if (!CollectionUtils.isEmpty(this.properties)) {
            properties.putAll(this.properties);
        }

        return properties;
    }

    /**
     * 消费者配置构建
     */
    public Map<String, Object> buildConsumerProperties() {
        Map<String, Object> properties = buildCommonProperties();
        Map<String, Object> consumerMap = consumer.buildProperties();
        mergeProps(properties, consumerMap);
        return properties;
    }

    /**
     * 生产者配置构建
     */
    public Map<String, Object> buildProducerProperties() {
        Map<String, Object> properties = buildCommonProperties();
        Map<String, Object> producerMap = producer.buildProperties();
        mergeProps(properties, producerMap);
        return properties;
    }

    /**
     * 合并配置信息
     *
     * @param commonMap
     * @param map
     */
    private void mergeProps(Map<String, Object> commonMap, Map<String, Object> map) {
        map.forEach((at, val) -> {
            if (val != null) {
                commonMap.put(at, val);
            }
        });
    }

    /**
     * 消费者
     */
    @Data
    public static class Consumer {
        /**
         * 自动提交间隔
         */
        private Integer autoCommitInterval = 5000;
        /**
         * 消费偏移规则
         */
        private String autoOffsetReset = RulesType.LATEST.getName();
        /**
         * 服务器
         */
        private List<String> bootstrapServers;
        /**
         * 发出请求时传递给服务器的ID。用于服务器端日志记录。
         */
        private String clientId;
        /**
         * 是否自动提交
         */
        private Boolean enableAutoCommit = true;
        /**
         * 获取数据最大等待时间
         */
        private Duration fetchMaxWait = Duration.ofMillis(500);
        /**
         * 获取最大值
         */
        private Integer fetchMaxSize = 50 * 1024 * 1024;
        /**
         * 获取最小值
         */
        private Integer fetchMinSize = 10 * 1024 * 1024;
        /**
         * 组id
         */
        private String groupId;
        /**
         * 键序列化
         */
        private Class<?> keyDeserializer = StringDeserializer.class;
        /**
         * 值序列化
         */
        private Class<?> valueDeserializer = StringDeserializer.class;
        /**
         * 单次调用poll（）返回的最大记录数。
         */
        private Integer maxPollRecords = 500;
        /**
         * 最大
         */
        private Duration maxPollInterval = Duration.ofMillis(30 * 1000);
        /**
         * 属性配置
         */
        private final Map<String, String> properties = new HashMap<>();

        /**
         * 构建配置
         */
        public Map<String, Object> buildProperties() {
            Map<String, Object> properties = new HashMap<>();
            properties.put("bootstrap.servers", this.getBootstrapServers());
            properties.put("group.id", this.getGroupId());
            properties.put("enable.auto.commit", this.getEnableAutoCommit());
            properties.put("auto.offset.reset", this.getAutoOffsetReset());
            properties.put("max.poll.records", this.getMaxPollRecords());
            properties.put("auto.commit.interval.ms", this.getAutoCommitInterval());
            properties.put("max.poll.interval.ms", (int) this.getMaxPollInterval().toMillis());
            properties.put("fetch.max.wait.ms", (int) this.getFetchMaxWait().toMillis());
            properties.put("fetch.max.bytes", this.getFetchMaxSize());
            properties.put("fetch.min.bytes", this.getFetchMinSize());
            properties.put("key.deserializer", this.getKeyDeserializer());
            properties.put("value.deserializer", this.getValueDeserializer());
            properties.putAll(this.getProperties());
            return properties;
        }

        /**
         * 偏移量类型
         */
        public enum RulesType {
            /**
             * 自动重置偏移量到最早的偏移量
             */
            EARLIEST("earliest"),
            /**
             * 默认，自动重置偏移量为最新的偏移量
             */
            LATEST("latest"),
            /**
             * 如果消费组原来的偏移量不存在，则向消费者抛异常
             */
            NONE("none");

            private final String name;

            RulesType(String name) {
                this.name = name;
            }

            public String getName() {
                return name;
            }
        }
    }

    /**
     * 生产者
     */
    @Data
    public static class Producer {
        /**
         * 服务器
         */
        private List<String> bootstrapServers;
        /**
         * 批量大小
         */
        private Integer batchSize = 16;
        /**
         * 缓冲大小
         */
        private Integer bufferMemory = 32 * 1024 * 1024;
        /**
         * 发出请求时传递给服务器的ID。用于服务器端日志记录。
         */
        private String clientId;
        /**
         * 当大于零时，允许重试失败的发送。
         */
        private Integer retries;
        /**
         * 键序列化
         */
        private Class<?> keySerializer = StringSerializer.class;
        /**
         * 值序列化
         */
        private Class<?> valueSerializer = StringSerializer.class;
        /**
         * 生产者配置
         */
        private final Map<String, String> properties = new HashMap<>();

        /**
         * 构建配置
         */
        public Map<String, Object> buildProperties() {
            Map<String, Object> properties = new HashMap<>();
            properties.put("bootstrap.servers", this.getBootstrapServers());
            properties.put("client.id", this.getClientId());
            properties.put("retries", this.getRetries());
            properties.put("batch.size", this.getBatchSize());
            properties.put("buffer.memory", this.getBufferMemory());
            properties.put("key.serializer", this.getKeySerializer());
            properties.put("value.serializer", this.getValueSerializer());
            properties.putAll(this.getProperties());
            return properties;
        }
    }

}
