Spring AMQP & RabbitMQ

@Configuration
public class RabbitMQConfig {

    public static final String ORDER_EXCHANGE = "order-exchange";
    public static final String ORDER_QUEUE = "order-processing";
    public static final String ORDER_ROUTING_KEY = "order.created";

    @Bean
    public DirectExchange orderExchange() {
        return ExchangeBuilder.directExchange(ORDER_EXCHANGE)
            .durable(true)
            .build();
    }

    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable(ORDER_QUEUE)
            .withArgument("x-dead-letter-exchange", "order-dlx")
            .withArgument("x-dead-letter-routing-key", "order.dead")
            .withArgument("x-max-length", 50000)
            .build();
    }

    @Bean
    public Binding orderBinding(Queue orderQueue, DirectExchange orderExchange) {
        return BindingBuilder
            .bind(orderQueue)
            .to(orderExchange)
            .with(ORDER_ROUTING_KEY);
    }

    // Topic exchange example
    @Bean
    public TopicExchange eventExchange() {
        return ExchangeBuilder.topicExchange("event-exchange")
            .durable(true)
            .build();
    }

    @Bean
    public Queue allOrderEventsQueue() {
        return QueueBuilder.durable("order-all-events").build();
    }

    @Bean
    public Binding allOrderEventsBinding() {
        return BindingBuilder
            .bind(allOrderEventsQueue())
            .to(eventExchange())
            .with("order.#");
    }
}

@Service
@RequiredArgsConstructor
public class OrderEventPublisher {

    private final RabbitTemplate rabbitTemplate;

    // Simple send — uses configured MessageConverter
    public void publishOrderCreated(OrderEvent event) {
        rabbitTemplate.convertAndSend(
            "order-exchange",      // exchange
            "order.created",       // routing key
            event                  // payload (auto-converted to JSON)
        );
    }

    // Send with message post-processor (add headers)
    public void publishWithHeaders(OrderEvent event) {
        rabbitTemplate.convertAndSend("order-exchange", "order.created", event,
            message -> {
                message.getMessageProperties().setHeader("event-type", "ORDER_CREATED");
                message.getMessageProperties().setCorrelationId(event.getOrderId());
                message.getMessageProperties().setExpiration("60000"); // TTL 60s
                return message;
            });
    }

    // Send raw Message object
    public void publishRawMessage(OrderEvent event) {
        MessageProperties props = new MessageProperties();
        props.setContentType(MessageProperties.CONTENT_TYPE_JSON);
        props.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        props.setPriority(5);

        Message message = rabbitTemplate.getMessageConverter()
            .toMessage(event, props);
        rabbitTemplate.send("order-exchange", "order.created", message);
    }
}

// Synchronous RPC — sends request, blocks for reply
public OrderValidationResult validateOrder(OrderRequest request) {
    Object response = rabbitTemplate.convertSendAndReceive(
        "validation-exchange",
        "order.validate",
        request
    );
    return (OrderValidationResult) response;
}

// Async RPC with AsyncRabbitTemplate
@Bean
public AsyncRabbitTemplate asyncRabbitTemplate(RabbitTemplate rabbitTemplate) {
    return new AsyncRabbitTemplate(rabbitTemplate);
}

public CompletableFuture<OrderValidationResult> validateOrderAsync(OrderRequest req) {
    RabbitConverterFuture<OrderValidationResult> future =
        asyncRabbitTemplate.convertSendAndReceive(
            "validation-exchange", "order.validate", req);
    return future.toCompletableFuture();
}

@Bean
public MessageConverter messageConverter() {
    Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
    // Configure ObjectMapper if needed
    converter.setCreateMessageIds(true);
    return converter;
}

@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory,
                                     MessageConverter messageConverter) {
    RabbitTemplate template = new RabbitTemplate(connectionFactory);
    template.setMessageConverter(messageConverter);
    template.setExchange("order-exchange");
    template.setRoutingKey("order.created");
    return template;
}

@Component
@Slf4j
public class OrderEventConsumer {

    // Simple — auto-acknowledge on return
    @RabbitListener(queues = "order-processing")
    public void handleOrderCreated(OrderEvent event) {
        log.info("Received order: {}", event.getOrderId());
        orderService.processOrder(event);
    }

    // Access full Message + Channel for manual ack
    @RabbitListener(queues = "order-processing")
    public void handleWithManualAck(OrderEvent event, Channel channel,
                                     @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        try {
            orderService.processOrder(event);
            channel.basicAck(tag, false);  // false = only this message
        } catch (Exception e) {
            channel.basicNack(tag, false, true);  // requeue
        }
    }

