Wait for a single RabbitMQ message with a timeout

Question

I'd like to send a message to a RabbitMQ server and then wait for a reply message (on a "reply-to" queue). Of course, I don't want to wait forever in case the application processing these messages is down - there needs to be a timeout. It sounds like a very basic task, yet I can't find a way to do this. I've now run into this problem with both py-amqplib and the RabbitMQ .NET client.

The best solution I've got so far is to poll using basic_get with sleep in-between, but this is pretty ugly:

def _wait_for_message_with_timeout(channel, queue_name, timeout):
    slept = 0
    sleep_interval = 0.1

    while slept < timeout:
        reply = channel.basic_get(queue_name)
        if reply is not None:
            return reply

        time.sleep(sleep_interval)
        slept += sleep_interval

    raise Exception('Timeout (%g seconds) expired while waiting for an MQ response.' % timeout)

Surely there is some better way?

Answer 1

This seems to break the whole idea of asynchronous processing, but if you must I think the right way to do it is to use an RpcClient.

Answer 2

There's an example here using qpid with a msg = q.get(timeout=1) that should do what you want. Sorry, I don't know what other AMQP client libraries implement timeouts (and in particular I don't know the two specific ones you mentioned).

Answer 3

Here's what I ended up doing in the .NET client:

protected byte[] WaitForMessageWithTimeout(string queueName, int timeoutMs)
{
    var consumer = new QueueingBasicConsumer(Channel);
    var tag = Channel.BasicConsume(queueName, true, null, consumer);
    try
    {
        object result;
        if (!consumer.Queue.Dequeue(timeoutMs, out result))
            throw new ApplicationException(string.Format("Timeout ({0} seconds) expired while waiting for an MQ response.", timeoutMs / 1000.0));

        return ((BasicDeliverEventArgs)result).Body;
    }
    finally
    {
        Channel.BasicCancel(tag);
    }
}

Unfortunately, I cannot do the same with py-amqplib, because its basic_consume method does not call the callback unless you call channel.wait() and channel.wait() doesn't support timeouts! This silly limitation (which I keep running into) means that if you never receive another message your thread is frozen forever.

Answer 4

I just added timeout support for amqplib in carrot.

This is a subclass of amqplib.client0_8.Connection:

http://github.com/ask/carrot/blob/master/carrot/backends/pyamqplib.py#L19-97

wait_multi is a version of channel.wait able to receive on an arbitrary number of channels.

I guess this could be merged upstream at some point.