moksha/api/hub/consumer.py | 36 +++++++++++++++++++++++++----------- 1 file changed, 25 insertions(+), 11 deletions(-)
New commits: commit 057ea1b0ef3457f3af04ce1a3b9ff6df86b6a4e6 Author: Luke Macken lmacken@redhat.com Date: Fri Jan 7 20:12:48 2011 -0500
Temporary hack around our leaking amqphub abstraction
diff --git a/moksha/api/hub/consumer.py b/moksha/api/hub/consumer.py index 83b528f..5ce870f 100644 --- a/moksha/api/hub/consumer.py +++ b/moksha/api/hub/consumer.py @@ -29,6 +29,7 @@ import logging log = logging.getLogger('moksha.hub')
from moksha.hub.hub import MokshaHub +from moksha.hub.amqp.pyamqplib import AMQPLibHub from moksha.lib.helpers import listify, create_app_engine, json from sqlalchemy.orm import sessionmaker
@@ -45,18 +46,31 @@ class Consumer(object): if self.hub.amqp_broker and not self.hub.stomp_broker: for topic in listify(self.topic): log.debug('Subscribing to consumer topic %s' % topic) - server_queue_name = 'moksha_consumer_' + self.hub.session.name - self.hub.queue_declare(queue=server_queue_name, exclusive=True) - self.hub.exchange_bind(server_queue_name, binding_key=topic) - local_queue_name = 'moksha_consumer_' + self.hub.session.name - self.hub.local_queue = self.hub.session.incoming(local_queue_name) - self.hub.message_subscribe(queue=server_queue_name, - destination=local_queue_name) - self.hub.local_queue.start() - if self.jsonify: - self.hub.local_queue.listen(self._consume_json) + + if isinstance(self.hub, AMQPLibHub): + # AMQPLibHub specific + import uuid + queue_name = str(uuid.uiud4()) + self.hub.queue_declare(queue=queue_name, exclusive=True) + self.hub.exchange_bind(queue_name, binding_key=topic) + if self.jsonify: + self.hub.subscribe(queue_name, self._consume_json) + else: + self.hub.subscribe(queue_name, self._consume) else: - self.hub.local_queue.listen(self._consume) + # Assume we're using Qpid then. + server_queue_name = 'moksha_consumer_' + self.hub.session.name + self.hub.queue_declare(queue=server_queue_name, exclusive=True) + self.hub.exchange_bind(server_queue_name, binding_key=topic) + local_queue_name = 'moksha_consumer_' + self.hub.session.name + self.hub.local_queue = self.hub.session.incoming(local_queue_name) + self.hub.message_subscribe(queue=server_queue_name, + destination=local_queue_name) + self.hub.local_queue.start() + if self.jsonify: + self.hub.local_queue.listen(self._consume_json) + else: + self.hub.local_queue.listen(self._consume)
# If the consumer specifies an 'app', then setup `self.engine` to # be a SQLAlchemy engine, along with a configured DBSession
moksha-commits@lists.fedorahosted.org