diff --git a/coolamqp/argumentify.py b/coolamqp/argumentify.py
new file mode 100644
index 0000000000000000000000000000000000000000..a8f0fd2b2af2cac689e279d09c563f696d128020
--- /dev/null
+++ b/coolamqp/argumentify.py
@@ -0,0 +1,31 @@
+import six
+
+from coolamqp.framing.field_table import get_type_for
+
+
+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
+
+
+def tobytes(q):
+ if isinstance(q, memoryview):
+ return q.tobytes()
+ return q.encode('utf-8') if isinstance(q, six.text_type) else q
+
+
+def toutf8(q):
+ if isinstance(q, memoryview):
+ q = q.tobytes()
+ return q.decode('utf-8') if isinstance(q, six.binary_type) else q
\ No newline at end of file
diff --git a/coolamqp/attaches/channeler.py b/coolamqp/attaches/channeler.py
index b22bb479ed7e1597bee406e3b2524593db5ee55e..a6ef07a6d94395265de06dc2371f06146acc9730 100644
--- a/coolamqp/attaches/channeler.py
+++ b/coolamqp/attaches/channeler.py
@@ -8,6 +8,7 @@ from __future__ import print_function, absolute_import, division
import logging
import typing as tp
+import coolamqp.argumentify
from coolamqp.framing.definitions import ChannelOpen, ChannelOpenOk, \
ChannelClose, ChannelCloseOk, BasicCancel, \
BasicCancelOk
@@ -161,8 +162,7 @@ class Channeler(Attache):
self.channel_id = None
if isinstance(payload, ChannelClose):
- logger.debug('Channel closed: %s %s', payload.reply_code,
- payload.reply_text.tobytes())
+ logger.debug('Channel closed: %s %s', payload.reply_code, payload.reply_text)
def methods(self, payloads):
# type: (tp.Iterable[coolamqp.framing.base.AMQPMethodPayload]) -> None
diff --git a/coolamqp/attaches/consumer.py b/coolamqp/attaches/consumer.py
index 816a9c5ea2ee9cd38e7fb082da6e211c871cdc0a..f5d6474e20d4b4d282a9b74f4f73b84cc44fd524 100644
--- a/coolamqp/attaches/consumer.py
+++ b/coolamqp/attaches/consumer.py
@@ -8,6 +8,7 @@ import uuid
import warnings
from concurrent.futures import Future
+import coolamqp.argumentify
from coolamqp.attaches.channeler import Channeler, ST_ONLINE, ST_OFFLINE
from coolamqp.exceptions import AMQPError
from coolamqp.framing.definitions import ChannelOpenOk, BasicConsume, \
@@ -395,7 +396,7 @@ class Consumer(Channeler):
elif isinstance(payload, QueueDeclareOk):
# did we need an anonymous name?
if not self.queue.name:
- self.queue.name = payload.queue.tobytes()
+ self.queue.name = self.queue.name.tobytes()
queue_declared = False
# We need any form of binding.
@@ -404,7 +405,7 @@ class Consumer(Channeler):
qb = QueueBind(
self.queue.name,
self.queue.exchange.name.encode('utf-8'),
- self.queue.routing_key, False, [])
+ self.queue.routing_key, False, self.queue.arguments_bind)
logger.debug('Running %s' % (repr(qb)))
self.method_and_watch(
qb,
diff --git a/coolamqp/framing/compilation/content_property.py b/coolamqp/framing/compilation/content_property.py
index 0149db264f76358c1ce10e8c1a2c95261a18abf8..cf836613362ffa988ae30ae205217fe55e889fbc 100644
--- a/coolamqp/framing/compilation/content_property.py
+++ b/coolamqp/framing/compilation/content_property.py
@@ -1,7 +1,7 @@
# coding=UTF-8
from __future__ import absolute_import, division, print_function
-from coolamqp.framing.field_table import deframe_table
+from coolamqp.framing.field_table import deframe_table, frame_table_size, enframe_table
"""Generate serializers/unserializers/length getters for given property_flags"""
import six
@@ -141,7 +141,6 @@ def _compile_particular_content_property_list_class(zpf, fields):
mod.append(get_counter(present_fields, prefix=u'self.', indent_level=2)[
:-1]) # skip eol
mod.append(u' + %s\n' % (zpf_length,)) # account for pf length
-
return u''.join(mod), structers
@@ -156,6 +155,8 @@ def compile_particular_content_property_list_class(zpf, fields):
locals_ = {
'AMQPContentPropertyList': AMQPContentPropertyList,
'deframe_table': deframe_table,
+ 'frame_table_size': frame_table_size,
+ 'enframe_table': enframe_table,
}
for structer in structers:
if structer not in STRUCTERS_FOR_NOW:
diff --git a/coolamqp/framing/definitions.py b/coolamqp/framing/definitions.py
index 68278f4291aa0e4cf4f3554639a284ad0483d875..f106e5bc3a01dbc2f35fea76d177110769e8ca40 100644
--- a/coolamqp/framing/definitions.py
+++ b/coolamqp/framing/definitions.py
@@ -1,5 +1,9 @@
# coding=UTF-8
from __future__ import print_function, absolute_import
+
+import coolamqp.argumentify
+from coolamqp.argumentify import argumentify
+
"""
A Python version of the AMQP machine-readable specification.
@@ -2941,6 +2945,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) |
@@ -3021,7 +3028,8 @@ class BasicContentPropertyList(AMQPContentPropertyList):
while buf[offset + pfl - 1] & 1:
pfl += 2
zpf = BasicContentPropertyList.zero_property_flags(buf[offset:offset +
- pfl]).tobytes()
+ pfl]).tobytes()
+
if zpf in BasicContentPropertyList.PARTICULAR_CLASSES:
return BasicContentPropertyList.PARTICULAR_CLASSES[
zpf].from_buffer(buf, offset)
diff --git a/coolamqp/objects.py b/coolamqp/objects.py
index 7647c66b9f77900110f65e61ca3825124210c912..3ab5a671592abd8ffdd4b5c308c4ccccc418c70e 100644
--- a/coolamqp/objects.py
+++ b/coolamqp/objects.py
@@ -10,44 +10,14 @@ import warnings
import six
-from coolamqp.framing.definitions import \
- BasicContentPropertyList as MessageProperties
-from coolamqp.framing.field_table import get_type_for
+from coolamqp.argumentify import argumentify, tobytes, toutf8
+from coolamqp.framing.definitions import BasicContentPropertyList as MessageProperties
logger = logging.getLogger(__name__)
EMPTY_PROPERTIES = MessageProperties()
-
-def toutf8(q):
- if isinstance(q, memoryview):
- q = q.tobytes()
- return q.decode('utf-8') if isinstance(q, six.binary_type) else q
-
-
-def tobytes(q):
- if isinstance(q, memoryview):
- return q.tobytes()
- 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.
@@ -271,6 +241,9 @@ class Queue(object):
:param arguments: either a list of (bytes, values) or a dict of (str, value) to pass as an extra argument
:param routing_key: routing key that will be used to bind to an exchange. Used only when this
queue is associated with an exchange. Default value of blank should suffice.
+ :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
@@ -279,7 +252,7 @@ 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', 'routing_key')
+ 'anonymous', 'consumer_tag', 'arguments', 'routing_key', 'arguments_bind')
def __init__(self, name=None, # type: tp.Union[str, bytes, None]
durable=False, # type: bool
@@ -287,7 +260,8 @@ class Queue(object):
exclusive=True, # type: bool
auto_delete=True, # type: bool
arguments=None, # type: tp.Union[tp.List[bytes, tp.Any], tp.Dict[str, tp.Any]],
- routing_key=b'' #: type: tp.Union[str, bytes]
+ routing_key=b'', #: type: tp.Union[str, bytes]
+ arguments_bind=None,
):
if name is None:
self.name = None
@@ -300,6 +274,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/coolamqp/uplink/connection/connection.py b/coolamqp/uplink/connection/connection.py
index 0974f5cf876993f70ea46bdd7f81961ed42c5043..cf0cce9ff980ab798fd692863600116b825a321e 100644
--- a/coolamqp/uplink/connection/connection.py
+++ b/coolamqp/uplink/connection/connection.py
@@ -8,6 +8,7 @@ import time
import typing as tp
import uuid
+import coolamqp.argumentify
from coolamqp.utils import monotonic
from coolamqp.exceptions import ConnectionDead
diff --git a/coolamqp/uplink/connection/recv_framer.py b/coolamqp/uplink/connection/recv_framer.py
index cd35ec6bf8f18382f477e280ecf9534c5d25cb1a..4f4545229dd8836a93d6b00942b33bc51b95e41a 100644
--- a/coolamqp/uplink/connection/recv_framer.py
+++ b/coolamqp/uplink/connection/recv_framer.py
@@ -7,6 +7,7 @@ import struct
import six
+import coolamqp.argumentify
from coolamqp.framing.definitions import FRAME_HEADER, FRAME_HEARTBEAT, \
FRAME_END, FRAME_METHOD, FRAME_BODY
from coolamqp.framing.frames import AMQPBodyFrame, AMQPHeaderFrame, \
diff --git a/coolamqp/uplink/handshake.py b/coolamqp/uplink/handshake.py
index 99d99bbfc7e2bfc67fcadd5c15b38b4dd59b858d..77d805a6d42aee240b7485149e3abeb4da1d6df6 100644
--- a/coolamqp/uplink/handshake.py
+++ b/coolamqp/uplink/handshake.py
@@ -1,6 +1,8 @@
# coding=UTF-8
from __future__ import absolute_import, division, print_function
+import coolamqp.argumentify
+
"""
Provides reactors that can authenticate an AQMP session
"""
diff --git a/tests/test_clustering/test_topic_exchanges.py b/tests/test_clustering/test_topic_exchanges.py
index dc8c18ea29c1178c54286560b3ac3f17ca55a175..b44efe7131cab8e8cba7921b3e3bd3bf76dcb0d9 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,33 @@ 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(uuid.uuid4().hex, exchange=exchange, arguments_bind={'x-match': 'all', 'location': 'brisbane'})
+ self.c.declare(queue1).result()
+ queue2 = Queue(uuid.uuid4().hex, exchange=exchange, arguments_bind={'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)
+ cons2, fut2 = self.c.consume(queue2, no_ack=False)
+ fut1.result()
+ fut2.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)