acemq_amqp

31class AceMQError(Exception):
32    """Something this library refused to do."""

70@dataclass(frozen=True, slots=True)
71class Ack:
72    """A handler's decision: it worked, try it again, never again, or nobody
73    can read it.
74
75    Returned rather than performed, so a handler that forgets to decide is a
76    handler that returns ``None`` and is caught immediately, rather than one
77    whose message sits unacknowledged until the connection drops and then comes
78    back — usually to the same handler, with the same outcome.
79    """
80
81    action: Action
82    error: BaseException | None = None
83    #: How long to wait before the message is looked at again. Read for
84    #: :attr:`Action.IN_PROGRESS` only; a retry's wait is the policy's.
85    delay: timedelta | None = None
86
87    def __str__(self) -> str:
88        return self.action.value

60class Action(Enum):
61    """The things that can be done with a delivered message."""
62
63    ACCEPT = "accept"
64    RETRY = "retry"
65    REJECT = "reject"
66    PARK = "park"
67    IN_PROGRESS = "in_progress"

230@runtime_checkable
231class BlockedState(Protocol):
232    """A transport that can say whether the broker has blocked its connection.
233
234    RabbitMQ sends ``connection.blocked`` when it is low on memory or disk and
235    ``connection.unblocked`` when the alarm clears. In between it stops reading
236    from the socket, so everything sent on that connection — a publish, a
237    declaration, a health probe — waits rather than failing. From outside that
238    is indistinguishable from a broker that has silently gone away, and the two
239    want opposite responses: a blocked broker is one to wait for, a dead one is
240    one to fail over from.
241
242    Separate from :class:`Transport` because not every transport has an answer.
243    A fake in a test is never blocked and can say so; a client library that
244    keeps the state to itself cannot be asked at all, which is what ``None``
245    below is for. It is the same shape as Java's ``isBlocked()`` and
246    ``blockedReason()``, .NET's ``IsBlocked``/``BlockedReason`` and Go's
247    ``BlockedReason()``.
248    """
249
250    @property
251    def blocked(self) -> bool | None:
252        """Whether the broker has blocked this connection.
253
254        ``None`` rather than ``False`` when the question could not be asked at
255        all — a client that does not expose the state, a connection that is
256        between reconnections. The two are different facts and a report that
257        folds them together says a broker is fine when nobody looked.
258        """
259
260    @property
261    def blocked_reason(self) -> str | None:
262        """What the broker said when it blocked this connection.
263
264        ``None`` when it is not blocked, and also ``None`` when the client read
265        the reason and did not keep it — which is what aio-pika does, and why
266        this is separate from :attr:`blocked` rather than a string that is empty
267        when there is nothing to say. A reason invented here would read exactly
268        like one RabbitMQ sent.
269        """

1977@dataclass(frozen=True, slots=True)
1978class BrokerHealth:
1979    """A connection as a :class:`~acemq_amqp.telemetry.HealthCheck`.
1980
1981    So that a broker check and an application's own — a database, a downstream
1982    service — can be handed to
1983    :func:`~acemq_amqp.telemetry.aggregate_health` together, which is what a
1984    readiness endpoint actually needs.
1985
1986    A blocked broker comes back through here as **up, with the reason**, and
1987    nothing in between folds it into anything else: the aggregate takes the
1988    worst status it is given, so a check that reported back pressure as degraded
1989    would quietly degrade a report that had decided otherwise. The deadline is
1990    the same argument — :attr:`timeout` is shorter than the aggregate's, so that
1991    a broker which has stopped answering is described by this check rather than
1992    by the aggregate giving up on it.
1993
1994    :param connection: what to ask
1995    :param label: what to call it in a combined report
1996    :param timeout: how long to give the broker's answer. See
1997        :meth:`Connection.health`
1998    """
1999
2000    connection: Connection
2001    label: str = "broker"
2002    timeout: float = DEFAULT_HEALTH_TIMEOUT
2003
2004    @property
2005    def name(self) -> str:
2006        return self.label
2007
2008    async def check(self) -> HealthReport:
2009        return await self.connection.health(self.timeout)

145class BytesCodec:
146    """Passes bodies through without touching them.
147
148    For a payload that is already encoded — an image, something another system's
149    serializer produced — and for reading a message this process has no type
150    for. Replaying a dead-lettered message uses it: the bytes that were
151    committed are the bytes that should go back, and re-encoding through a class
152    that has since gained a field would produce something else.
153
154    It answers for every content type, which is why it has to be asked for
155    rather than found: put it last in a :class:`CompositeCodec`, or nothing
156    after it is ever reached.
157    """
158
159    @property
160    def content_type(self) -> str:
161        return BYTES_CONTENT_TYPE
162
163    def encode(self, payload: Any) -> bytes:
164        if payload is None:
165            return b""
166        if isinstance(payload, bytes):
167            return payload
168        if isinstance(payload, bytearray | memoryview):
169            return bytes(payload)
170        if isinstance(payload, str):
171            return payload.encode("utf-8")
172        raise TypeError(
173            f"acemq: BytesCodec cannot encode a {type(payload).__name__}; "
174            "it takes bytes or a string"
175        )
176
177    def decode(self, body: bytes, content_type: str | None = None) -> Any:
178        return bytes(body)
179
180    def can_decode(self, content_type: str | None) -> bool:
181        return True

