package com.elab.core.kafka.config;

import com.elab.core.kafka.consumer.AbstractBatchKafkaConsumer;
import com.elab.core.kafka.consumer.AbstractKafkaConsumer;
import com.elab.core.kafka.consumer.BatchKafkaMsgListener;
import com.elab.core.kafka.consumer.KafkaMsgListener;
import com.elab.core.kafka.sender.KafkaMonitorProducer;
import com.elab.core.kafka.sender.KafkaProducerImpl;
import com.elab.log.ext.CatIdMonitorRule;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.support.ProducerListener;

import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Configuration
// @EnableKafka // 这里是 可以开启@Listener注解的
@EnableConfigurationProperties({KafkaProperties.class})
public class KafkaConsumerConfig {
    private Logger logger = LoggerFactory.getLogger(getClass());

    @Value(value = "${java.security.auth.login.trust-store-location:}")
    private String sslTruststoreLocation;

    @Value(value = "${java.security.auth.login.config:}")
    private String authConfig;


    @Autowired(required = false)
    private ProducerListener<String, String> producerListeners;

    //    @Bean
//    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory(KafkaProperties kafkaProperties) {
//        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
//        factory.setConsumerFactory(consumerFactory(kafkaProperties));
//        // 多线程并发消费
////        factory.setConcurrency(concurrency);
//        // 设置手动提交
//        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
//        factory.getContainerProperties().setPollTimeout(1500);
//        return factory;
//    }

    @Bean
    @ConditionalOnMissingBean(value = {KafkaMonitorProducer.class})
    public KafkaMonitorProducer kafkaMonitorProducer() {
        return new KafkaMonitorProducer();
    }

    @Bean
    @ConditionalOnBean(value = {AbstractKafkaConsumer.class})
    public KafkaMessageListenerContainer<String, String> kafkaMessageListenerContainer(KafkaProperties kafkaProperties, List<AbstractKafkaConsumer> kafkaConsumer) {
        List<String> topicList = kafkaConsumer.stream().map(AbstractKafkaConsumer::subscribeTopic).collect(Collectors.toList());
        logger.info("关注kafka的topic:" + topicList);
        ContainerProperties containerProps = new ContainerProperties(topicList.toArray(new String[topicList.size()]));

        KafkaMsgListener kafkaMsgListener = new KafkaMsgListener(kafkaConsumer);
        // 包装成监控

        containerProps.setMessageListener(kafkaMsgListener);
        containerProps.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        containerProps.setSyncCommits(true);
        DefaultKafkaConsumerFactory<String, String> cf =
                new DefaultKafkaConsumerFactory<String, String>(getConfigConsumerMap(kafkaProperties));
        KafkaMessageListenerContainer<String, String> container =
                new KafkaMessageListenerContainer<>(cf, containerProps);
        return container;
    }

    @Bean
    @ConditionalOnBean(value = {AbstractBatchKafkaConsumer.class})
    public KafkaMessageListenerContainer<String, String> batchKafkaMessageListenerContainer(KafkaProperties kafkaProperties, List<AbstractBatchKafkaConsumer> kafkaConsumer) {

        List<String> topicList = kafkaConsumer.stream().map(AbstractBatchKafkaConsumer::subscribeTopic).collect(Collectors.toList());

        logger.info("关注kafka的topic:" + topicList);
        ContainerProperties containerProps = new ContainerProperties(topicList.toArray(new String[topicList.size()]));

        BatchKafkaMsgListener kafkaMsgListener = new BatchKafkaMsgListener(kafkaConsumer);

        // 包装成监控
        containerProps.setMessageListener(kafkaMsgListener);
        containerProps.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        containerProps.setSyncCommits(true);
        DefaultKafkaConsumerFactory<String, String> cf =
                new DefaultKafkaConsumerFactory<String, String>(getConfigConsumerMap(kafkaProperties));
        KafkaMessageListenerContainer<String, String> container =
                new KafkaMessageListenerContainer<>(cf, containerProps);

        return container;
    }

    @Bean
    @ConditionalOnMissingBean(value = {CatIdMonitorRule.class})
    public CatIdMonitorRule topicMonitorRule() {
        return new CatIdMonitorRule();
    }


    public ConsumerFactory<String, String> consumerFactory(KafkaProperties kafkaProperties) {
        Map<String, Object> props = getConfigConsumerMap(kafkaProperties);
        return new DefaultKafkaConsumerFactory<String, String>(props);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate(KafkaProperties kafkaProperties) {
        Map<String, Object> props = new LinkedHashMap<>();
        //设置接入点，请通过控制台获取对应 Topic 的接入点
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
        //消息队列 Kafka 消息的序列化方式
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        //请求的最长等待时间
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
        //设置客户端内部重试次数。
        props.put(ProducerConfig.RETRIES_CONFIG, 5);
        //设置客户端内部重试间隔。
        props.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 3000);
        if (StringUtils.isNotBlank(sslTruststoreLocation)) {
            // 外网配置
            props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, sslTruststoreLocation);
            props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
            props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
            //SASL鉴权方式，保持不变
            props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
            System.setProperty("java.security.auth.login.config", authConfig);
            props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
        }
        DefaultKafkaProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<String, String>(props);
        KafkaProducerImpl<String, String> kafkaTemplate = new KafkaProducerImpl<>(producerFactory, false);
        kafkaTemplate.setProducerListener(producerListeners);
        return kafkaTemplate;
    }


//    public Map<String, Object> getConfigProducerMap(KafkaProperties kafkaProperties) {
//        System.setProperty("java.security.auth.login.config", "D:/application/kafka/kafka_client_jaas.conf");
//        Map<String, Object> props = kafkaProperties.buildProducerProperties();
//        //设置接入点，请通过控制台获取对应 Topic 的接入点
//        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
//        //消息队列 Kafka 消息的序列化方式
//        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
//        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
//        //请求的最长等待时间
//        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
//        //设置客户端内部重试次数。
//        props.put(ProducerConfig.RETRIES_CONFIG, 5);
//        //设置客户端内部重试间隔。
//        props.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 3000);
//        // 外网配置
//        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "D:/application/kafka/kafka.client.truststore.jks");
//        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
//        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
//        //SASL鉴权方式，保持不变
//        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
//        props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
//        return props;
//    }

    private Map<String, Object> getConfigConsumerMap(KafkaProperties kafkaProperties) {

        Map<String, Object> props = kafkaProperties.buildConsumerProperties();
        if (StringUtils.isNotBlank(sslTruststoreLocation)) {
            System.setProperty("java.security.auth.login.config", authConfig);
            props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, sslTruststoreLocation);
            //根证书存储的密码，保持不变。
            props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
            //接入协议，目前支持使用SASL_SSL协议接入。
            props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
            //SASL鉴权方式，保持不变。
            props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        }
        //两次Poll之间的最大允许间隔。
        //消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，服务端将消费者从Consumer Group移除并触发rebalance，默认30s。
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        //设置单次拉取的量，走公网访问时，该参数会有较大影响。
        props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 32000);
        props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 32000);
        //Hostname校验改成空。
        props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
        return props;
    }

}