SpringAMQP - Retry/Resend messages dlx

Question

I'm trying to use a retry mechanism using DLX. So, basically I want to send an message for 3 times and than stop and keep this message stopped on dlx queue;

What I did: Created WorkQueue bound to WorkExchange Created RetryQueue bound to RetryExchange

WorkQueue -> set x-dead-letter-exchange to RetryExchange

RetryQueue -> set x-dead-letter-exchange to WorkExchange AND x-message-ttl to 300000 ms (5 minutes)

So, now when I send any message to WorkQueue and it fail.. this message goes to RetryQueue for 5min and than back to WorkQueue.. but it can keep failing and I would do like to stop it after 3 attemps ...

It is possible? Is possible set to RetryQueue try to 3 times and after stop?

thanks.

Answer 1

There is no way to do this in the broker alone.

You can add code to your listener - examine the x-death header to determine how many times the message has been retried and discard/log it (and/or send it to a third queue) in your listener when you want to give up.

EDIT

@SpringBootApplication
public class So59741067Application {

    public static void main(String[] args) {
        SpringApplication.run(So59741067Application.class, args);
    }

    @Bean
    public Queue main() {
        return QueueBuilder.durable("mainQueue")
                .deadLetterExchange("")
                .deadLetterRoutingKey("dlQueue")
                .build();
    }

    @Bean
    public Queue dlq() {
        return QueueBuilder.durable("dlQueue")
                .deadLetterExchange("")
                .deadLetterRoutingKey("mainQueue")
                .ttl(5_000)
                .build();
    }

    @RabbitListener(queues = "mainQueue")
    public void listen(String in,
            @Header(name = "x-death", required = false) List<Map<String, ?>> xDeath) {

        System.out.println(in + xDeath);
        if (xDeath != null && (long) xDeath.get(0).get("count") > 2L) {
            System.out.println("Given up on this one");
        }
        else {
            throw new AmqpRejectAndDontRequeueException("test");
        }
    }

}