    // Multiple queues
    @RabbitListener(queues = {"order-processing", "order-priority"})
    public void handleMultipleQueues(OrderEvent event) {
        orderService.processOrder(event);
    }
}

@Configuration
public class ListenerConfig {

    @Bean
    public SimpleRabbitListenerContainerFactory highThroughputFactory(
            ConnectionFactory connectionFactory,
            MessageConverter messageConverter) {
        SimpleRabbitListenerContainerFactory factory =
            new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(messageConverter);
        factory.setConcurrentConsumers(5);
        factory.setMaxConcurrentConsumers(20);
        factory.setPrefetchCount(25);
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return factory;
    }
}

@RabbitListener(
    queues = "high-volume-queue",
    containerFactory = "highThroughputFactory",
    concurrency = "5-20"  // min 5, max 20 consumer threads
)
public void handleHighVolume(OrderEvent event, Channel channel,
                              @Header(AmqpHeaders.DELIVERY_TAG) long tag)
        throws IOException {
    orderService.processOrder(event);
    channel.basicAck(tag, false);
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(value = "notification-queue", durable = "true"),
    exchange = @Exchange(value = "event-exchange", type = ExchangeTypes.TOPIC),
    key = "notification.#"
))
public void handleNotification(NotificationEvent event) {
    notificationService.send(event);
}

@Configuration
public class RetryConfig {

    // Main queue — points to DLX on failure
    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable("order-queue")
            .withArgument("x-dead-letter-exchange", "order-retry-exchange")
            .withArgument("x-dead-letter-routing-key", "order.retry")
            .build();
    }

    // Wait queue — messages sit here for TTL then route back
    @Bean
    public Queue orderRetryQueue() {
        return QueueBuilder.durable("order-retry-queue")
            .withArgument("x-dead-letter-exchange", "order-exchange")
            .withArgument("x-dead-letter-routing-key", "order.created")
            .withArgument("x-message-ttl", 30000)  // 30 seconds wait
            .build();
    }

    // Final dead letter queue — after max retries
    @Bean
    public Queue orderDeadLetterQueue() {
        return QueueBuilder.durable("order-dead-letter").build();
    }

    @Bean
    public DirectExchange retryExchange() {
        return new DirectExchange("order-retry-exchange");
    }

    @Bean
    public Binding retryBinding() {
        return BindingBuilder
            .bind(orderRetryQueue())
            .to(retryExchange())
            .with("order.retry");
    }
}

@RabbitListener(queues = "order-queue")
public void handleOrder(OrderEvent event, Channel channel,
                         @Header(AmqpHeaders.DELIVERY_TAG) long tag,
                         @Header(name = "x-retry-count",
                                 required = false,
                                 defaultValue = "0") int retryCount)
        throws IOException {
    try {
        orderService.processOrder(event);
        channel.basicAck(tag, false);
    } catch (TransientException e) {
        if (retryCount < 3) {
            // Send to retry queue with incremented count
            rabbitTemplate.convertAndSend("order-retry-exchange", "order.retry", event,
                msg -> {
                    msg.getMessageProperties().setHeader("x-retry-count", retryCount + 1);
                    return msg;
                });
            channel.basicAck(tag, false);  // ack original
        } else {
            // Max retries exceeded — send to dead letter
            channel.basicNack(tag, false, false);  // don't requeue
        }
    }
}

@Configuration
public class PublisherConfirmConfig {

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory,
                                         MessageConverter messageConverter) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMessageConverter(messageConverter);
        template.setMandatory(true);  // enable returns for unroutable messages

        // Confirm callback — broker ACK/NACK
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.debug("Message confirmed: {}", correlationData);
            } else {
                log.error("Message NACK'd: {} reason: {}", correlationData, cause);
                // Trigger retry or alert
            }
        });

        // Return callback — message could not be routed
        template.setReturnsCallback(returned -> {
            log.error("Message returned: exchange={}, routingKey={}, replyCode={}, replyText={}",
                returned.getExchange(),
                returned.getRoutingKey(),
                returned.getReplyCode(),
                returned.getReplyText());
            // Store for retry or send to fallback queue
        });

        return template;
    }
}

public void publishWithConfirm(OrderEvent event) {
    CorrelationData correlationData = new CorrelationData(event.getOrderId());

    rabbitTemplate.convertAndSend("order-exchange", "order.created",
        event, correlationData);

    // Optionally wait for confirm (blocking)
    try {
        CorrelationData.Confirm confirm =
            correlationData.getFuture().get(5, TimeUnit.SECONDS);
        if (!confirm.isAck()) {
            log.error("Publish not confirmed for order: {}", event.getOrderId());
            // Handle failure
        }
    } catch (TimeoutException e) {
        log.error("Confirm timeout for order: {}", event.getOrderId());
    }
}

