[rabbitmq-discuss] STOMP plugin, changes, usage example in python

Thu Dec 16 18:58:12 GMT 2010

Greetings,

I have been using the STOMP gateway plugin for RabbitMQ 2.1.1 for some time, and it works great. The only hurdles I had in the past was a lack of documentation and usage examples. I believe part of the reason is the volatility of the protocol specification itself..?

That being said, the most recent upgrade I preformed borked my implementation. Of course this was just testing and thank goodness this update was not preformed on my production environment. 

I am currently using the stomp python script from codehaus. My python source for the wrapper is at the bottom of this email.

The code runs as is 2.1.1, and will receive broadcast style messages. The code does not function correctly against the 2.2.0 release, I receive a:

ListenerException: Error received headers:{'message': 'Unknown destination', 'content-type': 'text/plain', 'content-length': '86'}, message:'' is not a valid destination.
Valid exchange types are: /exchange, /topic or /queue.

To remedy the error I changed my subscription line to include '/topic' as the destination. 

My only concern is that this may change again in the future. Why did this change? 

Thanks,
Matt 

""" Heleper function for listening to an update via STOMP
"""
import json
import stomp
import logging

from uuid import uuid1

LOG_FILENAME = 'stomp.log'
logging.basicConfig(filename=LOG_FILENAME,level=logging.DEBUG)

stomp_logger = logging.getLogger('stomp')

class ListenerException(Exception):
    pass

class MsgReceiver(stomp.ConnectionListener):
    def __init__(self, callback=None):
        if callback is not None:
            self.callback = callback
    def on_error(self, headers, message):
        raise ListenerException("Error received headers:%s, message:%s " % (headers, message))
    def on_message(self, headers, message):
        #self.callback(headers, message) 
        stomp_logger.info("Message received: (headers:%s) %s" % (headers, message))

class Listener (object):

    host = "localhost"
    port = 61613
    user = "guest"
    passw = "guest"

    def __init__(self, callback=None, *args, **kwargs):

        if hasattr(self, 'conn'):
            if self.conn.is_connected():
                self.conn.disconnect()
        if not callback:
            raise Exception("Please specify a callback function")

        self.host = kwargs.get('host') if kwargs.get('host') else self.host
        self.user = kwargs.get('user') if kwargs.get('user') else self.user
        self.passw = kwargs.get('passw') if kwargs.get('passw') else self.passw
        self.port = kwargs.get('port') if kwargs.get('port') else self.port

        # build the stomp connection XXX: Add error handling 
        self.conn = stomp.Connection(
                [(self.host, self.port)], 
                self.user, self.passw
                )

        self.conn.set_listener('callback-%s' % uuid1(), MsgReceiver(callback))
        self.conn.start()
        self.conn.connect()

    def add_thing(self):
        if self.conn.is_connected():
            self.conn.subscribe(
                    destination="",
                    exchange="amq.topic", 
                    routing_key="#")

    def close(self):
        if self.conn.is_connected():
            self.conn.disconnect()