Delayed messages loop with RabbitMQ

Question

I'm trying to achieve a reject/delay loop using Rabbit's operations, i.e. :

I Have:

Main Queue with Main Exchange binded to it and DLX to StandBy Exchange.
StandBy Queue with StandBy Exchange binded to it with 60s TTL and DLX to Main Exchange

Basically I want to:

Consume from Main Queue
Rejects message (under certain circunstances)
Will get redirect it to StandBy Queue because rejection
When TTL expire, re-queue message to Main Queue.

The steps 1, 2 and 3 are OK but the last one drop the message instead of re-queue it.

Some theory from RabbitMQ's docs what I used to design this was:

Messages from a queue can be 'dead-lettered'; that is, republished to another exchange when any of the following events occur:

The message is rejected (basic.reject or basic.nack) with requeue=false,

The TTL for the message expires; or

The queue length limit is exceeded.

...

It is possible to form a cycle of message dead-lettering. For instance, this can happen when a queue dead-letters messages to the default exchange without specifiying a dead-letter routing key. Messages in such cycles (i.e. messages that reach the same queue twice) will be dropped if there was no rejections in the entire cycle.

The theory says that it should be re-queue because it has a rejection in the cycle from step #2, so, can you help me figure it out why it drops the message instead of re-queue it?

UPDATE:

The version I was targeting was 2.8.4 and it seems that in that moment the if there was no rejections in the entire cycle wasn't in the uses cases, anyway you can check this yourselves RabbitMQ 2.8.x Docs

I'll accept @george answer as the original objective can be achieved by this code.

George Octavian Rabanca George Octavian Rabanca · Accepted Answer · 2016-12-19T04:16:08

Rafael, I am not sure what client you are using but with the Pika client in Python you could implement something like this. For simplicity I only use one exchange. Are you sure you are setting the exchange and the routing-key properly?

sender.py

import sys
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
               'localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='cycle', type='direct')
channel.queue_declare(queue='standby_queue',
                      arguments={
                          'x-message-ttl': 10000,
                          'x-dead-letter-exchange': 'cycle',
                          'x-dead-letter-routing-key': 'main_queue'})
channel.queue_declare(queue='main_queue',
                      arguments={
                          'x-dead-letter-exchange': 'cycle',
                          'x-dead-letter-routing-key': 'standby_queue'})
channel.queue_bind(queue='main_queue', exchange='cycle')
channel.queue_bind(queue='standby_queue', exchange='cycle')
channel.basic_publish(exchange='cycle',
                      routing_key='main_queue',
                      body="message body")
connection.close()

receiver.py

import sys
import pika
def callback(ch, method, properties, body):
    print "Processing message: {}".format(body)
    # replace with condition for rejection
    if True:
        print "Rejecting message"
        ch.basic_nack(method.delivery_tag, False, False)
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.basic_consume(callback, queue='main_queue')
channel.start_consuming()

Delayed messages loop with RabbitMQ

1 Answers