@Bean
public SimpleRabbitListenerContainerFactory retryContainerFactory(
        ConnectionFactory connectionFactory,
        MessageConverter messageConverter) {
    SimpleRabbitListenerContainerFactory factory =
        new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setMessageConverter(messageConverter);
    factory.setAdviceChain(retryInterceptor());
    return factory;
}

@Bean
public RetryOperationsInterceptor retryInterceptor() {
    return RetryInterceptorBuilder.stateless()
        .maxAttempts(3)
        .backOffOptions(1000, 2.0, 10000)  // initial, multiplier, max
        .recoverer(republishMessageRecoverer())
        .build();
}

@Bean
public RepublishMessageRecoverer republishMessageRecoverer() {
    return new RepublishMessageRecoverer(
        rabbitTemplate,
        "error-exchange",        // target exchange
        "error.order-processing" // routing key
    );
}

@Bean
public SimpleRabbitListenerContainerFactory customFactory(
        ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory =
        new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setErrorHandler(new ConditionalRejectingErrorHandler(
        new CustomFatalExceptionStrategy()));
    return factory;
}

public class CustomFatalExceptionStrategy
        extends ConditionalRejectingErrorHandler.DefaultExceptionStrategy {

    @Override
    public boolean isFatal(Throwable t) {
        // Don't retry these — they will never succeed
        return t.getCause() instanceof JsonParseException
            || t.getCause() instanceof ValidationException;
    }
}

@Configuration
public class StreamConfig {

    // Consumer — receives from "order-events" topic
    @Bean
    public Function<OrderEvent, ProcessedOrderEvent> orderProcessor() {
        return event -> {
            log.info("Processing order: {}", event.getOrderId());
            // transform and forward to output binding
            return new ProcessedOrderEvent(event.getOrderId(), "PROCESSED");
        };
    }

    // Consumer only (no output)
    @Bean
    public Consumer<NotificationEvent> notificationConsumer() {
        return event -> notificationService.send(event);
    }

    // Supplier — produces messages on a schedule
    @Bean
    public Supplier<Flux<HeartbeatEvent>> heartbeatProducer() {
        return () -> Flux.interval(Duration.ofSeconds(30))
            .map(i -> new HeartbeatEvent("order-service", Instant.now()));
    }
}

@SpringBootTest
@RabbitListenerTest(capture = true)
class OrderEventConsumerTest {

    @Autowired
    private RabbitListenerTestHarness harness;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    void shouldProcessOrderEvent() throws Exception {
        OrderEvent event = new OrderEvent("ORD-123", "CREATED");

        rabbitTemplate.convertAndSend("order-exchange", "order.created", event);

        // Get the invocation result from the listener
        InvocationData invocation = harness
            .getNextInvocationDataFor("orderListener", 5, TimeUnit.SECONDS);

        assertThat(invocation).isNotNull();
        assertThat(invocation.getArguments()[0]).isEqualTo(event);
    }
}

// The listener must have an id:
@RabbitListener(id = "orderListener", queues = "order-processing")
public void handleOrder(OrderEvent event) { ... }

@SpringBootTest
@Testcontainers
class RabbitMQIntegrationTest {

    @Container
    static RabbitMQContainer rabbit = new RabbitMQContainer("rabbitmq:3.12-management")
        .withExposedPorts(5672, 15672);

    @DynamicPropertySource
    static void configure(DynamicPropertyRegistry registry) {
        registry.add("spring.rabbitmq.host", rabbit::getHost);
        registry.add("spring.rabbitmq.port", rabbit::getAmqpPort);
        registry.add("spring.rabbitmq.username", () -> "guest");
        registry.add("spring.rabbitmq.password", () -> "guest");
    }

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private OrderRepository orderRepository;

    @Test
    void endToEnd_orderCreatedEvent_persistsOrder() {
        OrderEvent event = new OrderEvent("ORD-456", "CREATED");

        rabbitTemplate.convertAndSend("order-exchange", "order.created", event);

        // Wait for async consumer to process
        await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
            Optional<Order> order = orderRepository.findById("ORD-456");
            assertThat(order).isPresent();
            assertThat(order.get().getStatus()).isEqualTo("PROCESSED");
        });
    }

    @Test
    void poisonMessage_routedToDeadLetter() {
        // Send invalid payload
        rabbitTemplate.convertAndSend("order-exchange", "order.created",
            "not-a-valid-json-object");

        await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> {
            Message dlqMessage = rabbitTemplate.receive("order-dead-letter", 1000);
            assertThat(dlqMessage).isNotNull();
            assertThat(dlqMessage.getMessageProperties().getHeaders())
                .containsKey("x-exception-message");
        });
    }
}