diff --git a/coolamqp/argumentify.py b/coolamqp/argumentify.py
new file mode 100644
index 0000000000000000000000000000000000000000..13a10dfe46894f40a2f7d58358cdea4c30651168
--- /dev/null
+++ b/coolamqp/argumentify.py
@@ -0,0 +1,18 @@
+from coolamqp.framing.field_table import get_type_for
+from coolamqp.objects import tobytes
+
+
+def argumentify(arguments):
+ if arguments is None:
+ return []
+ args = []
+ if isinstance(arguments, dict):
+ for key, value in arguments.items():
+ key = tobytes(key)
+ args.append((key, (value, get_type_for(value))))
+ else:
+ for key, value in arguments:
+ key = tobytes(key)
+ args.append((key, (value, get_type_for(value))))
+
+ return args
\ No newline at end of file
diff --git a/coolamqp/attaches/consumer.py b/coolamqp/attaches/consumer.py
index 78077085670e3c300c4a9ef388e83de6b1fbe1e7..86938649b18b3132547c7a38c09d482bc1e0f2ed 100644
--- a/coolamqp/attaches/consumer.py
+++ b/coolamqp/attaches/consumer.py
@@ -413,7 +413,7 @@ class Consumer(Channeler):
QueueBind(
self.queue.name,
self.queue.exchange.name.encode('utf8'),
- b'', False, []),
+ b'', False, self.queue.arguments_bind),
QueueBindOk,
self.on_setup
)
diff --git a/coolamqp/framing/definitions.py b/coolamqp/framing/definitions.py
index 68278f4291aa0e4cf4f3554639a284ad0483d875..581b7c5a66deeb2723176b462caf202fff639037 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.argumentify 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/coolamqp/objects.py b/coolamqp/objects.py
index f1cf91999c650bee2cc642c26ee4a5a7e0df4b88..33ea5f94cce0c0ca401f6d50cb95c5be3f3b60ab 100644
--- a/coolamqp/objects.py
+++ b/coolamqp/objects.py
@@ -10,9 +10,9 @@ import warnings
import six
+from coolamqp.argumentify import argumentify
from coolamqp.framing.definitions import \
BasicContentPropertyList as MessageProperties
-from coolamqp.framing.field_table import get_type_for
logger = logging.getLogger(__name__)
@@ -32,22 +32,6 @@ def tobytes(q):
return q.encode('utf-8') if isinstance(q, six.text_type) else q
-def argumentify(arguments):
- if arguments is None:
- return None
- args = []
- if isinstance(arguments, dict):
- for key, value in arguments.items():
- key = tobytes(key)
- args.append((key, (value, get_type_for(value))))
- else:
- for key, value in arguments:
- key = tobytes(key)
- args.append((key, (value, get_type_for(value))))
-
- return args
-
-
class Callable(object):
"""
Add a bunch of callables to one list, and just invoke'm.
@@ -268,7 +252,9 @@ class Queue(object):
:param exchange: Exchange for this queue to bind to. None for no binding.
:param exclusive: This queue will be deleted when the connection closes
:param auto_delete: This queue will be deleted when the last consumer unsubscribes
- :param arguments: either a list of (bytes, values) or a dict of (str, value) to pass as an extra argument
+ :param arguments: either a list of (bytes, values) or a dict of (str, value) to pass as an extra argument during
+ declaration
+ :param arguments_bind: arguments to pass to binding to a (headers, I suppose exchange)
:raises ValueError: tried to create a queue that was not durable or auto_delete
:raises ValueError: tried to create a queue that was not exclusive or auto_delete and not anonymous
:raises ValueError: tried to create a queue that was anonymous and not auto_delete or durable
@@ -277,14 +263,15 @@ class Queue(object):
:warning UserWarning: if you're declaring an auto_delete or exclusive, anonymous queue
"""
__slots__ = ('name', 'durable', 'exchange', 'auto_delete', 'exclusive',
- 'anonymous', 'consumer_tag', 'arguments')
+ 'anonymous', 'consumer_tag', 'arguments', 'arguments_bind')
def __init__(self, name=None, # type: tp.Union[str, bytes, None]
durable=False, # type: bool
exchange=None, # type: tp.Optional[Exchange]
exclusive=True, # type: bool
auto_delete=True, # type: bool
- arguments=None # type: tp.Union[tp.List[bytes, tp.Any], tp.Dict[str, tp.Any]]
+ arguments=None, # type: tp.Union[tp.List[bytes, tp.Any], tp.Dict[str, tp.Any]],
+ arguments_bind=None,
):
if name is None:
self.name = None
@@ -296,6 +283,7 @@ class Queue(object):
self.auto_delete = auto_delete
self.exclusive = exclusive
self.arguments = argumentify(arguments)
+ self.arguments_bind = argumentify(arguments_bind)
self.anonymous = self.name is None # if this queue is anonymous, it must be regenerated upon reconnect
if self.auto_delete and self.durable:
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()
diff --git a/tests/test_framing.py b/tests/test_framing.py
index 3b4c10b2a1a2280dadbb16b896b8393e6cfe4769..5442da31c53daab3423eca2e52cd95eda886df72 100644
--- a/tests/test_framing.py
+++ b/tests/test_framing.py
@@ -3,7 +3,7 @@ import unittest
from coolamqp.framing.field_table import enframe_table
-from coolamqp.objects import argumentify
+from coolamqp.argumentify import argumentify
class TestFraming(unittest.TestCase):
diff --git a/tests/test_objects.py b/tests/test_objects.py
index 1415ba61ded72bd1fd99a0f2d8e591fadae7c949..9d37c62d2cedd4f257a95a6e5646483e6e1ae2e2 100644
--- a/tests/test_objects.py
+++ b/tests/test_objects.py
@@ -8,8 +8,8 @@ import warnings
from coolamqp.framing.definitions import QueueDeclare
-from coolamqp.objects import NodeDefinition, MessageProperties, Queue, argumentify
-
+from coolamqp.objects import NodeDefinition, MessageProperties, Queue
+from coolamqp.argumentify import argumentify
logger = logging.getLogger(__name__)
logging.getLogger('coolamqp').setLevel(logging.DEBUG)