package cn.lihongjie.coal.rabbitmq;

import com.fasterxml.jackson.databind.ObjectMapper;

import org.springframework.amqp.core.*;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.RabbitTemplateCustomizer;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.boot.convert.DurationStyle;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.format.DateTimeFormatter;
import java.util.*;

@EnableRabbit
@Configuration
public class RabbitMQConfiguration {

    public static final String SYS_EXCHANGE = "sysExchange";
    public static final String SYS_DELAY_EXCHANGE = "sysDelayExchange";
    public static final String DELAY_QUEUE = "delayQueue";

    public static final String[] DELAY_QUEUES =
            new String[] {
                "1s", "5s", "10s", "30s", "1m", "2m", "3m", "4m", "5m", "10m", "15m", "20m", "25m",
                "30m", "35m", "40m", "45m", "50m", "55m", "1h", "2h", "3h", "4h", "5h", "6h", "12h",
                "1d"
            };

    public static final Long[] DELAY_QUEUES_TIME =
            Arrays.stream(DELAY_QUEUES)
                    .map(x -> DurationStyle.detectAndParse(x).toMillis())
                    .toArray(Long[]::new);
    public static final DateTimeFormatter DATE_TIME_FORMATTER =
            DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    @Autowired private ObjectMapper objectMapper;

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            ConnectionFactory connectionFactory,
            SimpleRabbitListenerContainerFactoryConfigurer configurer) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        factory.setDefaultRequeueRejected(false);
        factory.setMessageConverter(new Jackson2JsonMessageConverter(new ObjectMapper()));
        return factory;
    }

    @Bean
    RabbitTemplateCustomizer rabbitTemplateCustomizer() {

        return template -> {
            Jackson2JsonMessageConverter messageConverter =
                    new Jackson2JsonMessageConverter(objectMapper);

            SimpleMessageConverter simpleMessageConverter = new SimpleMessageConverter();

            template.setMessageConverter(
                    new MessageConverter() {
                        @Override
                        public Message toMessage(Object object, MessageProperties messageProperties)
                                throws MessageConversionException {

                            if (object instanceof String || object instanceof byte[]) {
                                return simpleMessageConverter.toMessage(object, messageProperties);
                            }


                            return messageConverter.toMessage(object, messageProperties);
                        }

                        @Override
                        public Object fromMessage(Message message)
                                throws MessageConversionException {
                            return messageConverter.fromMessage(message);
                        }
                    });
        };
    }

    @Bean
    public TopicExchange topicExchange() {
        TopicExchange sysExchange = new TopicExchange(SYS_EXCHANGE, true, false, new HashMap<>());

        return sysExchange;
    }

    @Bean
    public HeadersExchange delayExchange() {
        HashMap<String, Object> arguments = new HashMap<>();
        HeadersExchange sysDelayExchange =
                new HeadersExchange(SYS_DELAY_EXCHANGE, true, false, arguments);

        return sysDelayExchange;
    }

    @Bean
    public Declarables delayQueues() {
        List<Declarable> declarables = new ArrayList<>();

        for (String delayQueue : DELAY_QUEUES) {
            HashMap<String, Object> arguments = new HashMap<>();
            arguments.put("x-dead-letter-exchange", SYS_EXCHANGE);
            arguments.put("x-message-ttl", DurationStyle.detectAndParse(delayQueue).toMillis());
            declarables.add(new Queue("delayQueue." + delayQueue, true, false, false, arguments));

            HashMap<String, Object> bindArgs = new HashMap<>();
            bindArgs.put("x-match", "all");
            bindArgs.put("x-delay-queue-name", delayQueue);
            declarables.add(
                    new Binding(
                            "delayQueue." + delayQueue,
                            Binding.DestinationType.QUEUE,
                            SYS_DELAY_EXCHANGE,
                            "",
                            bindArgs));
        }

        return new Declarables(declarables);
    }
}