53@runtime_checkable
54class Codec(Protocol):
55    """Reads and writes message bodies in one format."""
56
57    @property
58    def content_type(self) -> str:
59        """What this codec writes onto a message it encodes."""
60
61    def encode(self, payload: Any) -> bytes:
62        """Turns a payload into bytes.
63
64        :param payload: what the application wants to send
65        :returns: the body
66        :raises TypeError: when the payload is not something this codec writes
67        """
68
69    def decode(self, body: bytes, content_type: str | None = None) -> Any:
70        """Turns bytes back into a payload.
71
72        :param body: the message body
73        :param content_type: what the sender said it was, or ``None`` when the
74            sender said nothing
75        :returns: the payload
76        :raises FatalError: when the body cannot be read, which is fatal because
77            the same bytes will not read any better on a second attempt
78        """
79
80    def can_decode(self, content_type: str | None) -> bool:
81        """Whether this codec should be offered a message.
82
83        :param content_type: the sender's content type, or ``None`` when there
84            was none
85        :returns: whether this codec claims it
86        """

224class CompositeCodec:
225    """Picks a codec by the message's content type.
226
227    For a queue carrying more than one format — during a migration, or where
228    several producers were written at different times::
229
230        codec = CompositeCodec(JsonCodec(), TextCodec())
231
232    The first codec is what it writes. All of them are offered a message to
233    read, in the order they were given, and the first that claims the content
234    type *and* succeeds wins. Order matters when two overlap: :class:`BytesCodec`
235    answers for everything, so anything after it would never be reached.
236    """
237
238    def __init__(self, *codecs: Codec) -> None:
239        if not codecs:
240            raise ValueError("acemq: a CompositeCodec needs at least one codec in it")
241        self._codecs: tuple[Codec, ...] = codecs
242
243    @property
244    def codecs(self) -> tuple[Codec, ...]:
245        """The codecs, in the order they are tried."""
246        return self._codecs
247
248    @property
249    def content_type(self) -> str:
250        """The first codec's, since that is the one that encodes."""
251        return self._codecs[0].content_type
252
253    def encode(self, payload: Any) -> bytes:
254        return self._codecs[0].encode(payload)
255
256    def decode(self, body: bytes, content_type: str | None = None) -> Any:
257        """Reads a body with whichever codec recognises the content type.
258
259        With no content type nothing can be ruled out, so every codec is a
260        candidate rather than the first being assumed right. Candidates are
261        tried in turn and the first that reads the body wins — so a message that
262        claims one format and is written in another is still read — and only
263        when every one has refused is anything raised, carrying all the reasons
264        rather than just the last.
265        """
266        if content_type is None:
267            candidates = list(self._codecs)
268        else:
269            candidates = [codec for codec in self._codecs if codec.can_decode(content_type)]
270
271        if not candidates:
272            raise FatalError(
273                f"acemq: no codec here reads {content_type!r}; "
274                f"this one holds {self._describe()}"
275            )
276
277        refusals: list[str] = []
278        for codec in candidates:
279            try:
280                return codec.decode(body, content_type)
281            except Exception as failure:
282                # Collected rather than raised: a codec refusing says nothing
283                # about the next one, and the message that matters at the end is
284                # every reason together rather than whichever came last.
285                refusals.append(f"{type(codec).__name__}: {failure}")
286
287        raise FatalError(
288            "acemq: no codec could read this message. Tried " + "; ".join(refusals)
289        )
290
291    def can_decode(self, content_type: str | None) -> bool:
292        return any(codec.can_decode(content_type) for codec in self._codecs)
293
294    def _describe(self) -> str:
295        return ", ".join(codec.content_type for codec in self._codecs)
296
297    def __repr__(self) -> str:
298        return f"CompositeCodec({self._describe()})"

