acemq_amqp.topology

  1# Copyright 2026 AceMQ.
  2#
  3# Licensed under the Apache License, Version 2.0 (the "License");
  4# you may not use this file except in compliance with the License.
  5# You may obtain a copy of the License at
  6#
  7#     https://www.apache.org/licenses/LICENSE-2.0
  8#
  9# Unless required by applicable law or agreed to in writing, software
 10# distributed under the License is distributed on an "AS IS" BASIS,
 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 12# See the License for the specific language governing permissions and
 13# limitations under the License.
 14
 15"""The exchanges, queues and bindings a service expects to find.
 16
 17Declaring them one call at a time works, and stops working the moment somebody
 18needs to know what a service will do to a broker *before* it does it. A topology
 19can be printed, checked and applied, which is the difference between a
 20deployment somebody approves and one they find out about::
 21
 22    topology = (
 23        Topology()
 24        .exchange("orders-events", "topic")
 25        .queue("shipping-orders", dead_letter=True)
 26        .binding("shipping-orders", "orders-events", "order.placed")
 27    )
 28    print(topology)
 29    await topology.apply(transport)
 30
 31Mistakes are raised where they are made rather than collected for later. Go
 32accumulates them on the builder because a Go builder has nowhere else to put
 33them; Python has an exception and a traceback that points at the line that is
 34wrong, which is more useful than a message that says a topology is bad.
 35
 36Two exchanges are this library's own rather than the caller's:
 37:data:`RETRY_EXCHANGE` returns an expired rung message to the queue it came
 38from, and :data:`DEAD_LETTER_EXCHANGE` reaches the dead-letter and parking
 39queues. Both are declared, and bound, by :meth:`Topology.queue` when it is asked
 40for retries or for dead-lettering, so that the whole arrangement appears in
 41:meth:`Topology.plan` and nothing depends on a caller remembering a binding.
 42
 43A consumer declares that same half again when it starts, through
 44:func:`declare_where_failures_go`, without waiting to be asked. Declaring it
 45twice costs two round trips and nothing else, because both declarations carry
 46the same arguments; not declaring it at all costs a message, because a service
 47deployed without its topology republishes a dead letter to a queue the broker
 48cannot route to, and the broker discards what it cannot route in silence.
 49
 50A durable queue asked for here is a **quorum** queue. That is the default in
 51every AceMQ library, and it is a contract rather than a preference for the same
 52reason the rung arguments are: a Java service and a Python service consuming
 53``orders`` both declare ``orders``, and a queue type is compared as strictly as
 54any other argument, so the second one to arrive is answered
 55``PRECONDITION_FAILED`` and cannot consume at all. See :meth:`Topology.queue`
 56for what stays classic and why.
 57"""
 58
 59from __future__ import annotations
 60
 61from collections.abc import Mapping
 62from dataclasses import dataclass, field
 63from datetime import timedelta
 64from typing import Any
 65
 66from . import naming
 67from .retry import RetryPolicy
 68from .transport import ExchangeSpec, QueueSpec, Transport
 69
 70#: Where the broker sends a message this queue rejects or expires.
 71DEAD_LETTER_EXCHANGE_ARG = "x-dead-letter-exchange"
 72
 73#: What routing key it is sent under, which a direct exchange matches against
 74#: its bindings.
 75DEAD_LETTER_ROUTING_KEY_ARG = "x-dead-letter-routing-key"
 76
 77#: How long a message may sit on a queue before the broker expires it, in
 78#: milliseconds. On the queue, never on the message: see :meth:`Topology.queue`.
 79MESSAGE_TTL_ARG = "x-message-ttl"
 80
 81#: The exchange a rung queue dead-letters through on its way back to the queue
 82#: the message came from, bound on the source queue's own name.
 83#:
 84#: This constant and :data:`DEAD_LETTER_EXCHANGE` are the only two places that
 85#: name an exchange this library manages, which is deliberate. A rung is a
 86#: contract rather than a preference: two services consuming the same queue
 87#: declare the same ``{queue}.retry.{delay}`` by name, and a rung declared with
 88#: different arguments answers the second service ``PRECONDITION_FAILED`` and
 89#: leaves it unable to consume at all. Python used to dead-letter a rung through
 90#: the default exchange, which needs no exchange and no binding and works
 91#: perfectly well on its own — but Java has always used a named exchange, most
 92#: of the released code follows Java, and two libraries cannot both be right
 93#: about one queue. This is the settled answer.
 94RETRY_EXCHANGE = "acemq.retry"
 95
 96#: The exchange the dead-letter and parking queues are reached through, each
 97#: bound on its own name.
 98DEAD_LETTER_EXCHANGE = "acemq.dlx"
 99
