diff --git a/CHANGELOG.md b/CHANGELOG.md
index 966126608bf9dbcdc79f98277cba4d1127722a3f..24a56d88173ee2b198012684ea3494de8f55d144 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,18 @@
Previous release notes are hosted on [GitHub](https://github.com/smok-serwis/coolamqp/releases).
Since v1.3.2 they'll be put here and in release description.
+# v2.0.0
+========
+
+* changes to Queues:
+ * anonymous queues are back, for usage refer [here](https://smokserwis.docs.smok.co/coolamqp/advanced.html)
+ * changed some default arguments for Queues for them to better make sense
+ * some argument combinations just raise ValueError
+ * PendingDeprecationWarning changed into a DeprecationWarning
+* changes to Cluster:
+ * declare will refuse to declare an anonymous queue
+ * removed publish(tx)
+
# v1.5.0
========
diff --git a/coolamqp/clustering/cluster.py b/coolamqp/clustering/cluster.py
index 82c729d490e9864996ffc277044833ee7187132d..d8eeb8b65ba9ffa99d3d86bc17db4c768006e9ab 100644
--- a/coolamqp/clustering/cluster.py
+++ b/coolamqp/clustering/cluster.py
@@ -129,7 +129,10 @@ class Cluster(object):
:param span: optional parent span, if opentracing is installed
:param dont_trace: if True, a span won't be output
:return: Future
+ :raises ValueError: tried to declare an anonymous queue
"""
+ if isinstance(obj, Queue) and obj.anonymous:
+ raise ValueError('You cannot declare an anonymous queue!')
if span is not None and not dont_trace:
child_span = self._make_span('declare', span)
else:
@@ -183,6 +186,9 @@ class Cluster(object):
Take care not to lose the Consumer object - it's the only way to cancel a consumer!
+ .. note:: You don't need to explicitly declare queues and exchanges that you will be using beforehand,
+ this will do this for you on the same channel.
+
:param queue: Queue object, being consumed from right now.
Note that name of anonymous queue might change at any time!
:param on_message: callable that will process incoming messages
@@ -232,7 +238,6 @@ class Cluster(object):
def publish(self, message, # type: Message
exchange=None, # type: tp.Union[Exchange, str, bytes]
routing_key=u'', # type: tp.Union[str, bytes]
- tx=None, # type: tp.Optional[bool]
confirm=None, # type: tp.Optional[bool]
span=None, # type: tp.Optional[opentracing.Span]
dont_trace=False # type: bool
@@ -246,9 +251,8 @@ class Cluster(object):
:param confirm: Whether to publish it using confirms/transactions.
If you choose so, you will receive a Future that can be used
to check it broker took responsibility for this message.
- Note that if tx if False, and message cannot be delivered to broker at once,
+ Note that if confirm is False, and message cannot be delivered to broker at once,
it will be discarded
- :param tx: deprecated, alias for confirm
:param span: optionally, current span, if opentracing is installed
:param dont_trace: if set to True, a span won't be generated
:return: Future to be finished on completion or None, is confirm/tx was not chosen
@@ -266,19 +270,8 @@ class Cluster(object):
if isinstance(routing_key, six.text_type):
routing_key = routing_key.encode('utf8')
- if tx is not None: # confirm is a drop-in replacement. tx is unfortunately named
- warnings.warn(u'Use confirm kwarg instead', DeprecationWarning)
-
- if confirm is not None:
- raise RuntimeError(
- u'Using both tx= and confirm= at once does not make sense')
- elif confirm is not None:
- tx = confirm
- else:
- tx = False
-
try:
- if tx:
+ if confirm:
clb = self.pub_tr
else:
clb = self.pub_na
diff --git a/coolamqp/objects.py b/coolamqp/objects.py
index eb6731d1bc2d4658705a259d1299f58eef042c1c..6261633ec8a648ea24712f9b9c40546959a2ac51 100644
--- a/coolamqp/objects.py
+++ b/coolamqp/objects.py
@@ -19,36 +19,33 @@ 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):
+ 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():
- if not isinstance(key, six.binary_type):
- key = key.encode('utf-8')
+ key = tobytes(key)
args.append((key, (value, get_type_for(value))))
else:
for key, value in arguments:
- if not isinstance(key, six.binary_type):
- key = key.encode('utf-8')
+ key = tobytes(key)
args.append((key, (value, get_type_for(value))))
return args
-def toutf8(q):
- if isinstance(q, six.binary_type):
- q = q.decode('utf8')
- return q
-
-
-def tobytes(q):
- if isinstance(q, six.text_type):
- q = q.encode('utf8')
- return q
-
-
class Callable(object):
"""
Add a bunch of callables to one list, and just invoke'm.
@@ -88,8 +85,7 @@ class Message(object):
You can pass a dict - it will be passed to
MessageProperties,
but it's slow - don't do that.
- :type properties: MessageProperties instance, None or a dict (SLOW!)
-
+ :type properties: :class:`coolamqp.objects.MessageProperties` instance
"""
__slots__ = ('body', 'properties')
@@ -231,7 +227,6 @@ class Exchange(object):
Exchange.direct = Exchange()
-
class ServerProperties(object):
"""
An object describing properties of the target server.
@@ -255,11 +250,8 @@ class ServerProperties(object):
elif isinstance(prop_value, list):
prop_value = [toutf8(prop[0]) for prop in prop_value]
self.properties[prop_name] = prop_value
- self.mechanisms = data.mechanisms.tobytes().decode('utf-8').split(' ')
- self.locales = data.locales.tobytes().decode('utf-8')
-
- def __str__(self):
- return '%s %s %s %s' % (self.version, repr(self.properties), self.mechanisms, self.locales)
+ self.mechanisms = toutf8(data.mechanisms).split(' ')
+ self.locales = toutf8(data.locales)
class Queue(object):
@@ -267,50 +259,60 @@ class Queue(object):
This object represents a Queue that applications consume from or publish to.
Create a queue definition.
- :param name: name of the queue. Generates a random uuid.uuid4().hex if not given. Note that this kind of queue
- will probably require to be declared.
+ :param name: name of the queue.
+ None (default) for autogeneration. Just follow the rules for :ref:`anonymq`.
+ If empty string, a UUID name will be generated, and you won't have an anonymous queue anymore.
:param durable: Is the queue durable?
:param exchange: Exchange for this queue to bind to. None for no binding.
- :param exclusive: Is this queue exclusive?
- :param auto_delete: Is this queue auto_delete ?
+ :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
- :warning PendingDeprecationWarning: if a non-exclusive auto_delete queue is created or some other combinations
+ :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
+ :warning DeprecationWarning: if a non-exclusive auto_delete queue is created or some other combinations
that will be soon unavailable (eg. RabbitMQ 4.0).
+ :warning UserWarning: if you're declaring an auto_delete or exclusive, anonymous queue
"""
__slots__ = ('name', 'durable', 'exchange', 'auto_delete', 'exclusive',
'anonymous', 'consumer_tag', 'arguments')
- def __init__(self, name=b'', # type: tp.Union[str, bytes]
+ def __init__(self, name=None, # type: tp.Union[str, bytes, None]
durable=False, # type: bool
exchange=None, # type: tp.Optional[Exchange]
- exclusive=False, # type: bool
- auto_delete=False, # type: bool
+ exclusive=True, # type: bool
+ auto_delete=True, # type: bool
arguments=None # type: tp.Union[tp.List[bytes, tp.Any], tp.Dict[str, tp.Any]]
):
- if not name:
- name = uuid.uuid4().hex
- self.name = tobytes(name) #: public, must be bytes
- # if name is '', this will be filled in with broker-generated name upon declaration
+ if name is None:
+ self.name = None
+ else:
+ name = uuid.uuid4().hex if not name else name
+ self.name = tobytes(name)
+
self.durable = durable
self.exchange = exchange
self.auto_delete = auto_delete
self.exclusive = exclusive
self.arguments = argumentify(arguments)
+ self.anonymous = self.name is None # if this queue is anonymous, it must be regenerated upon reconnect
if self.auto_delete and self.durable:
- warnings.warn('This will be removed in RabbitMQ 4.0', PendingDeprecationWarning)
+ raise ValueError('Cannot create an auto_delete and exclusive queue')
- if self.auto_delete and not self.exclusive:
- warnings.warn('This will be removed in RabbitMQ 4.0', PendingDeprecationWarning)
+ if not self.anonymous:
+ if self.auto_delete or self.exclusive:
+ warnings.warn('This may cause unpredictable behaviour', UserWarning)
+ elif self.durable:
+ raise ValueError('Cannot declare an anonymous durable queue')
- self.anonymous = not len(
- self.name) # if this queue is anonymous, it must be regenerated upon reconnect
+ if self.auto_delete and not self.exclusive and not self.anonymous:
+ raise ValueError('Cannot create an auto_delete and durable queue non-anonymous')
- self.consumer_tag = self.name if not self.anonymous else uuid.uuid4().hex.encode(
- 'utf8') # bytes, consumer tag to use in AMQP comms
+ self.consumer_tag = self.name if not self.anonymous else tobytes(uuid.uuid4().hex)
- assert isinstance(self.name, six.binary_type)
- assert isinstance(self.consumer_tag, six.binary_type)
+ if not self.exclusive and self.auto_delete:
+ warnings.warn('This will be removed in RabbitMQ 4.0', DeprecationWarning)
def __eq__(self, other):
return self.name == other.name
diff --git a/docs/advanced.rst b/docs/advanced.rst
index 2f743c98cab9967642959243cc758a78e2aaccf0..de6b35fbf435b8d66a8dddbf9494570718c8cd93 100644
--- a/docs/advanced.rst
+++ b/docs/advanced.rst
@@ -3,3 +3,15 @@ Advanced things
.. autoclass:: coolamqp.uplink.connection.Connection
:members:
+
+
+Declaring anonymous queues
+--------------------------
+
+.. _anonymq:
+
+In order to make use of an anonymous queue, you must first :meth:`coolamqp.clustering.Cluster.consume` it, since
+:meth:`coolamqp.clustering.Cluster.declare` will use a separate channel, in which the queue will be invalid. It will
+raise ValueError if you try to do that, anyway.
+
+Anonymous queues must be auto_delete and exclusive, ValueError will be raised otherwise.
diff --git a/docs/whatsnew.rst b/docs/whatsnew.rst
index 4e2b86d9feeffb1bc98f317a59794e7e82ffe835..d696f22d3cbbf9af1bacaf8b53404289ae10f308 100644
--- a/docs/whatsnew.rst
+++ b/docs/whatsnew.rst
@@ -1,4 +1,4 @@
-What's new? What's new?
+What's new?
===========
CoolAMQP 2.0.0 marks a slight philosophy shift. Whereas 1.x used auto-generated UUID names, 2.0 will let the server
diff --git a/tests/test_clustering/test_exchanges.py b/tests/test_clustering/test_exchanges.py
index 206ec06a66ef1922e6574f940ed270ea95a3b044..f6e8f932fd8b141bc06b2b4e39c8177db1c84343 100644
--- a/tests/test_clustering/test_exchanges.py
+++ b/tests/test_clustering/test_exchanges.py
@@ -23,6 +23,13 @@ class TestExchanges(unittest.TestCase):
def tearDown(self):
self.c.shutdown()
+ def test_declare_anonymq(self):
+ que = Queue(auto_delete=True)
+ self.assertRaises(ValueError, self.c.declare, que)
+ cons, fut = self.c.consume(que, no_ack=True)
+ fut.result()
+ cons.cancel().result()
+
def test_deadlettering(self):
xchg_name = uuid.uuid4().hex
dead_queue_name = uuid.uuid4().hex
diff --git a/tests/test_objects.py b/tests/test_objects.py
index 9fbf5b156a777f194f03cc68c93dd25eb1681d4d..49e119f7fcad9d18a2eee015b964a66b63df19d8 100644
--- a/tests/test_objects.py
+++ b/tests/test_objects.py
@@ -16,6 +16,12 @@ logging.getLogger('coolamqp').setLevel(logging.DEBUG)
class TestObjects(unittest.TestCase):
+ def test_queue_failures(self):
+ self.assertRaises(ValueError, Queue, None, durable=True)
+ self.assertRaises(ValueError, Queue, 'test', auto_delete=True, durable=True)
+ self.assertRaises(ValueError, Queue, None, auto_delete=False)
+ self.assertRaises(ValueError, Queue, 'test', auto_delete=True, exclusive=False)
+
def test_queue_declare(self):
args = argumentify({'x-dead-letter-exchange': 'deadletter',
'x-message-ttl': 1000})