How to consume RabbitMQ messages via pika for some limited time?

All the examples in pika tutorial end with the client invoking start_consuming(), which starts an infinite loop. These examples work for me.

However, I do not want my client to run forever. Instead, I need my client to consume messages for some time, such as 15 minutes, then stop.

How do I accomplish that?

Solution

You can consume messages one at a time with your own loops, say you have a channel and queue setup. The following will check if the queue is empty, and if not, pop a single message off of it.

queue_state = channel.queue_declare(queue, durable=True, passive=True)
queue_empty = queue_state.method.message_count == 0

declaring a queue that already exists, and setting the passive flags allows you to query it's state. Next we process a message:

if not queue_empty:
    method, properties, body = channel.basic_get(queue, no_ack=True)
    callback_func(channel, method, properties, body)

Here callback_func is our normal callback. Make sure not to register the callback with the queue when you want to process this way.

# DO NOT
channel.basic_consume(callback_func, queue, no_ack=True)

This will make the manual consume do weird things. I have seen the queue_declare code actually process a message if I have made this call beforehand.