package com.yeejoin.amos.message.kafka.config;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.config.SslConfigs;
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 org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.HashMap;
import java.util.Map;

/**
 * topic初始化
 *
 * @author litw
 * @create 2022/11/1 10:06
 */
@Configuration
class KafkaConfig {

    @Value("${kafka.auto-startup:false}")
    private boolean autoStartup;

    @Value("${queue.kafka.bootstrap-servers:}")
    private String bootstrapServers;

    @Value("${queue.kafka.consumer.group-id:}")
    private String groupId;

    @Value("${queue.kafka.consumer.enable-auto-commit:false}")
    private boolean enableAutoCommit;

    @Value("${queue.kafka.ssl.enabled:false}")
    private boolean sslEnabled;

    @Value("${queue.kafka.ssl.truststore.location:}")
    private String sslTruststoreLocation;

    @Value("${queue.kafka.ssl.truststore.password:}")
    private String sslTruststorePassword;

    @Value("${queue.kafka.confluent.sasl.jaas.config:}")
    private String saslConfig;

    @Value("${queue.kafka.confluent.sasl.mechanism:}")
    private String saslMechanism;

    @Value("${queue.kafka.confluent.security.protocol:}")
    private String securityProtocol;

    @Value("${queue.kafka.confluent.ssl.algorithm:}")
    private String sslAlgorithm;

    @Value("${queue.kafka.max.poll.records:}")
    private String maxPollRecords;

    @Bean(name = "kafkaRomaContainerFactory")
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaEsContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties()
                .setPollTimeout(3000);
        factory.getContainerProperties()
                .setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        factory.getContainerProperties()
                .setPollTimeout(15000);
        // 禁止消费者监听器自启动
        factory.setAutoStartup(autoStartup);
        return factory;
    }

    public ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    private Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        if (sslEnabled) {
            props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, sslTruststoreLocation);
            props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, sslTruststorePassword);
            props.put("sasl.mechanism", saslMechanism);
            props.put("sasl.jaas.config", saslConfig);
            props.put("ssl.endpoint.identification.algorithm", sslAlgorithm);
            props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol);
            props.put("max.poll.records", maxPollRecords);
            props.put("max.poll.interval.ms", "30000");
            props.put("session.timeout.ms", "30000");
        }
        return props;
    }
}