package org.dromara.common.kafka.config;

import lombok.RequiredArgsConstructor;
import org.dromara.common.kafka.config.properties.MyKafkaProperties;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.*;

import java.util.Map;

/**
 * Kafka 自动配置
 *
 * @author ym
 * @version 1.0.0
 * @date 2024-08-23 13:37
 */
@AutoConfiguration
@EnableConfigurationProperties(MyKafkaProperties.class)
@ConditionalOnProperty(value = "kafka.enabled", havingValue = "true")
@RequiredArgsConstructor
public class KafkaProducerConfig {

    private final MyKafkaProperties kafkaProperties;

    @Bean
    public KafkaTemplate<?, ?> kafkaTemplate(ProducerFactory<Object, Object> defaultKafkaProducerFactory) {
        return new KafkaTemplate<>(defaultKafkaProducerFactory);
    }

    /**
     * 生产者工厂
     *
     * @return
     */
    @Bean
    public ProducerFactory<?, ?> kafkaProducerFactory() {
        Map<String, Object> producerProperties = kafkaProperties.buildProducerProperties();
        return new DefaultKafkaProducerFactory<>(producerProperties);
    }

    /**
     * 消费者工厂
     *
     * @return
     */
    @Bean
    public ConsumerFactory<?, ?> kafkaConsumerFactory() {
        Map<String, Object> consumerProperties = kafkaProperties.buildConsumerProperties();
        return new DefaultKafkaConsumerFactory<>(consumerProperties);
    }

}
