Merge pull request #36 from mokshaproject/feature/stomp-queues

Support STOMP consumers that want a queue, not a topic.

moksha.hub/moksha/hub/hub.py

@@ -17,6 +17,7 @@
 #          Ralph Bean  <rbean@redhat.com>
 
 
+import fnmatch
 import os
 import six
 import sys
@@ -206,8 +207,18 @@ def consume_stomp_message(self, message):
 
         # feed all of our consumers
         envelope = {'body': body, 'topic': topic, 'headers': headers}
-        for callback in self.topics.get(topic, []):
-            reactor.callInThread(callback, envelope)
+
+        # Some consumers subscribe to topics directly
+        for pattern, callbacks in self.topics.items():
+            if fnmatch.fnmatch(topic, pattern):
+                for callback in callbacks:
+                    reactor.callInThread(callback, envelope)
+
+        # Others subscribe to a queue composed of many topics..
+        subscription = headers.get('subscription')
+        if subscription != topic:
+            for callback in self.topics.get(subscription, []):
+                reactor.callInThread(callback, envelope)
 
 
 class CentralMokshaHub(MokshaHub):
@@ -377,8 +388,8 @@ def __init_consumers(self):
                     if topic not in self.topics:
                         self.topics[topic] = []
 
-                    if c.consume not in self.topics[topic]:
-                        self.topics[topic].append(c.consume)
+                    if c._consume not in self.topics[topic]:
+                        self.topics[topic].append(c._consume)
 
             except Exception as e:
                 log.exception("Failed to init %r consumer." % c_class)

moksha.hub/moksha/hub/stomp/stomp.py

@@ -55,6 +55,16 @@ def __init__(self, hub, config):
             host = self.config.get('stomp_broker')
             uri = "%s:%i" % (host, port)
 
+        # Sometimes, a stomp consumer may wish to be subscribed to a queue
+        # which is composed of messages from many different topics.  In this
+        # case, the hub hands dispatching messages to the right consumers.
+        # This extension is only concerned with the queue, and negotiating that
+        # with the broker.
+        stomp_queue = self.config.get('stomp_queue', None)
+        if stomp_queue:
+            # Overwrite the declarations of all of our consumers.
+            self._topics = [stomp_queue]
+
         # A list of addresses over which we emulate failover()
         self.addresses = [pair.split(":") for pair in uri.split(',')]
         self.address_index = 0