x/python-ganttclient
Code Issues Proposed changes

Send message ack in rpc.call and make queues durable.

Browse Source
This commit is contained in:
Joshua McKenty
2010-07-19 03:38:46 +00:00
committed by Tarmac
parent 2d98b30f2a b2321fa4ed
commit b1ee82d905
1 changed files with 7 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
nova/rpc.py
View File

@@ -110,6 +110,7 @@ class TopicConsumer(Consumer):
self.queue = topic self.queue = topic
self.routing_key = topic self.routing_key = topic
self.exchange = FLAGS.control_exchange self.exchange = FLAGS.control_exchange
self.durable = False
super(TopicConsumer, self).__init__(connection=connection) super(TopicConsumer, self).__init__(connection=connection)
@@ -195,7 +196,10 @@ def call(topic, msg):
conn = Connection.instance() conn = Connection.instance()
d = defer.Deferred() d = defer.Deferred()
consumer = DirectConsumer(connection=conn, msg_id=msg_id) consumer = DirectConsumer(connection=conn, msg_id=msg_id)
consumer.register_callback(lambda data, message: d.callback(data)) def deferred_receive(data, message):
message.ack()
d.callback(data)
consumer.register_callback(deferred_receive)
injected = consumer.attach_to_tornado() injected = consumer.attach_to_tornado()
# clean up after the injected listened and return x # clean up after the injected listened and return x
@@ -233,7 +237,8 @@ def send_message(topic, message, wait=True):
exchange=msg_id, exchange=msg_id,
auto_delete=True, auto_delete=True,
exchange_type="direct", exchange_type="direct",
routing_key=msg_id) routing_key=msg_id,
durable=False)
consumer.register_callback(generic_response) consumer.register_callback(generic_response)
publisher = messaging.Publisher(connection=Connection.instance(), publisher = messaging.Publisher(connection=Connection.instance(),