acemq_amqp.transport

  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"""What a broker has to provide, described without naming one.
 16
 17Everything above this module — the envelope, the codecs, the retry engine, the
 18topology — is written against these types rather than against a client library.
 19That is what makes a fake transport in a test exercise the same code path as
 20RabbitMQ does in production, rather than a second implementation of it that can
 21drift.
 22
 23The types are deliberately thin. A transport moves bytes and settles deliveries;
 24it does not know what an envelope is, which is why :class:`Outbound` carries a
 25plain header map and :class:`Delivery` hands one back.
 26"""
 27
 28from __future__ import annotations
 29
 30from collections.abc import Awaitable, Callable, Mapping
 31from dataclasses import dataclass, field
 32from typing import Any, Protocol, runtime_checkable
 33
 34
 35@dataclass(frozen=True, slots=True)
 36class QueueSpec:
 37    """How a queue should be declared.
 38
 39    :param durable: survives a broker restart
 40    :param auto_delete: goes away when its last consumer does
 41    :param exclusive: usable only by the connection that declared it
 42    :param args: broker-specific arguments, such as ``x-dead-letter-exchange``
 43    """
 44
 45    durable: bool = True
 46    auto_delete: bool = False
 47    exclusive: bool = False
 48    args: Mapping[str, Any] = field(default_factory=dict)
 49
 50
 51@dataclass(frozen=True, slots=True)
 52class ExchangeSpec:
 53    """How an exchange should be declared.
 54
 55    :param kind: ``direct``, ``topic``, ``fanout`` or ``headers``
 56    :param durable: survives a broker restart
 57    :param auto_delete: goes away when its last binding does
 58    :param args: broker-specific arguments
 59    """
 60
 61    kind: str = "topic"
 62    durable: bool = True
 63    auto_delete: bool = False
 64    args: Mapping[str, Any] = field(default_factory=dict)
 65
 66
 67@dataclass(frozen=True, slots=True)
 68class ConsumeSpec:
 69    """How a consumer should be set up.
 70
 71    :param prefetch: how many unacknowledged messages the broker will send at
 72        once, or zero for the transport's default
 73    :param tag: what to call this consumer to the broker, which is what an
 74        operator sees when working out who is holding a message
 75    :param args: broker-specific consumer arguments, such as the
 76        ``x-stream-offset`` a stream consumer needs to say where it starts
 77    """
 78
 79    prefetch: int = 0
 80    tag: str = ""
 81    args: Mapping[str, Any] = field(default_factory=dict)
 82
 83
 84@dataclass(frozen=True, slots=True)
 85class Outbound:
 86    """A message on its way to the broker, already encoded.
 87
 88    :param body: the encoded payload
 89    :param content_type: what the body is, for the consumer's codec
 90    :param message_id: the AMQP message id, conventionally the envelope's
 91    :param reply_to: AMQP's own ``reply-to`` property, naming where an answer
 92        should be sent. Empty for the overwhelming majority of messages, which
 93        are not questions. It is here as well as in the headers because the four
 94        other libraries read the native property, and a Python responder that
 95        only ever wrote a header could not be answered by a Java or .NET caller
 96    :param headers: the envelope's headers plus the application's
 97    :param persistent: asks the broker to write the message to disk, which is
 98        not a guarantee on its own — a persistent message on a queue that is not
 99        durable still dies with the broker
100    :param mandatory: asks the broker to return the message rather than drop it
101        when no queue is bound to receive it
102    """
103
104    body: bytes
105    content_type: str = ""
106    message_id: str = ""
107    reply_to: str = ""
108    headers: Mapping[str, Any] = field(default_factory=dict)
109    persistent: bool = True
110    mandatory: bool = False
111
112
113@dataclass(frozen=True, slots=True)
114class PublishResult:
115    """What the broker said about a published message.
116
117    :param message_id: the identifier it went out with
118    :param confirmed: the broker taking responsibility for the message. Without
119        publisher confirms this is false and nothing has been promised: the
120        message reached a socket, which is not the same thing
121    :param routed: false when a mandatory message reached no queue at all. A
122        message nothing is bound to receive is not an error to the broker; it is
123        dropped, silently, which is exactly the failure worth hearing about
124    :param return_reason: the broker's explanation when ``routed`` is false
125    """
126
127    message_id: str = ""
128    confirmed: bool = False
129    routed: bool = True
130    return_reason: str = ""
131
132
133@dataclass(frozen=True, slots=True)
134class Delivery:
135    """A message as it arrived, before any codec has looked at it.
136
137    :param body: the bytes, undecoded
138    :param content_type: what the sender said the body is, or ``None`` when the
139        sender said nothing — which is a different thing from an empty string
140        and is why the codecs distinguish them
141    :param routing_key: the key it arrived under
142    :param message_id: the AMQP message id
143    :param headers: everything the sender wrote, reserved names included
144    :param redelivered: the broker saying it has handed this message over
145        before. It is the broker's own view and says nothing about which attempt
146        this is: a requeue returns the bytes the broker was given, so the
147        attempt header still reads whatever the publisher wrote
148    :param ack: confirms the message and removes it from the queue
149    :param nack: returns the message, requeued or not
150    :param reply_to: AMQP's own ``reply-to`` property, empty when the sender did
151        not set one. A transport that cannot report it leaves it empty and the
152        responder falls back to the header, which is the only reason this has a
153        default at all
154    """
155
156    body: bytes
157    content_type: str | None
158    routing_key: str
159    message_id: str
160    headers: Mapping[str, Any]
161    redelivered: bool
162    ack: Callable[[], Awaitable[None]]
163    nack: Callable[[bool], Awaitable[None]]
164    reply_to: str = ""
165
166
167class Subscription(Protocol):
168    """A running consumer on a transport.
169
170    Stopping and releasing are two steps rather than one because a message that
171    has been delivered still has to be settled, and on AMQP a settlement travels
172    on the channel the delivery came in on. Closing that channel while a handler
173    is still working means acknowledging on a channel that has gone — which
174    RabbitMQ answers by closing the whole connection, taking down every other
175    consumer and publisher on it.
176    """
177
178    async def stop(self) -> None:
179        """Stops delivery, leaving what has already arrived settleable.
180
181        It must not return until the transport has promised no further
182        deliveries, because that promise is what makes it safe for the consumer
183        above to shut its workers down.
184        """
185
186    async def close(self) -> None:
187        """Releases what the subscription was holding.
188
189        Called once nothing is left unsettled.
190        """
191
192
193@runtime_checkable
194class Transport(Protocol):
195    """A connection to a broker.
196
197    Async because that is the shape a broker client has: every one of these is a
198    round trip, and a library that hid them behind blocking calls would make one
199    slow broker stall a whole process. The blocking API in
200    :mod:`acemq_amqp.sync` is built on top of this rather than beside it.
201    """
202
203    async def declare_queue(self, name: str, spec: QueueSpec) -> None:
204        """Creates a queue if it is not already there."""
205
206    async def declare_exchange(self, name: str, spec: ExchangeSpec) -> None:
207        """Creates an exchange if it is not already there."""
208
209    async def bind(self, queue: str, exchange: str, routing_key: str) -> None:
210        """Routes messages matching a key from an exchange to a queue."""
211
212    async def publish(
213        self, exchange: str, routing_key: str, message: Outbound
214    ) -> PublishResult:
215        """Sends one message and reports what the broker said about it."""
216
217    async def consume(
218        self,
219        queue: str,
220        spec: ConsumeSpec,
221        deliver: Callable[[Delivery], Awaitable[None]],
222    ) -> Subscription:
223        """Delivers messages to ``deliver`` until the subscription is closed."""
224
225    async def close(self) -> None:
226        """Releases the connection."""
227
228
229@runtime_checkable
230class BlockedState(Protocol):
231    """A transport that can say whether the broker has blocked its connection.
232
233    RabbitMQ sends ``connection.blocked`` when it is low on memory or disk and
234    ``connection.unblocked`` when the alarm clears. In between it stops reading
235    from the socket, so everything sent on that connection — a publish, a
236    declaration, a health probe — waits rather than failing. From outside that
237    is indistinguishable from a broker that has silently gone away, and the two
238    want opposite responses: a blocked broker is one to wait for, a dead one is
239    one to fail over from.
240
241    Separate from :class:`Transport` because not every transport has an answer.
242    A fake in a test is never blocked and can say so; a client library that
243    keeps the state to itself cannot be asked at all, which is what ``None``
244    below is for. It is the same shape as Java's ``isBlocked()`` and
245    ``blockedReason()``, .NET's ``IsBlocked``/``BlockedReason`` and Go's
246    ``BlockedReason()``.
247    """
248
249    @property
250    def blocked(self) -> bool | None:
251        """Whether the broker has blocked this connection.
252
253        ``None`` rather than ``False`` when the question could not be asked at
254        all — a client that does not expose the state, a connection that is
255        between reconnections. The two are different facts and a report that
256        folds them together says a broker is fine when nobody looked.
257        """
258
259    @property
260    def blocked_reason(self) -> str | None:
261        """What the broker said when it blocked this connection.
262
263        ``None`` when it is not blocked, and also ``None`` when the client read
264        the reason and did not keep it — which is what aio-pika does, and why
265        this is separate from :attr:`blocked` rather than a string that is empty
266        when there is nothing to say. A reason invented here would read exactly
267        like one RabbitMQ sent.
268        """
269
270
271@runtime_checkable
272class MessageSource(Protocol):
273    """A transport that can be asked for one message instead of subscribed to.
274
275    Separate from :class:`Transport` because it answers a question a
276    subscription cannot: *is there anything left*. A consumer is told when a
277    message arrives and never told that none will, so a tool that has to work
278    through what is on a queue and then stop — a replay, a drain, a one-off
279    inspection — cannot be built on one.
280
281    It is for tools rather than for services. Pulling one message at a time is a
282    round trip per message where a consumer gets a stream, so a service built on
283    this is a slow service.
284    """
285
286    async def pull(self, queue: str) -> Delivery | None:
287        """Takes the message at the head of a queue, unsettled.
288
289        :param queue: what to read from
290        :returns: the delivery, or ``None`` when the queue has nothing waiting.
291            It comes back unsettled, so a caller that neither acknowledges nor
292            rejects it is holding it until the connection goes
293        """
294
295
296@runtime_checkable
297class QueueAdmin(Protocol):
298    """A transport that can also report on and remove queues.
299
300    Separate from :class:`Transport` because not every transport has an answer:
301    "how many messages are waiting" means nothing to one that does not hold
302    them. Asking for it is therefore a question about the transport in hand,
303    which is why the connection checks before it asks.
304    """
305
306    async def queue_exists(self, name: str) -> bool:
307        """Whether a queue is on the broker.
308
309        It must not create the queue. A declaration is the only thing AMQP
310        offers, and declaring a queue that is missing creates it — so anything
311        built on a declaration would create the very queues it was asked only to
312        look for.
313        """
314
315    async def message_count(self, name: str) -> int:
316        """How many messages are waiting on a queue.
317
318        A number for a dashboard or a test, not a decision to make in a handler:
319        it is a snapshot of a queue that is still moving, and by the time it is
320        read somewhere else it is already wrong.
321        """
322
323    async def delete_queue(self, name: str) -> None:
324        """Removes a queue and every message still on it.
325
326        For tests and for tools. A service that deletes queues is usually a
327        service that has confused a queue with a session.
328        """