How to use listen on basic.return in python client of AMQP

Question

I'd like to make sure that my message was delivered to a queue.

To do so I'm adding the mandatory param to the basic_publish. What else should I do to receive the basic.return message if my message wasn't successfully delivered?

I can't use channel.wait() to listen for the basic.return because when my message is successfully delivered the wait() function hangs forever. (There is no timeout) On the other hand. When I don't call channel.wait() the channel.returned_messages will remain empty, even if the message isn't delivered.

I use py-amqplib version 0.6.

Any solution is welcome.

Answer 1

It is currently impossible as the basic.return is sent asynchronously when a message is dropped in broker. When message was sent successfully no data is reported from server. So pyAMQP can't listen for such messages.

I've read few threads about this problem. Possible solution were:

• use txAMQP, twisted version of amqp that handles basic.return
• use pyAMQP with wait with timeout. (I'm not sure if that is currently possible)
• ping server frequently with synchronous commands so that pyAMQP will able to pick basic.return messages when they arrive.

Because the level of support for pyAMQP and rabbitMQ in general is quite low, we decided not to use amqp broker at all.

Answer 2

Have you tried the only Python AMQP library that is complete? It isn't as widely used because it is not neatly packaged.

Step 1. compile the C library - you may need sudo apt-get install autotools-dev autoconf automake libtool

mkdir rabbitc
cd rabbitc
hg clone http://hg.rabbitmq.com/rabbitmq-codegen/
hg clone http://hg.rabbitmq.com/rabbitmq-c/
cd rabbitmq-c
autoreconf -i
make clean
./configure --prefix=/usr
make
sudo make install

Step 2. Install the Python library

pip install pylibrabbitmq

Answer 3

You can't do this synchronously as it is an asynchronous system. But you can solve this problem using threads.

The basic idea is that you start a thread which does the wait on the channel, whenever it comes out of the wait it calls the call_back function for any returned message in the returned message queue. You can then deal with that message however you want to in the call_back function

def registerCallback(channel, call_back):
    """ This method sets up a thread which deals with the asynchronous callback for a message which could not be routed by the exchange.
    """
    def wait():
        try:
            channel.wait()
        except Exception, e:
            print("Problem waiting on publish channel: %s" % str(e))

        while not channel.returned_messages.empty():
            returnedMessage = channel.returned_messages.get()
            processReturnedMessageThread = Thread(target=call_back, args=(returnedMessage))
            processReturnedMessageThread.start()

        wait()

    waiting = Thread(target=wait) 
    waiting.start()