diff --git a/coolamqp/framing/definitions.py b/coolamqp/framing/definitions.py index 68278f4291aa0e4cf4f3554639a284ad0483d875..3909a159a8184be57552f94d42c325582a77120e 100644 --- a/coolamqp/framing/definitions.py +++ b/coolamqp/framing/definitions.py @@ -1,5 +1,8 @@ # coding=UTF-8 from __future__ import print_function, absolute_import + +from coolamqp.objects import argumentify + """ A Python version of the AMQP machine-readable specification. @@ -2941,6 +2944,9 @@ class BasicContentPropertyList(AMQPContentPropertyList): :param reserved: reserved, must be empty :type reserved: binary type (max length 255) (AMQP as shortstr) """ + if 'headers' in kwargs: + kwargs['headers'] = argumentify(kwargs['headers']) + zpf = bytearray([ (('content_type' in kwargs) << 7) | (('content_encoding' in kwargs) << 6) | diff --git a/tests/test_clustering/test_topic_exchanges.py b/tests/test_clustering/test_topic_exchanges.py index dc8c18ea29c1178c54286560b3ac3f17ca55a175..4d6554bb56dc7d00955604e96317f8554bb0d72a 100644 --- a/tests/test_clustering/test_topic_exchanges.py +++ b/tests/test_clustering/test_topic_exchanges.py @@ -3,11 +3,12 @@ import time import os import unittest import logging +import uuid import monotonic from coolamqp.clustering import Cluster -from coolamqp.objects import Exchange, Queue, NodeDefinition, Message +from coolamqp.objects import Exchange, Queue, NodeDefinition, Message, MessageProperties XCHG = Exchange('smok5.results', type='topic', durable=True) @@ -20,6 +21,7 @@ logging.getLogger('coolamqp').setLevel(logging.DEBUG) did_receive = False + class TestTopic(unittest.TestCase): def setUp(self): self.c = Cluster([NODE]) @@ -32,6 +34,31 @@ class TestTopic(unittest.TestCase): pass self.c.shutdown() + def test_headers_exchange(self): + xchg_name = uuid.uuid4().hex + exchange = Exchange(xchg_name, b'headers') + self.c.declare(exchange).result() + queue1 = Queue(exchange=exchange, arguments={'x-match': 'all', 'location': 'brisbane'}) + self.c.declare(queue1).result() + queue2 = Queue(exchange=exchange, arguments={'x-match': 'all', 'location': 'sydney'}) + self.c.declare(queue2).result() + + test = {'a': 0} + def do_message(msg): + msg.ack() + test['a'] += 1 + + cons1, fut1 = self.c.consume(queue1, no_ack=False).result() + cons2, fut2 = self.c.consume(queue2, no_ack=False).result() + + self.c.publish(Message(b'dupa', MessageProperties(headers={'location': 'sydney'})), exchange=exchange, confirm=True).result() + self.c.publish(Message(b'dupa', MessageProperties(headers={'location': 'brisbane'})), exchange=exchange, confirm=True).result() + self.c.publish(Message(b'dupa', MessageProperties(headers={'location': 'wtf'})), exchange=exchange, confirm=True).result() + time.sleep(1) + cons1.cancel().result() + cons2.cancel().result() + self.assertEqual(test['a'], 2) + def test_bind_stuff(self): self.c.declare(QUEUE) self.c.declare(XCHG).result()