Rabbitmq Configuration for Spring Boot <3

RabbitmqConfiguration.java

import lombok.extern.slf4j.Slf4j;
import net.aalkhodiry.api.configurations.queues.CustomDLQueueDLQueue;
import net.aalkhodiry.api.utils.CustomJsonMessageConverter;
import org.aopalliance.intercept.Interceptor;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler;

/**
 * For RabbitMq
 */
@Configuration
@EnableRabbit
@Slf4j
public class RabbitmqConfiguration {

    @Autowired
    private CachingConnectionFactory cachingConnectionFactory;

    /**
     * To configure @RabbitListener
     *
     * @return
     */
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConcurrentTaskScheduler taskScheduler) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setTaskExecutor(taskScheduler);
        factory.setConnectionFactory(cachingConnectionFactory);
        factory.setMessageConverter(jsonMessageConverter());
        factory.setConcurrentConsumers(2);
        factory.setMaxConcurrentConsumers(6);
        factory.setTransactionManager(rabbitTransactionManager());

        // use a non-transactional template for the DLQ
        RabbitTemplate dlqTemplate = new RabbitTemplate(this.cachingConnectionFactory);
        // When retries reach the maximum number that massage is dead-lettered to jz-exceptions-exchange
        Interceptor retryInterceptor = RetryInterceptorBuilder.stateless()
                .maxAttempts(5)
                .recoverer(new RepublishMessageRecoverer(dlqTemplate, CustomDLQueue.EXCEPTION_EXCHANGE))
                .build();

        factory.setAdviceChain(retryInterceptor);
        return factory;
    }

    public RabbitTransactionManager rabbitTransactionManager() {
        return new RabbitTransactionManager(cachingConnectionFactory);
    }

    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(cachingConnectionFactory);
        template.setMessageConverter(jsonMessageConverter());
        return template;
    }

    /**
     * Required for executing administration functions against an AMQP Broker
     */
    @Bean
    public AmqpAdmin rabbitAdmin() {
        return new RabbitAdmin(cachingConnectionFactory);
    }

    @Bean
    public MessageConverter jsonMessageConverter() {
        final CustomJsonMessageConverter customJsonMessageConverter = new CustomJsonMessageConverter();
        return customJsonMessageConverter;
    }
}