in network, python, rabbitmq |

Default behaviour in implementation of STOMP protocol in RabbitMQ with python

Why STOMP?

Why STOMP, and not directly AMQP, as I’m using RabbitMQ. No real reason, but the fact that there are less dependencies on a STOMP client, as it’s just a socket with text sent.

Implementations

There are several implementations of the STOMP protocol for Python. The module I chose is python-stomp (version 0.2.9), from Benjamin W. Smith. It’s simple and easy to understand.

Simple Code Examples

sto_send.py:

from stompy.simple import Client

stomp = Client(host='rabbitmq2')
stomp.connect(username='guest',password='noneofyourbusiness')
stomp.put('Thomas est une b*te à Tetris...', destination='/queue/jeuvideo')
stomp.disconnect()

sto_receive.py:

from stompy.simple import Client

stomp = Client(host='rabbitmq2')
stomp.connect(username='guest',password='noneofyourbusiness')
stomp.subscribe('/queue/jeuvideo')
message = stomp.get()

print message.body

#stomp.ack(message)
stomp.unsubscribe('/queue/video')
stomp.disconnect()

Everything is working fine, when launching sto_receive.py, I receive the message. But when I launched several receivers, I noticed, that ONLY ONE programs received the message! After some research, I found the answer: As documented in the RabbitMQ wiki, the default exchange is ‘direct’:

[…]when messages leave a queue for a consumer, they are not duplicated. One message, sitting on a queue, is delivered to only one of the available consumers. […] If there are multiple clients, all SUBSCRIBEing to the same queue, then there will be multiple consumers all on the same queue, leading to round-robin delivery to those clients.

There is an explanation on how to change the behaviour, by changing the exchange type, and some of particular bits (like the id). I even found an example of modification for use in the equivalent STOMP Ruby module.

Here are the modifications. The good news is that there is no need to patch the stompy module, as the author provided the possibility to pass arbitrary parameters to the headers by the use of the ‘conf’ variable.

The important points are:

  • You need to define an exchange of type amq.topic
  • You need to set an id, which is different for each client
  • As you’re using topics, you’ll have to specify a routing_key

sto_receive.py:

from stompy.simple import Client
import uuid

unique_id = uuid.uuid4()

stomp = Client(host='rabbitmq2')
stomp.connect(username='guest',password='nonononono')

stomp.subscribe('',
                conf={'exchange': 'amq.topic',
                      'routing_key':'x.#',
                      'id': unique_id,
                      })

# Wait for a message to appear
while 1:
    message = stomp.get()
    print message.body

#stomp.ack(message)
stomp.unsubscribe('',conf={'id': unique_id})
stomp.disconnect()

sto_send.py:

from stompy.simple import Client



stomp = Client(host='rabbitmq2')
stomp.connect(username='guest',password='nonononono')


for i in range(10000):
 stomp.put('Thomas est une b*te au Tetris...', destination='x.y',
          conf={'exchange':'amq.topic',
                #  'routing_key':'x.y'
                })

stomp.disconnect()

Published