openstack/oslo.messaging
Code Issues Proposed changes

Refactor processing reply in ReplyWaiter

Browse Source 
At the moment for each msg_id we receive two amqp message - first one
with the payload, a second one to ensure the other have finish to send
the payload. This was made, because a long time ago 'reply' allowed
generator as payload to send multiple messages on one 'rpc.call' - [1]

It's a bad idea - to double RPC messages for each call, so we are going
to remove this the second AMQP message sending. This patch allows
receiver side to proceed correctly old case - two AMQP messages (first
with data and second with 'ending' parameter) same as the new one (a
single message with data 'ending' parameter)

Blueprint: remove-double-reply

[1] - https://github.com/openstack/oslo-incubator/blob/stable/icehouse/openstack/common/rpc/amqp.py#L464

Change-Id: Ic09fe619694c300c4502acb7157d7ecdd47c5fd7
This commit is contained in:
Victor Sergeyev 2015-05-07 13:03:48 +03:00
parent cf98f07f1f
commit 75cba16d15
1 changed files with 10 additions and 6 deletions

16
oslo_messaging/_drivers/amqpdriver.py

@ -216,23 +216,24 @@ class ReplyWaiter(object):
_('Timed out waiting for a reply to message ID %s.') % msg_id)
def _process_reply(self, data):
result = None
ending = False
self.msg_id_cache.check_duplicate_message(data)
if data['failure']:
failure = data['failure']
result = rpc_common.deserialize_remote_exception(
failure, self.allowed_remote_exmods)
elif data.get('ending', False):
ending = True
else:
result = data['result']
result = data.get('result', None)
ending = data.get('ending', False)
return result, ending
def wait(self, msg_id, timeout):
# NOTE(sileht): for each msg_id we receive two amqp message
# first one with the payload, a second one to ensure the other
# have finish to send the payload
# NOTE(viktors): We are going to remove this behavior in the N
# release, but we need to keep backward compatibility, so we should
# support both cases for now.
timer = rpc_common.DecayingTimer(duration=timeout)
timer.start()
final_reply = None
@ -245,7 +246,10 @@ class ReplyWaiter(object):
self._raise_timeout_exception(msg_id)
reply, ending = self._process_reply(message)
if not ending:
if reply is not None:
# NOTE(viktors): This can be either first _send_reply() with an
# empty `result` field or a second _send_reply() with
# ending=True and no `result` field.
final_reply = reply
return final_reply