openstack
/
oslo.messaging
Code Issues Proposed changes

Properly deserialize received AMQP 1.0 messages 

Change-Id: I5c27c06ca4f5b90e5c54a9519b70f9be0232abff
Closes-bug: #1438742
This commit is contained in:
Kenneth Giusti 2015-03-31 11:23:28 -04:00
parent 711f4187d4
commit 6f4e32ad53
2 changed files with 19 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
oslo_messaging/_drivers/protocols/amqp/driver.py
View File

@ -142,7 +142,8 @@ def marshal_request(request, context, envelope):
def unmarshal_request(message):
data = jsonutils.loads(message.body)
return (data.get("request"), data.get("context"))
msg = common.deserialize_msg(data.get("request"))
return (msg, data.get("context"))
class ProtonIncomingMessage(base.IncomingMessage):

27
oslo_messaging/tests/test_amqp_driver.py
View File

@ -260,23 +260,30 @@ class TestAmqpNotification(_AmqpBrokerTestCase):
(oslo_messaging.Target(topic="topic-2"), 'debug')]
nl = driver.listen_for_notifications(notifications, None)
listener = _ListenerThread(nl, 3)
# send one for each support version:
msg_count = len(notifications) * 2
listener = _ListenerThread(nl, msg_count)
targets = ['topic-1.info',
'topic-1.bad', # should be dropped
'bad-topic.debug', # should be dropped
'topic-1.error', 'topic-2.debug']
'topic-1.error',
'topic-2.debug']
for version in (1.0, 2.0):
for t in targets:
driver.send_notification(oslo_messaging.Target(topic=t),
"context", {'target': t},
version)
for t in targets:
driver.send_notification(oslo_messaging.Target(topic=t),
"context", {'target': t},
1.0)
listener.join(timeout=30)
self.assertFalse(listener.isAlive())
topics = [x.message.get('target') for x in listener.get_messages()]
self.assertTrue('topic-1.info' in topics)
self.assertTrue('topic-1.error' in topics)
self.assertTrue('topic-2.debug' in topics)
self.assertEqual(self._broker.dropped_count, 2)
self.assertEqual(len(topics), msg_count)
self.assertEqual(topics.count('topic-1.info'), 2)
self.assertEqual(topics.count('topic-1.error'), 2)
self.assertEqual(topics.count('topic-2.debug'), 2)
self.assertEqual(self._broker.dropped_count, 4)
driver.cleanup()