1221class Connection:
1222    """A connection to a broker, and the defaults everything on it inherits.
1223
1224    Built by :func:`connect` in the ordinary case. Taking a transport directly
1225    is for a fake in a test, or for a connection made with settings a URL cannot
1226    carry.
1227
1228    :param transport: how to reach the broker
1229    :param codec: what publishers and consumers use unless they say otherwise
1230    :param origin: what to stamp on published messages, conventionally
1231        ``service@host``
1232    :param retry: what consumers use unless they say otherwise. The default is
1233        one delivery and no second chance, so a handler that calls
1234        ``retry()`` on a connection with no policy dead-letters the message and
1235        says so in the reason. That is deliberate: the alternative default is an
1236        immediate requeue, which is a hot loop against a broker that nobody
1237        asked for
1238    :param prefetch: how many unacknowledged messages a consumer holds
1239    :param max_outstanding_publishes: how many publishes may be waiting for the
1240        broker at once. The ceiling that turns a memory leak into backpressure:
1241        without one, a caller publishing faster than the broker answers
1242        accumulates unconfirmed messages until the process dies, which looks
1243        like throughput right up to the moment it does not. A thousand by
1244        default, which is what Java's ``maxOutstandingPublishes`` and .NET's
1245        ``MaxOutstandingPublishes`` default to
1246    :param confirm_timeout: how long a publish waits for one of those permits,
1247        and then for the broker's confirm, before giving up. Ten seconds by
1248        default, the same as Java's ``confirmTimeout``. Reaching it raises
1249        rather than waiting on, because a publisher stalled for ever behind a
1250        broker that has stopped answering is the failure this bound exists to make visible
1251    :param on_publish: cross-cutting behaviour to wrap every publish on this
1252        connection in, outermost first. See :mod:`acemq_amqp.interceptors`
1253    :param on_consume: the same around every handler.
1254        :meth:`intercept_publish` and :meth:`intercept_consume` add more later
1255    :param observer: where the numbers go. Nothing by default, so a service
1256        that wants none pays for none. See :mod:`acemq_amqp.telemetry`
1257    """
1258
1259    def __init__(
1260        self,
1261        transport: Transport,
1262        *,
1263        codec: Codec | None = None,
1264        origin: str | None = None,
1265        retry: RetryPolicy | None = None,
1266        prefetch: int = DEFAULT_PREFETCH,
1267        max_outstanding_publishes: int = DEFAULT_MAX_OUTSTANDING_PUBLISHES,
1268        confirm_timeout: timedelta = DEFAULT_CONFIRM_TIMEOUT,
1269        on_publish: Sequence[PublishInterceptor] = (),
1270        on_consume: Sequence[ConsumeInterceptor] = (),
1271        observer: Observer | None = None,
1272    ) -> None:
1273        if prefetch < 0:
1274            raise ValueError(f"acemq: prefetch must not be negative, got {prefetch}")
1275        if max_outstanding_publishes < 1:
1276            raise ValueError(
1277                "acemq: max_outstanding_publishes must be at least 1, got "
1278                f"{max_outstanding_publishes}"
1279            )
1280        if confirm_timeout <= timedelta(0):
1281            raise ValueError(
1282                f"acemq: confirm_timeout must be positive, got {confirm_timeout}"
1283            )
1284        self._transport = transport
1285        self._observer: Observer = observer or NullObserver()
1286        self._codec = codec or JsonCodec()
1287        self._origin = origin or default_origin()
1288        self._retry = retry or no_retry()
1289        self._prefetch = prefetch
1290        self._max_outstanding = max_outstanding_publishes
1291        self._confirm_timeout = confirm_timeout
1292        # Built here rather than on first use. An asyncio.Semaphore has bound no
1293        # loop at construction since 3.10, so it is safe to make one outside a
1294        # running loop, and making it here means every publish on this connection
1295        # shares the one counter however many tasks are publishing.
1296        self._outstanding = asyncio.Semaphore(max_outstanding_publishes)
1297        self._in_flight_publishes = 0
1298        self._consumers: list[Consumer] = []
1299        self._publishing: list[PublishInterceptor] = list(on_publish)
1300        self._consuming: list[ConsumeInterceptor] = list(on_consume)
1301        self._closed = False
1302
1303    @property
1304    def transport(self) -> Transport:
1305        """The transport underneath, for anything this library does not wrap."""
1306        return self._transport
1307
1308    @property
1309    def codec(self) -> Codec:
1310        """The codec publishers and consumers use unless told otherwise."""
1311        return self._codec
1312
1313    @property
1314    def origin(self) -> str:
1315        """What published messages are stamped with."""
1316        return self._origin
1317
1318    @property
1319    def retry(self) -> RetryPolicy:
1320        """The retry policy consumers use unless told otherwise."""
1321        return self._retry
1322
1323    @property
1324    def observer(self) -> Observer:
1325        """Where this connection's numbers go."""
1326        return self._observer
1327
1328    @property
1329    def max_outstanding_publishes(self) -> int:
1330        """How many publishes may be waiting for the broker at once."""
1331        return self._max_outstanding
1332
1333    @property
1334    def confirm_timeout(self) -> timedelta:
1335        """How long a publish waits for room before giving up."""
1336        return self._confirm_timeout
1337
1338    @property
1339    def outstanding_publishes(self) -> int:
1340        """How many publishes are waiting for the broker right now.
1341
1342        What a stalled publisher looks like from outside: a number pinned at
1343        :attr:`max_outstanding_publishes` is a broker that has stopped answering,
1344        and it says so before the first :class:`~acemq_amqp.PublishError` does.
1345        """
1346        # Counted here rather than read off the semaphore, which keeps its count
1347        # private. Raised after the permit is taken and lowered before it is
1348        # given back, both inside publish_raw, so it never reads high.
1349        return self._in_flight_publishes
1350
1351    @property
1352    def consumers(self) -> tuple[Consumer, ...]:
1353        """The consumers running on this connection.
1354
1355        A snapshot, and public because a health check that has to say whether
1356        the consumers are alive should not need private access to find them —
1357        the same rule an interceptor follows.
1358        """
1359        return tuple(self._consumers)
1360
1361    @property
1362    def closed(self) -> bool:
1363        """Whether this connection has been closed."""
1364        return self._closed
1365
1366    @property
1367    def blocked(self) -> bool | None:
1368        """Whether the broker has blocked this connection, or ``None`` when the
1369        transport cannot be asked.
1370
1371        RabbitMQ blocks a connection when it is low on memory or disk, and while
1372        it lasts it stops reading that connection's socket: everything sent —
1373        a publish, a declaration, :meth:`health`'s own probe — waits rather than
1374        failing. It is the broker protecting itself and it is not an error, so
1375        it is worth telling apart from the broker having gone away, which looks
1376        identical from outside and wants the opposite response.
1377
1378        ``None`` means the question could not be asked at all, which is a
1379        different fact from ``False`` and is reported as one: a transport that
1380        does not implement
1381        :class:`~acemq_amqp.transport.BlockedState`, or one whose client has
1382        moved the state out of reach. See :attr:`blocked_reason` for why there
1383        may be a boolean and no reason to go with it.
1384        """
1385        transport = self._transport
1386        return transport.blocked if isinstance(transport, BlockedState) else None
1387
1388    @property
1389    def blocked_reason(self) -> str | None:
1390        """What the broker said when it blocked this connection, if anything.
1391
1392        ``None`` when the connection is not blocked, and also ``None`` when the
1393        transport is blocked but cannot say why — which is what the RabbitMQ
1394        transport answers, because aio-pika reads RabbitMQ's reason and does not
1395        keep it. Reporting the block without the reason is the honest half;
1396        inventing a reason would read exactly like one the broker sent.
1397        """
1398        transport = self._transport
1399        return transport.blocked_reason if isinstance(transport, BlockedState) else None
1400
1401    async def health(self, timeout: float = DEFAULT_HEALTH_TIMEOUT) -> HealthReport:
1402        """Whether this connection can reach its broker, and its consumers run.
1403
1404        The broker half is a declaration of a temporary queue, which is the
1405        cheapest thing AMQP offers that actually proves the connection works. A
1406        TCP connection that is open but wedged — the broker paused, the network
1407        black-holing — answers a socket-level check exactly as a healthy one
1408        does, right up until something is asked of it.
1409
1410        The consumer half is the one a probe usually wants and nothing outside
1411        can see. A consumer whose workers have died without it being closed is
1412        one the broker is still sending messages to and nothing is reading, and
1413        from outside that is indistinguishable from a quiet queue.
1414
1415        **A blocked connection is up, with the reason.** The broker blocks a
1416        connection when it is low on disk or memory, and an application that
1417        fails its own readiness check for it is one an orchestrator restarts
1418        into the same blocked broker, having thrown away whatever it was
1419        holding. ``parts["blocked"]`` carries the state either way, and is
1420        ``None`` when the transport could not be asked — see :attr:`blocked`.
1421
1422        **The probe is bounded.** A blocked broker stops reading the socket, so
1423        the round trip this makes is precisely the thing that does not come back
1424        when something is wrong; without a deadline of its own a health check
1425        hangs, and a readiness endpoint that hangs takes the instance out of
1426        rotation with no report at all. A probe that runs out of time is
1427        abandoned rather than waited for, because a request that cannot be
1428        answered cannot reliably be cancelled either.
1429
1430        It costs a round trip, so it is not something to call per request. Wire
1431        it to a readiness probe and let the probe's interval decide how often.
1432
1433        :param timeout: how long to give the broker's answer, three seconds by
1434            default. Shorter than the deadline
1435            :func:`~acemq_amqp.telemetry.aggregate_health` puts around a whole
1436            set of checks, so that this report rather than that one is what an
1437            operator reads when a broker stops answering
1438        :returns: the report. Degraded means working but worth an alert
1439        """
1440        checked = datetime.now(timezone.utc)
1441        consumers = self.consumers
1442        stalled = [consumer.queue for consumer in consumers if not consumer.running]
1443        blocked = self.blocked
1444        parts: dict[str, Any] = {
1445            "consumers": len(consumers),
1446            "in-flight": sum(consumer.in_flight for consumer in consumers),
1447            "blocked": blocked,
1448        }
1449        reason = self.blocked_reason
1450        if reason:
1451            parts["blocked-reason"] = reason
1452
1453        if self._closed:
1454            return HealthReport(
1455                HealthStatus.DOWN, "the connection has been closed", checked, parts
1456            )
1457
1458        if blocked:
1459            # Asked before the probe rather than after it, because the probe is
1460            # what a blocked broker will not answer: a round trip started here
1461            # would spend the whole deadline arriving at the answer this already
1462            # has.
1463            return self._blocked_report(checked, parts, stalled, reason)
1464
1465        # Asking whether a queue nothing has ever declared exists, rather than
1466        # declaring a temporary one. It is the same round trip and it proves the
1467        # same thing, and it creates nothing at all: an exclusive queue is only
1468        # released when the channel that declared it closes, so a probe that
1469        # declared one would leave a queue on the broker for the life of the
1470        # connection and a new one behind every restart.
1471        probe = f"acemq-health-{uuid.uuid4().hex}"
1472        started = time.monotonic()
1473        try:
1474            await self._probe_within(probe, timeout)
1475        except asyncio.TimeoutError:
1476            # A round trip that does not come back is the ordinary way a block
1477            # shows itself, and the notification may have arrived while this was
1478            # waiting. Asked again rather than assumed, because "the broker is
1479            # protecting itself" and "the broker has gone" are the same silence.
1480            blocked = self.blocked
1481            parts["blocked"] = blocked
1482            reason = self.blocked_reason
1483            if reason:
1484                parts["blocked-reason"] = reason
1485            if blocked:
1486                return self._blocked_report(checked, parts, stalled, reason)
1487            return HealthReport(
1488                HealthStatus.DOWN,
1489                f"the broker did not answer within {timeout:g}s",
1490                checked,
1491                parts,
1492            )
1493        except Exception as failure:
1494            parts["error"] = _describe(failure)
1495            return HealthReport(
1496                HealthStatus.DOWN, f"the broker did not answer: {failure}", checked, parts
1497            )
1498        parts["round-trip"] = round(time.monotonic() - started, 4)
1499
1500        if stalled:
1501            # Degraded rather than down: the connection works, and a replacement
1502            # instance would almost certainly stall the same way. Worth waking
1503            # somebody; not worth taking this one out of rotation.
1504            return HealthReport(
1505                HealthStatus.DEGRADED,
1506                _stalled_detail(stalled),
1507                checked,
1508                {**parts, "stalled": sorted(stalled)},
1509            )
1510        return HealthReport(HealthStatus.UP, "", checked, parts)
1511
1512    def _blocked_report(
1513        self,
1514        checked: datetime,
1515        parts: dict[str, Any],
1516        stalled: list[str],
1517        reason: str | None,
1518    ) -> HealthReport:
1519        """What a blocked connection reports: up, with the reason.
1520
1521        Up for the block itself, and still degraded when a consumer of this
1522        process has stopped reading — that is this instance's own failure rather
1523        than the broker's, it outlives the alarm, and a block is not a reason to
1524        stop saying so.
1525        """
1526        detail = self._blocked_detail(reason)
1527        if stalled:
1528            return HealthReport(
1529                HealthStatus.DEGRADED,
1530                f"{_stalled_detail(stalled)}; and {detail}",
1531                checked,
1532                {**parts, "stalled": sorted(stalled)},
1533            )
1534        return HealthReport(HealthStatus.UP, detail, checked, parts)
1535
1536    @staticmethod
1537    def _blocked_detail(reason: str | None) -> str:
1538        """The blocked sentence, with the broker's reason when there is one.
1539
1540        The reason follows a colon, which is how Go and Ruby write it too —
1541        ``blockedPrefix + ": " + reason`` and ``"#{BLOCKED}: #{reason}"``. One
1542        shape across the family means one alert rule reads a blocked broker
1543        whatever language the service happens to be written in.
1544        """
1545        return BLOCKED_DETAIL if not reason else f"{BLOCKED_DETAIL}: {reason}"
1546
1547    async def _probe_within(self, name: str, timeout: float) -> None:
1548        """One round trip to the broker, given ``timeout`` to come back.
1549
1550        The probe runs as a task that is abandoned rather than awaited when the
1551        deadline passes, which is the difference between a bounded check and one
1552        that merely looks bounded: cancelling a request to a broker that has
1553        stopped reading its socket means waiting for a cancellation that travels
1554        the same way the request did.
1555
1556        :raises asyncio.TimeoutError: when the broker did not answer in time
1557        """
1558        probe = asyncio.ensure_future(self._probe(name))
1559        try:
1560            done, _ = await asyncio.wait({probe}, timeout=timeout)
1561        except asyncio.CancelledError:
1562            # The caller gave up on the whole check. Waiting on a task is not
1563            # cancelling it, so it is cancelled here rather than left running
1564            # against a broker nobody is listening to any more.
1565            probe.cancel()
1566            probe.add_done_callback(_retrieved)
1567            raise
1568        if not done:
1569            probe.cancel()
1570            # Whatever it ends as is nobody's news now, and an abandoned task
1571            # whose exception is never retrieved is a warning at collection time
1572            # about a failure that has already been reported as a timeout.
1573            probe.add_done_callback(_retrieved)
1574            raise asyncio.TimeoutError
1575        # Raises what the probe raised, which is what the caller reports.
1576        probe.result()
1577
1578    async def _probe(self, name: str) -> None:
1579        """One round trip to the broker that leaves nothing behind.
1580
1581        A transport that cannot be asked whether a queue exists gets the next
1582        cheapest thing: a temporary queue, removed again straight away. Both
1583        prove the same thing, which is that the connection answers rather than
1584        merely being open.
1585
1586        The temporary queue is classic, and has to be. It is exclusive and
1587        auto-deleting so that it goes when the channel does, and RabbitMQ
1588        refuses a quorum queue that is either — a health check that could not be
1589        declared would report a healthy broker as down.
1590        """
1591        if isinstance(self._transport, QueueAdmin):
1592            await self._transport.queue_exists(name)
1593            return
1594        await self._transport.declare_queue(
1595            name, QueueSpec(durable=False, auto_delete=True, exclusive=True)
1596        )
1597
1598    @property
1599    def publish_interceptors(self) -> tuple[PublishInterceptor, ...]:
1600        """What wraps every publish, outermost first.
1601
1602        A snapshot rather than the list itself, so a publish in progress cannot
1603        be handed a chain that is being edited underneath it.
1604        """
1605        return tuple(self._publishing)
1606
1607    @property
1608    def consume_interceptors(self) -> tuple[ConsumeInterceptor, ...]:
1609        """What wraps every handler, outermost first."""
1610        return tuple(self._consuming)
1611
1612    def intercept_publish(self, interceptor: PublishInterceptor) -> Connection:
1613        """Wraps every publish on this connection in ``interceptor``.
1614
1615        Registered ones run in the order they were added, so this one is
1616        innermost until the next is added. Read at publish time rather than
1617        copied into each publisher, so an interceptor registered during start-up
1618        applies to publishers that already exist — which is what makes it usable
1619        from a framework's wiring rather than only from the line that connects.
1620
1621        Adding at start-up is still what to do. A message already in flight will
1622        not see it, and an interceptor whose whole job is to stamp every message
1623        is not doing it if the first few went without.
1624
1625        :param interceptor: what to wrap the publish in. See
1626            :mod:`acemq_amqp.interceptors`
1627        :returns: this connection, so registrations read as one expression
1628        """
1629        self._publishing.append(interceptor)
1630        return self
1631
1632    def intercept_consume(self, interceptor: ConsumeInterceptor) -> Connection:
1633        """Wraps every handler on this connection in ``interceptor``.
1634
1635        :param interceptor: what to wrap the handler in
1636        :returns: this connection
1637        """
1638        self._consuming.append(interceptor)
1639        return self
1640
1641    async def declare(self, topology: Topology) -> None:
1642        """Applies a topology to this connection's broker.
1643
1644        :param topology: what the broker should have
1645        """
1646        await topology.apply(self._transport)
1647
1648    def publisher(
1649        self,
1650        exchange: str = "",
1651        routing_key: str = "",
1652        *,
1653        codec: Codec | None = None,
1654        persistent: bool = True,
1655        mandatory: bool = False,
1656    ) -> Publisher:
1657        """Builds a publisher for an exchange and routing key.
1658
1659        Publish straight to a queue by leaving the exchange empty: the default
1660        exchange routes to the queue whose name matches the routing key.
1661
1662        :param exchange: where to publish, empty for the default exchange
1663        :param routing_key: what to publish under, or a queue name
1664        :param codec: a codec other than this connection's
1665        :param persistent: ask the broker to write these messages to disk
1666        :param mandatory: fail when a message reaches no queue at all, rather
1667            than letting the broker drop it silently
1668        :returns: the publisher
1669        """
1670        return Publisher(
1671            self,
1672            exchange,
1673            routing_key,
1674            codec=codec,
1675            persistent=persistent,
1676            mandatory=mandatory,
1677        )
1678
1679    async def consume(
1680        self,
1681        queue: str,
1682        handler: Handler,
1683        *,
1684        codec: Codec | None = None,
1685        retry: RetryPolicy | None = None,
1686        prefetch: int | None = None,
1687        concurrency: int = 1,
1688        tag: str = "",
1689        args: Mapping[str, Any] | None = None,
1690        declare: bool = True,
1691    ) -> Consumer:
1692        """Reads messages from a queue until the returned consumer is closed.
1693
1694        Before it subscribes it declares the queues it will need when a message
1695        fails: ``{queue}.dlq``, ``{queue}.parked``, ``acemq.dlx`` and, when the
1696        retry policy has waits the broker holds, ``acemq.retry`` and the rungs.
1697        See :func:`acemq_amqp.topology.declare_where_failures_go` for exactly
1698        what and why. The source queue is not among them, and neither is
1699        anything a producer needs, so this is not a replacement for declaring a
1700        topology — it is the floor underneath a service that was deployed
1701        without one.
1702
1703        :param queue: what to read
1704        :param handler: what to do with a message; returning an
1705            :class:`~acemq_amqp.Ack` is the decision
1706        :param codec: a codec other than this connection's
1707        :param retry: a policy other than this connection's
1708        :param prefetch: how many unacknowledged messages to hold
1709        :param concurrency: how many messages to work on at once. One by
1710            default, which keeps a queue's messages in order; raising it trades
1711            that order for throughput, which is the right trade for handlers
1712            that spend their time waiting on something else
1713        :param tag: what to call this consumer to the broker
1714        :param args: broker-specific consumer arguments
1715        :param declare: declare the queues above before subscribing. On by
1716            default, because the alternative default loses messages silently.
1717            Turn it off for a login with no ``configure`` permission on the
1718            vhost, which is refused by the broker rather than ignored, and for a
1719            tool draining a queue it does not own — a one-off reader of
1720            ``orders.dlq`` has no business creating ``orders.dlq.dlq``. A
1721            consumer started this way still publishes to those queues, so
1722            somebody else has to have declared them
1723        :returns: the running consumer
1724        """
1725        if self._closed:
1726            raise AceMQError("acemq: this connection is closed")
1727
1728        consumer = Consumer(
1729            self,
1730            queue,
1731            handler,
1732            codec=codec or self._codec,
1733            retry=retry or self._retry,
1734            concurrency=concurrency,
1735            declare=declare,
1736        )
1737        self._consumers.append(consumer)
1738        try:
1739            await consumer._start(
1740                self._prefetch if prefetch is None else prefetch,
1741                tag,
1742                args or {},
1743            )
1744        except BaseException:
1745            self._untrack(consumer)
1746            raise
1747        return consumer
1748
1749    async def publish_raw(
1750        self, exchange: str, routing_key: str, message: Outbound
1751    ) -> PublishResult:
1752        """Sends bytes that are already encoded, with headers already rendered.
1753
1754        For the retry and dead-letter hops, and for anything else replaying a
1755        message recorded earlier: the payload's class may not exist any more,
1756        and re-encoding it would produce different bytes from the ones that were
1757        committed. Ordinary publishing goes through :class:`Publisher`.
1758
1759        **This is the one place the outstanding-publish bound is taken.** Every
1760        publish in the library comes through here — a :class:`Publisher`, a
1761        retry rung, a dead letter, a replay — so one permit counts them all, and
1762        :meth:`Publisher.send_all` cannot put more messages on the wire at once
1763        than :attr:`max_outstanding_publishes` allows however long the list it
1764        was given is.
1765
1766        :param exchange: where to publish, empty for the default exchange
1767        :param routing_key: what to publish under
1768        :param message: the message, encoded
1769        :returns: what the broker said
1770        :raises PublishingPausedError: when the broker has blocked this
1771            connection. Nothing was sent; retry once it unblocks
1772        :raises PublishError: when every permit is taken and none came free
1773            within :attr:`confirm_timeout`, or when the broker did not confirm
1774            this publish within it. The second may or may not have arrived
1775        """
1776        # Refused rather than queued, before a permit or a byte is spent. A
1777        # blocked broker stops reading the socket, so a publish sent now sits in
1778        # the client's write buffer with no confirm coming until the alarm
1779        # clears, then times out as a failure that may or may not have arrived.
1780        # Go and .NET refuse at the same point for the same reason. ``None``
1781        # means the state could not be read and is not treated as blocked.
1782        if self.blocked is True:
1783            reason = self.blocked_reason
1784            raise PublishingPausedError(
1785                message.message_id,
1786                exchange,
1787                routing_key,
1788                (
1789                    f"{self._blocked_detail(reason)}. Nothing was sent; back off and "
1790                    "retry once the broker unblocks"
1791                ),
1792            )
1793
1794        # Taken before the message is written and not after, which is the whole
1795        # point: a bound applied afterwards has already let the message into
1796        # memory. The permit is given back in the ``finally`` below on every
1797        # path there is — the publish returning, the publish raising, and this
1798        # task being cancelled while the broker is still thinking — because a
1799        # permit lost on any one of those is a connection that publishes a
1800        # thousand more messages and then stops for ever.
1801        try:
1802            await asyncio.wait_for(
1803                self._outstanding.acquire(), self._confirm_timeout.total_seconds()
1804            )
1805        except asyncio.TimeoutError as stalled:
1806            # Said rather than waited out. A publisher parked behind a broker
1807            # that has stopped answering is indistinguishable from a quiet
1808            # service, and this sentence is the one Java raises word for word so
1809            # that an operator reading a log has one thing to recognise.
1810            raise PublishError(
1811                message.message_id,
1812                exchange,
1813                routing_key,
1814                (
1815                    f"{self._max_outstanding} publishes are already waiting for a "
1816                    f"confirm and none completed within {self._confirm_timeout}. "
1817                    "The broker is not keeping up; publish more slowly rather than "
1818                    "buffering more."
1819                ),
1820            ) from stalled
1821
1822        self._in_flight_publishes += 1
1823        try:
1824            # Bounded as well, by the same timeout. RabbitMQ says a connection is
1825            # blocked only once it publishes, so the first publish after an alarm
1826            # is written and then held with no confirm until the alarm clears --
1827            # minutes, or never. Said here as a failure, not a refusal: it was
1828            # written, so it may have arrived. The publish after it sees the
1829            # block and is refused above. .NET bounds the same wait the same way.
1830            result = await asyncio.wait_for(
1831                self._transport.publish(exchange, routing_key, message),
1832                self._confirm_timeout.total_seconds(),
1833            )
1834        except asyncio.TimeoutError as unconfirmed:
1835            raise PublishError(
1836                message.message_id,
1837                exchange,
1838                routing_key,
1839                (
1840                    f"the broker did not confirm this publish within "
1841                    f"{self._confirm_timeout}. It may or may not have been stored, "
1842                    "so treat it as neither sent nor lost"
1843                ),
1844            ) from unconfirmed
1845        finally:
1846            self._in_flight_publishes -= 1
1847            self._outstanding.release()
1848
1849        # A nack is the broker declining to keep a message it was asked to take, and
1850        # it is raised here rather than returned.
1851        #
1852        # It used to be returned, and `confirmed` was read nowhere in this library
1853        # except a telemetry label -- so a caller who did not inspect the result
1854        # carried on believing the message had gone somewhere, and Java, .NET, Go and
1855        # Ruby all raised on the same event. Worse than the wrong belief: the outbox
1856        # relay removes a record because the publish did not raise, so a nack deleted
1857        # a record for a message the broker had refused. At-most-once, in the pattern
1858        # whose whole purpose is not losing anything.
1859        #
1860        # Checked here because this is the one place every publish in the library
1861        # comes through -- a Publisher, a retry rung, a dead letter, a replay, the
1862        # outbox -- so one check covers all of them, and anything that settles or
1863        # deletes on the strength of a publish returning now hears the refusal.
1864        #
1865        # Distinct from unroutable, which the Publisher path reports separately and
1866        # which sets `unroutable`: the broker could route this one and would not keep
1867        # it, and the two want different responses.
1868        #
1869        # So a returned message is passed back rather than raised here. The
1870        # RabbitMQ transport reports a Basic.Return as not confirmed as well as
1871        # not routed, and raising it as a nack hid ``routed`` from every caller
1872        # that reads it: a mandatory Publisher said "refused" rather than
1873        # "reached no queue", and the retry and dead-letter hops never reached
1874        # the fallbacks written for a queue that is not there.
1875        if not result.confirmed and result.routed:
1876            raise PublishError(
1877                message.message_id,
1878                exchange,
1879                routing_key,
1880                (
1881                    "the broker refused to confirm it. The message reached the broker "
1882                    "and the broker has not taken responsibility for it, so it must be "
1883                    "treated as not sent"
1884                ),
1885            )
1886
1887        return result
1888
1889    async def pull(self, queue: str) -> Delivery | None:
1890        """Takes one message off a queue, or ``None`` when there is none waiting.
1891
1892        For tools rather than for services: it is a round trip per message where
1893        :meth:`consume` gets a stream. What it can do that a consumer cannot is
1894        say that a queue is empty, which is what anything working through a
1895        backlog and then stopping — a replay, a drain — has to know.
1896
1897        The delivery comes back unsettled, so the caller decides whether it is
1898        gone or goes back.
1899
1900        :param queue: what to read from
1901        :returns: the delivery, or ``None``
1902        """
1903        return await self._source().pull(queue)
1904
1905    async def queue_exists(self, queue: str) -> bool:
1906        """Whether a queue is on the broker, creating nothing."""
1907        return await self._admin().queue_exists(queue)
1908
1909    async def message_count(self, queue: str) -> int:
1910        """How many messages are waiting on a queue."""
1911        return await self._admin().message_count(queue)
1912
1913    async def delete_queue(self, queue: str) -> None:
1914        """Removes a queue and every message still on it."""
1915        await self._admin().delete_queue(queue)
1916
1917    def _admin(self) -> QueueAdmin:
1918        if not isinstance(self._transport, QueueAdmin):
1919            raise AceMQError(
1920                f"acemq: the {type(self._transport).__name__} transport cannot manage queues"
1921            )
1922        return self._transport
1923
1924    def _source(self) -> MessageSource:
1925        if not isinstance(self._transport, MessageSource):
1926            raise AceMQError(
1927                f"acemq: the {type(self._transport).__name__} transport cannot be "
1928                "asked for one message at a time"
1929            )
1930        return self._transport
1931
1932    def _untrack(self, consumer: Consumer) -> None:
1933        if consumer in self._consumers:
1934            self._consumers.remove(consumer)
1935
1936    async def close(self, *, timeout: float | None = DEFAULT_DRAIN_TIMEOUT) -> bool:
1937        """Stops every consumer on this connection and releases it.
1938
1939        Consumers first, and their handlers given up to ``timeout`` to finish,
1940        so a message being worked on when this is called is settled rather than
1941        returned to the queue for somebody else to redo. The bound is one
1942        deadline for every handler of every consumer, not one each: they are
1943        all drained at once. A handler still running at the deadline is
1944        cancelled and its message left unsettled, for the broker to redeliver —
1945        see :meth:`Consumer.close`.
1946
1947        The connection is released whatever happens, including when the caller
1948        cancels closing; that cancellation is raised again afterwards.
1949
1950        :param timeout: seconds to wait for the running handlers; ``None``
1951            waits for as long as they take
1952        :returns: ``True`` if every handler finished, ``False`` if the time ran
1953            out first
1954        """
1955        if self._closed:
1956            return True
1957        self._closed = True
1958        try:
1959            drained = await asyncio.gather(
1960                *(consumer.close(timeout=timeout) for consumer in list(self._consumers)),
1961                return_exceptions=True,
1962            )
1963        finally:
1964            await self._transport.close()
1965        for outcome in drained:
1966            if isinstance(outcome, BaseException):
1967                raise outcome
1968        return all(drained)
1969
1970    async def __aenter__(self) -> Connection:
1971        return self
1972
1973    async def __aexit__(self, *_: object) -> None:
1974        await self.close()