100#: What kind of queue to declare. Fixed at declaration: a queue that exists
101#: cannot be turned into another kind, only deleted and declared again.
102QUEUE_TYPE_ARG = "x-queue-type"
103
104#: The kind a durable queue asked for through :meth:`Topology.queue` is declared
105#: as, which is what Java, Go, .NET and Ruby declare too.
106#:
107#: There is no matching constant for a classic queue, and that is deliberate: a
108#: classic queue is declared by leaving :data:`QUEUE_TYPE_ARG` off altogether,
109#: which is what the other libraries send and therefore the only spelling a
110#: broker will find equivalent to theirs.
111QUORUM_QUEUE_TYPE = "quorum"
112
113#: How both managed exchanges are declared: direct, because every binding on
114#: them matches a queue name exactly, and durable, because a topology that
115#: vanished with the broker would leave rungs dead-lettering into nothing.
116_MANAGED_EXCHANGE = ExchangeSpec(kind="direct", durable=True)
117
118
119def rung_args(source: str, delay: timedelta) -> dict[str, Any]:
120    """The arguments a ``{source}.retry.{delay}`` queue must be declared with.
121
122    Exactly three keys, and the same three in every AceMQ library: the delay as
123    ``x-message-ttl``, :data:`RETRY_EXCHANGE` as the dead-letter target, and the
124    source queue as the routing key it is sent under. Nothing consumes a rung;
125    the time-to-live is the only thing that ever takes a message out of one, and
126    the binding made by :meth:`Topology.queue` is what brings it home.
127
128    The TTL is on the queue rather than on each message, which is the whole
129    reason a policy needs several queues instead of one. RabbitMQ expires
130    messages only from the head of a queue, so a single queue holding
131    per-message TTLs releases nothing while a message with a long one sits at
132    the front: a thirty-second wait queued behind a ten-minute wait becomes a
133    ten-minute wait. The delays that come out would bear no relation to the ones
134    that went in, and the bug would only appear under the load that puts two
135    different waits on the queue at once.
136
137    :param source: the queue being consumed, which is where an expired message
138        goes back to
139    :param delay: how long the rung holds a message
140    :returns: the argument table, which is a contract and is pinned by a test
141    """
142    return {
143        MESSAGE_TTL_ARG: int(delay.total_seconds() * 1000),
144        DEAD_LETTER_EXCHANGE_ARG: RETRY_EXCHANGE,
145        DEAD_LETTER_ROUTING_KEY_ARG: source,
146    }
147
148
149@dataclass(frozen=True, slots=True)
150class Binding:
151    """Messages matching a routing key going from an exchange to a queue."""
152
153    queue: str
154    exchange: str
155    routing_key: str = ""
156
157    def __str__(self) -> str:
158        return f"{self.exchange} -> {self.queue} ({self.routing_key})"
159
160
161@dataclass(frozen=True, slots=True)
162class PlanAction:
163    """One thing applying a topology would do."""
164
165    kind: str
166    name: str
167    detail: str = ""
168
169    def __str__(self) -> str:
170        if not self.detail:
171            return f"declare {self.kind} {self.name}"
172        return f"declare {self.kind} {self.name} ({self.detail})"
173
174
175@dataclass(frozen=True, slots=True)
176class _NamedQueue:
177    name: str
178    spec: QueueSpec
179
180
181@dataclass(frozen=True, slots=True)
182class _NamedExchange:
183    name: str
184    spec: ExchangeSpec
185
186
187@dataclass
188class Topology:
189    """A description of what a service needs a broker to have.
190
191    Every method that adds something returns the topology, so a description
192    reads as one expression. The order things are added in does not matter:
193    :meth:`apply` declares exchanges first, then queues, then the bindings
194    between them, because that is the order a broker will accept them in.
195    """
196
197    _exchanges: list[_NamedExchange] = field(default_factory=list)
198    _queues: list[_NamedQueue] = field(default_factory=list)
199    _bindings: list[Binding] = field(default_factory=list)
200
201    def exchange(
202        self,
203        name: str,
204        kind: str = "topic",
205        *,
206        durable: bool = True,
207        auto_delete: bool = False,
208        args: Mapping[str, Any] | None = None,
209    ) -> Topology:
210        """Adds an exchange.
211
212        :param name: what to call it
213        :param kind: ``direct``, ``topic``, ``fanout`` or ``headers``
214        :param durable: survives a broker restart
215        :param auto_delete: goes away when its last binding does
216        :param args: broker-specific arguments
217        :returns: this topology
218        """
219        if not name:
220            raise ValueError("acemq: an exchange needs a name")
221        if not kind:
222            raise ValueError(f"acemq: exchange {name!r} needs a kind (direct, topic, fanout)")
223        self._exchanges.append(
224            _NamedExchange(
225                name,
226                ExchangeSpec(
227                    kind=kind,
228                    durable=durable,
229                    auto_delete=auto_delete,
230                    args=dict(args or {}),
231                ),
232            )
233        )
234        return self
235
236    def queue(
237        self,
238        name: str,
239        *,
240        durable: bool = True,
241        auto_delete: bool = False,
242        exclusive: bool = False,
243        quorum: bool | None = None,
244        dead_letter: bool = False,
245        retry: RetryPolicy | None = None,
246        args: Mapping[str, Any] | None = None,
247    ) -> Topology:
248        """Adds a queue, and the queues that catch what it cannot handle.
249
250        A durable queue is a **quorum** queue unless it is asked to be something
251        else. That is what Java has always declared and what the other libraries
252        declare now, and it has to be the same everywhere: the queue type is one
253        more argument the broker compares, so a Java service and a Python
254        service that disagree about ``orders`` cannot both consume it. Pass
255        ``quorum=False`` for a classic queue, which is declared by leaving
256        ``x-queue-type`` off entirely — the spelling every AceMQ library uses,
257        and therefore the only one a broker finds equivalent to theirs.
258
259        ``exclusive``, ``auto_delete`` and ``durable=False`` all mean a queue
260        that belongs to one connection, and RabbitMQ refuses a quorum queue that
261        is any of those. Such a queue is declared classic without being asked,
262        because the alternative is a declaration the broker rejects with a
263        message that never mentions the word quorum. Asking for both — a queue
264        that is ``quorum=True`` and exclusive — is refused here instead, where
265        the contradiction is written down.
266
267        A queue whose ``args`` already name a kind keeps it: that is how
268        :func:`acemq_amqp.patterns.stream` declares a stream, and a caller who
269        wrote ``x-queue-type`` down meant it. Writing it down *and* passing
270        ``quorum`` is two answers to one question and is refused.
271
272        ``dead_letter`` declares ``{name}.dlq`` alongside it and points the
273        broker at it, so a message this queue rejects or lets expire lands
274        somewhere an operator can find by name rather than disappearing. It
275        declares ``{name}.parked`` too, because a consumer on this queue sends
276        anything it cannot decode there — and a library that parks messages into
277        a queue nobody declared has only moved the disappearance somewhere else.
278
279        Neither dead-letters itself: a dead-letter queue that dead-letters is a
280        loop, and a loop is how a poison message becomes an outage. Both are
281        reached through :data:`DEAD_LETTER_EXCHANGE`, bound on their own names,
282        which is what Java has always done and is therefore what a broker shared
283        with a Java service already has. A service that wants its own
284        dead-letter exchange instead should leave ``dead_letter`` alone and pass
285        the arguments in ``args``.
286
287        ``retry`` declares the rung queues that policy's long waits need: one
288        ``{name}.retry.{delay}`` per distinct delay at or above the policy's
289        threshold, each declared with :func:`rung_args` — the delay as
290        ``x-message-ttl``, :data:`RETRY_EXCHANGE` as the dead-letter target, and
291        this queue as the routing key. The binding that brings an expired
292        message home, ``{name}`` to :data:`RETRY_EXCHANGE` on ``{name}``, is
293        made here and not left to a caller to remember: without it every rung
294        expires into an exchange that routes nowhere, and a retry that is
295        dropped by the broker looks exactly like one that is still waiting.
296
297        It takes the policy rather than a list of delays on purpose. The rungs a
298        consumer will publish to are derived from the policy it is running, so
299        anything else here would be a second copy of the same list, free to drift
300        from the first — and the way that drift shows up is a retry published to
301        a queue nobody declared, at the moment the service is already failing.
302        One value produces both, or the topology is not a description of what the
303        service needs.
304
305        :param name: what to call it
306        :param durable: survives a broker restart
307        :param auto_delete: goes away when its last consumer does
308        :param exclusive: usable only by the connection that declared it
309        :param quorum: replicated rather than classic. ``None``, the default,
310            means quorum for a durable queue and classic for one belonging to a
311            single connection
312        :param dead_letter: also declare and wire ``{name}.dlq``, and declare
313            ``{name}.parked``
314        :param retry: also declare the rung queues this policy's long waits use
315        :param args: broker-specific arguments
316        :returns: this topology
317        :raises ValueError: when the queue cannot be what it was asked to be
318        """
319        if not name:
320            raise ValueError("acemq: a queue needs a name")
321
322        arguments: dict[str, Any] = dict(args or {})
323        kind = _queue_type(
324            name,
325            arguments,
326            durable=durable,
327            auto_delete=auto_delete,
328            exclusive=exclusive,
329            quorum=quorum,
330        )
331        if kind is not None:
332            arguments[QUEUE_TYPE_ARG] = kind
333        if dead_letter:
334            # Refused rather than resolved, because either answer would be a
335            # guess about which of two conflicting instructions was meant.
336            conflicting = sorted(
337                key
338                for key in arguments
339                if key in (DEAD_LETTER_EXCHANGE_ARG, DEAD_LETTER_ROUTING_KEY_ARG)
340            )
341            if conflicting:
342                raise ValueError(
343                    f"acemq: queue {name!r} asks for dead_letter=True and also sets "
344                    + ", ".join(conflicting)
345                    + "; pick one"
346                )
347            arguments[DEAD_LETTER_EXCHANGE_ARG] = DEAD_LETTER_EXCHANGE
348            arguments[DEAD_LETTER_ROUTING_KEY_ARG] = naming.dead_letter_queue(name)
349
350        self._queues.append(
351            _NamedQueue(
352                name,
353                QueueSpec(
354                    durable=durable,
355                    auto_delete=auto_delete,
356                    exclusive=exclusive,
357                    args=arguments,
358                ),
359            )
360        )
361        if dead_letter:
362            self._managed_exchange(DEAD_LETTER_EXCHANGE)
363            # Classic, deliberately, whatever the queue in front of them is.
364            # Java declares both QueueType.CLASSIC and these are shared names:
365            # a dead-letter queue nobody consumes gains nothing from being
366            # replicated, and gaining it here would cost a Java service its
367            # declaration of the same queue.
368            for target in (naming.dead_letter_queue(name), naming.parked_queue(name)):
369                self._queues.append(_NamedQueue(target, QueueSpec(durable=durable)))
370                self._bindings.append(
371                    Binding(queue=target, exchange=DEAD_LETTER_EXCHANGE, routing_key=target)
372                )
373        if retry is not None:
374            rungs = retry.broker_rungs()
375            if rungs:
376                self._managed_exchange(RETRY_EXCHANGE)
377                # Classic too, and for the same reason as ``.dlq``: a rung is
378                # declared by every service consuming the source queue, Java
379                # declares it QueueType.CLASSIC, and :func:`rung_args` is the
380                # whole argument table — three keys, no ``x-queue-type``.
381                for rung in rungs:
382                    self._queues.append(
383                        _NamedQueue(
384                            naming.retry_queue(name, rung),
385                            QueueSpec(durable=durable, args=rung_args(name, rung)),
386                        )
387                    )
388                # One binding brings every expired message back to the queue it
389                # came from. Made here rather than offered as something to add,
390                # because a missing binding loses every retry silently: the
391                # broker drops what a direct exchange cannot route and says
392                # nothing, so the queue looks quiet rather than broken.
393                self._bindings.append(
394                    Binding(queue=name, exchange=RETRY_EXCHANGE, routing_key=name)
395                )
396        return self
397
398    def _managed_exchange(self, name: str) -> None:
399        """Declares one of this library's own exchanges, once.
400
401        Several queues in one topology each need :data:`RETRY_EXCHANGE` and
402        :data:`DEAD_LETTER_EXCHANGE`, and declaring either twice is what
403        :meth:`validate` refuses. A caller who has already named the same
404        exchange keeps their own spec: theirs is the deliberate one.
405        """
406        if any(entry.name == name for entry in self._exchanges):
407            return
408        self._exchanges.append(_NamedExchange(name, _MANAGED_EXCHANGE))
409
410    def binding(self, queue: str, exchange: str, routing_key: str = "") -> Topology:
411        """Routes messages matching a key from an exchange to a queue.
412
413        :param queue: where messages end up, which this topology must declare
414        :param exchange: where they come from, which this topology must declare
415        :param routing_key: what to match, empty for a fanout
416        :returns: this topology
417        """
418        self._bindings.append(Binding(queue=queue, exchange=exchange, routing_key=routing_key))
419        return self
420
421    @property
422    def queues(self) -> list[str]:
423        """The queue names this topology declares, in declaration order."""
424        return [entry.name for entry in self._queues]
425
426    @property
427    def exchanges(self) -> list[str]:
428        """The exchange names this topology declares, in declaration order."""
429        return [entry.name for entry in self._exchanges]
430
431    @property
432    def bindings(self) -> list[Binding]:
433        """The bindings this topology declares."""
434        return list(self._bindings)
435
436    def validate(self) -> None:
437        """Reports what is wrong with the description itself.
438
439        A binding naming a queue the topology does not declare is the mistake
440        worth catching here. The broker would accept it if the queue happened to
441        exist already, and the service would then depend on something nothing
442        declares — which works until the day it is deployed somewhere new.
443
444        :raises ValueError: when the topology cannot be right
445        """
446        seen_queues: set[str] = set()
447        for queue in self._queues:
448            if queue.name in seen_queues:
449                raise ValueError(f"acemq: the topology declares queue {queue.name!r} twice")
450            seen_queues.add(queue.name)
451
452        seen_exchanges: set[str] = set()
453        for exchange in self._exchanges:
454            if exchange.name in seen_exchanges:
455                raise ValueError(
456                    f"acemq: the topology declares exchange {exchange.name!r} twice"
457                )
458            seen_exchanges.add(exchange.name)
459
460        for binding in self._bindings:
461            if binding.queue not in seen_queues:
462                raise ValueError(
463                    f"acemq: binding {binding} names queue {binding.queue!r}, "
464                    "which this topology does not declare"
465                )
466            if not binding.exchange:
467                raise ValueError(
468                    f"acemq: binding {binding} names the default exchange, "
469                    "which cannot be bound to"
470                )
471            if binding.exchange not in seen_exchanges:
472                raise ValueError(
473                    f"acemq: binding {binding} names exchange {binding.exchange!r}, "
474                    "which this topology does not declare"
475                )
476
477    def plan(self) -> list[PlanAction]:
478        """What :meth:`apply` would do, without doing it.
479
480        Deliberately not a difference against the live broker: AMQP offers no
481        way to enumerate what is there without the management API, and a plan
482        that quietly guessed would be worse than one that is honest about being
483        a statement of intent.
484
485        :returns: one action per thing that would be declared
486        :raises ValueError: when the topology cannot be right
487        """
488        self.validate()
489        actions = [
490            PlanAction("exchange", entry.name, _describe(entry.spec))
491            for entry in self._exchanges
492        ]
493        actions += [
494            PlanAction("queue", entry.name, _describe(entry.spec)) for entry in self._queues
495        ]
496        actions += [
497            PlanAction(
498                "binding", binding.queue, f"from {binding.exchange} on {binding.routing_key}"
499            )
500            for binding in self._bindings
501        ]
502        return actions
503
504    async def apply(self, transport: Transport) -> None:
505        """Declares everything, in the order a broker needs.
506
507        It stops at the first failure. A queue that already exists with
508        different settings is refused by the broker with ``PRECONDITION_FAILED``,
509        and that refusal is passed on rather than swallowed: it means this
510        service and the broker disagree about what the queue is, and carrying on
511        would leave the service using a queue that is not the one it asked for.
512
513        :param transport: where to declare it
514        :raises ValueError: when the topology cannot be right
515        """
516        self.validate()
517        for exchange in self._exchanges:
518            await transport.declare_exchange(exchange.name, exchange.spec)
519        for queue in self._queues:
520            await transport.declare_queue(queue.name, queue.spec)
521        for binding in self._bindings:
522            await transport.bind(binding.queue, binding.exchange, binding.routing_key)
523
524    def __str__(self) -> str:
525        try:
526            actions = self.plan()
527        except ValueError as failure:
528            return f"Topology(invalid: {failure})"
529        header = (
530            f"Topology: {len(self._exchanges)} exchanges, "
531            f"{len(self._queues)} queues, {len(self._bindings)} bindings"
532        )
533        return "\n".join([header, *(f"  {action}" for action in actions)])
534
535
536def _queue_type(
537    name: str,
538    args: Mapping[str, Any],
539    *,
540    durable: bool,
541    auto_delete: bool,
542    exclusive: bool,
543    quorum: bool | None,
544) -> str | None:
545    """What to declare a queue as, or ``None`` to leave the argument off.
546
547    ``None`` is classic and is a real answer rather than a missing one: every
548    AceMQ library declares a classic queue by sending no ``x-queue-type`` at
549    all, so this matches Java's ``QueueType.CLASSIC`` argument for argument.
550
551    RabbitMQ does in fact normalise the two — a queue declared with no
552    ``x-queue-type`` and one declared with ``classic`` are the same queue to
553    the broker, checked in both orders against a real one, while quorum against
554    either is refused. So sending ``classic`` would also work. Sending nothing
555    is still what this does, because "declare exactly what the other libraries
556    declare" is a rule that holds without anyone having to remember which
557    arguments the broker happens to normalise and which it compares literally.
558
559    :param name: the queue, for the message when the answer is a refusal
560    :param args: what the caller wrote down, which may already name a kind
561    :param durable: survives a broker restart
562    :param auto_delete: goes away when its last consumer does
563    :param exclusive: usable only by the connection that declared it
564    :param quorum: what the caller asked for, or ``None`` for the default
565    :returns: the value for :data:`QUEUE_TYPE_ARG`, or ``None`` for classic
566    :raises ValueError: when what was asked for cannot exist
567    """
568    written = args.get(QUEUE_TYPE_ARG)
569    if written is not None:
570        if quorum is not None:
571            raise ValueError(
572                f"acemq: queue {name!r} sets {QUEUE_TYPE_ARG}={written!r} and also passes "
573                f"quorum={quorum!r}; pick one"
574            )
575        return str(written)
576
577    # A queue tied to one connection cannot be a quorum queue. RabbitMQ refuses
578    # exclusive, auto-delete and transient outright, and the refusal it sends
579    # back does not say which of the three it objected to.
580    per_connection = exclusive or auto_delete or not durable
581    if quorum is None:
582        return None if per_connection else QUORUM_QUEUE_TYPE
583    if not quorum:
584        return None
585    if per_connection:
586        refused = ", ".join(
587            label
588            for label, set_ in (
589                ("exclusive", exclusive),
590                ("auto_delete", auto_delete),
591                ("durable=False", not durable),
592            )
593            if set_
594        )
595        raise ValueError(
596            f"acemq: queue {name!r} asks for quorum=True and also {refused}; "
597            "a quorum queue belongs to the broker rather than to a connection, "
598            "so it can be none of those"
599        )
600    return QUORUM_QUEUE_TYPE
601
602
603async def declare_where_failures_go(
604    transport: Transport, queue: str, retry: RetryPolicy | None = None
605) -> None:
606    """Declares the queues a consumer of ``queue`` will need when it fails.
607
608    ``acemq.dlx``, ``{queue}.dlq``, ``{queue}.parked`` and the two bindings that
609    reach them; and, when ``retry`` has waits long enough to be held by the
610    broker, ``acemq.retry``, one rung queue per distinct wait, and the binding
611    that brings an expired message back to ``queue``. Exactly what
612    :meth:`Topology.queue` writes for the same arguments, and exactly what Java's
613    consumer declares when it starts.
614
615    Called by :meth:`acemq_amqp.Connection.consume` before it subscribes, which
616    is the point of it: everything here is somewhere a message goes when the
617    handler has given up, and a message republished to a queue that is not there
618    is discarded by the broker without a word. Applying a topology first is still
619    the right way to run a service — it is reviewable, and it declares the source
620    queue with the ``x-dead-letter-exchange`` this cannot — but a service that was
621    deployed without one should lose a message to a bug rather than to a missing
622    queue.
623
624    The source queue is deliberately not declared. It belongs to whoever set the
625    service up, who chose its type and its arguments, and guessing wrong is a
626    ``PRECONDITION_FAILED`` that stops the consumer starting at all.
627
628    Safe to call repeatedly, and safe to call after :meth:`Topology.apply` or
629    before it: every declaration here carries the same arguments the topology
630    would send, so the second one to arrive is a no-op rather than a refusal.
631
632    :param transport: where to declare it
633    :param queue: the queue being consumed
634    :param retry: the policy the consumer is running, whose long waits decide
635        which rungs exist. ``None``, or a policy that holds every wait in the
636        consumer, declares only the dead-letter half
637    """
638    await transport.declare_exchange(DEAD_LETTER_EXCHANGE, _MANAGED_EXCHANGE)
639    for target in (naming.dead_letter_queue(queue), naming.parked_queue(queue)):
640        # Classic, whatever the queue in front of them is, for the reason
641        # :meth:`Topology.queue` gives: these are shared names, and a Java
642        # service declaring the same queue declares it classic.
643        await transport.declare_queue(target, QueueSpec(durable=True))
644        await transport.bind(target, DEAD_LETTER_EXCHANGE, target)
645
646    # Keyed by name rather than by delay. Two delays can render to the same name,
647    # and a second queue by that name with a different time-to-live is not a
648    # second rung — it is the PRECONDITION_FAILED that stops this consumer
649    # starting. The first delay to reach a name owns it, which is what Java does.
650    rungs: dict[str, timedelta] = {}
651    for delay in retry.broker_rungs() if retry is not None else []:
652        rungs.setdefault(naming.retry_queue(queue, delay), delay)
653    if not rungs:
654        return
655
656    await transport.declare_exchange(RETRY_EXCHANGE, _MANAGED_EXCHANGE)
657    for name, delay in rungs.items():
658        await transport.declare_queue(
659            name, QueueSpec(durable=True, args=rung_args(queue, delay))
660        )
661    # One binding brings every expired message home. Without it a rung expires
662    # into an exchange that routes nowhere, and the broker drops what it cannot
663    # route in silence: the retry looks like one still waiting rather than one
664    # that is gone.
665    await transport.bind(queue, RETRY_EXCHANGE, queue)
666
667
668def _describe(spec: QueueSpec | ExchangeSpec) -> str:
669    """A spec as the line an operator would want in a deployment log."""
670    parts: list[str] = []
671    if isinstance(spec, ExchangeSpec):
672        parts.append(spec.kind)
673    parts.append("durable" if spec.durable else "transient")
674    if spec.auto_delete:
675        parts.append("auto-delete")
676    if isinstance(spec, QueueSpec) and spec.exclusive:
677        parts.append("exclusive")
678    parts += [f"{key}={spec.args[key]!r}" for key in sorted(spec.args)]
679    return ", ".join(parts)