acemq_amqp.interceptors

  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 seam around publishing and handling, for the work that surrounds them.
 16
 17Every organisation has something every message needs and no library can guess: a
 18tenant identifier, a trace context, an authorisation token, a schema version, a
 19correlation identifier in the logging context, a timer, a size limit. Without a
 20seam these end up copied into every call site, where one of them is eventually
 21forgotten and nobody finds out until the message that needed it is the one that
 22went without it.
 23
 24An interceptor is one function that is handed the message and the rest of the
 25work::
 26
 27    async def timed(context: ConsumeContext, handle: ConsumeNext) -> Ack:
 28        started = time.monotonic()
 29        try:
 30            return await handle(context)
 31        finally:
 32            log.info("%s took %.3fs", context.queue, time.monotonic() - started)
 33
 34    mq.intercept(timed)
 35
 36That shape rather than the pair of before-and-after hooks the Java library uses,
 37because Python already has the construct: ``try``/``finally`` runs the way out
 38in the reverse of the way in, nests correctly without anybody having to reverse
 39a list, and makes an interceptor that opens something and closes it one function
 40instead of two halves that have to agree. It composes the way ASGI middleware
 41does, which is the arrangement a Python programmer already knows.
 42
 43Interceptors run in the order they were registered: the first registered is the
 44outermost, so it sees the message first on the way in and last on the way out.
 45That matters as soon as one of them reads what another wrote.
 46
 47Refusing is raising. A publish interceptor that raises stops the publish and the
 48caller sees the exception, which is the whole point of being able to intercept
 49rather than only observe: a message that must not go out can be stopped in one
 50place rather than in every publisher. A consume interceptor that raises is
 51treated exactly as a handler that raised — retried, and then dead-lettered —
 52because the alternative is acknowledging a message nothing processed.
 53
 54What a handler answered is not what happened to the message. A handler asking
 55for another attempt when there are none left is dead-lettered, and an
 56interceptor that wants to record how a delivery really ended has to wait for the
 57consumer to say — :meth:`ConsumeContext.when_settled` is where it waits, and
 58:class:`~acemq_amqp.ack.Settlement` is what it is told.
 59
 60Everything here is built out of the public API and needs no private access. An
 61interceptor sees the envelope, the payload, the destination and the body, and
 62may change any of them; that is the same rule the patterns in
 63:mod:`acemq_amqp.patterns` follow, and it is what keeps this a seam rather than
 64a privileged back door.
 65"""
 66
 67from __future__ import annotations
 68
 69import logging
 70from collections.abc import Awaitable, Callable, Sequence
 71from dataclasses import dataclass, field
 72from typing import Any, TypeAlias
 73
 74from .ack import Ack, Settlement
 75from .envelope import Envelope
 76from .transport import PublishResult
 77
 78log = logging.getLogger("acemq")
 79
 80#: Told what the consumer did with a delivery, once it had decided and before it
 81#: had done it. Registered with :meth:`ConsumeContext.when_settled`.
 82SettlementListener: TypeAlias = Callable[[Settlement], None]
 83
 84
 85@dataclass(slots=True)
 86class PublishContext:
 87    """A message on its way out, before it is encoded.
 88
 89    Mutable on purpose, and mutable rather than rebuilt-and-returned: an
 90    interceptor that only wants to add one header should not have to reconstruct
 91    the rest, and one that reads what an earlier interceptor wrote needs to be
 92    looking at the same object rather than at a copy of an earlier version.
 93
 94    :param exchange: where it is going. Changing it redirects the message
 95    :param routing_key: what it is published under
 96    :param envelope: the metadata. Reserved header names are still refused when
 97        the envelope is rendered, whoever wrote them
 98    :param payload: what is about to be encoded, before the codec sees it, so an
 99        interceptor can change the message and not only its metadata
100    :param persistent: whether the broker is asked to write it to disk
101    :param mandatory: whether reaching no queue is an error rather than a
102        silence
103    :param reply_to: where an answer should be sent, written to AMQP's own
104        ``reply-to`` property. Empty for anything that is not a question
105    """
106
107    exchange: str
108    routing_key: str
109    envelope: Envelope
110    payload: Any
111    persistent: bool = True
112    mandatory: bool = False
113    reply_to: str = ""
114
115    def set_header(self, name: str, value: Any) -> None:
116        """Adds an application header to the message being published.
117
118        A convenience over rebuilding the envelope, which is what the three
119        commonest interceptors — a tenant, a trace context, a schema version —
120        all want to do and nothing else.
121        """
122        self.envelope = self.envelope.with_(headers={**self.envelope.headers, name: value})
123
124
125@dataclass(slots=True)
126class ConsumeContext:
127    """A message that has arrived and been decoded, before the handler sees it.
128
129    :param queue: what it was read from
130    :param envelope: the metadata, as it will reach the handler. Changing it
131        changes what the handler is given
132    :param payload: the decoded body, likewise
133    :param body: the undecoded bytes, for an interceptor that wants to see what
134        actually arrived
135    :param content_type: what the sender said the body was, or ``None``
136    :param routing_key: the key it arrived under
137    :param redelivered: the broker saying it has handed this one over before
138    :param state: somewhere for an interceptor to leave something for a later
139        one, or for its own way out. Empty, and this library never reads it
140    :param reports_settlement: whether whatever is driving this delivery will
141        say how it was settled. True when a consumer is, and false when an
142        interceptor chain is being run by something else — a test, or a caller
143        composing the chain by hand
144    :param reply_to: AMQP's own ``reply-to`` property as it arrived, empty when
145        the sender set none
146    """
147
148    queue: str
149    envelope: Envelope
150    payload: Any
151    body: bytes
152    content_type: str | None
153    routing_key: str
154    redelivered: bool
155    state: dict[str, Any] = field(default_factory=dict)
156    reports_settlement: bool = False
157    reply_to: str = ""
158    _settlement_listeners: list[SettlementListener] = field(
159        default_factory=list, repr=False
160    )
161
162    def when_settled(self, listener: SettlementListener) -> bool:
163        """Asks to be told how this delivery was settled.
164
165        What the handler answered is not what happened to the message: a
166        handler asking for a retry with no attempts left is dead-lettered, and
167        an interceptor that read only the :class:`~acemq_amqp.ack.Ack` would
168        report a retry that never happened. That gap is exactly what somebody
169        querying a trace backend for dead letters falls into, so the consumer
170        reports its decision here and the tracing adapter waits for it.
171
172        The listener is called once, on the consumer's task, after the decision
173        and before it is carried out. It should not block and must not raise;
174        one that does is logged and otherwise ignored, because a delivery still
175        has to be settled when whoever was watching it breaks.
176
177        :param listener: called with the :class:`~acemq_amqp.ack.Settlement`
178        :returns: whether anything will ever call it. ``False`` when this
179            delivery is not being driven by a consumer, in which case the
180            caller should not wait for an answer that is not coming
181        """
182        if not self.reports_settlement:
183            return False
184        self._settlement_listeners.append(listener)
185        return True
186
187    def settled(self, settlement: Settlement) -> None:
188        """Tells the listeners what happened. Called by the consumer.
189
190        :param settlement: what the consumer decided to do with the delivery
191        """
192        for listener in self._settlement_listeners:
193            try:
194                listener(settlement)
195            except Exception:
196                # The delivery still has to be settled. A listener is an
197                # observer, and an observer that breaks must not take the
198                # message with it.
199                log.exception(
200                    "acemq: a settlement listener failed for a delivery from %s", self.queue
201                )
202
203
204#: The rest of a publish: the next interceptor, or the publish itself.
205PublishNext: TypeAlias = Callable[[PublishContext], Awaitable[PublishResult]]
206
207#: The rest of a delivery: the next interceptor, or the handler itself.
208ConsumeNext: TypeAlias = Callable[[ConsumeContext], Awaitable[Ack]]
209
210#: Runs around every publish on a connection.
211#:
212#: Call ``send`` to carry on and return what it returns; do not call it to
213#: refuse, and raise to make the refusal the caller's problem rather than a
214#: message that quietly went nowhere.
215PublishInterceptor: TypeAlias = Callable[
216    [PublishContext, PublishNext], Awaitable[PublishResult]
217]
218
219#: Runs around every handler on a connection.
220#:
221#: Call ``handle`` to carry on. Raising means the handler never runs and the
222#: delivery is treated as a failed one, which is the honest outcome: the
223#: alternative is acknowledging something nothing processed.
224ConsumeInterceptor: TypeAlias = Callable[[ConsumeContext, ConsumeNext], Awaitable[Ack]]
225
226
227def publish_chain(
228    interceptors: Sequence[PublishInterceptor], publish: PublishNext
229) -> PublishNext:
230    """Folds interceptors around a publish, first registered outermost.
231
232    Built backwards because that is what nesting is: the last interceptor wraps
233    the publish, the one before wraps that, and the first ends up on the
234    outside, seeing the message before anybody and its result after everybody.
235
236    :param interceptors: what to wrap it in, in registration order
237    :param publish: the innermost work, which actually sends the message
238    :returns: the whole thing as one callable, or ``publish`` when there is
239        nothing to wrap it in
240    """
241    composed = publish
242    for interceptor in reversed(interceptors):
243        composed = _publish_step(interceptor, composed)
244    return composed
245
246
247def consume_chain(
248    interceptors: Sequence[ConsumeInterceptor], handle: ConsumeNext
249) -> ConsumeNext:
250    """Folds interceptors around a handler, first registered outermost.
251
252    :param interceptors: what to wrap it in, in registration order
253    :param handle: the innermost work, which runs the handler
254    :returns: the whole thing as one callable, or ``handle`` when there is
255        nothing to wrap it in
256    """
257    composed = handle
258    for interceptor in reversed(interceptors):
259        composed = _consume_step(interceptor, composed)
260    return composed
261
262
263def _publish_step(interceptor: PublishInterceptor, rest: PublishNext) -> PublishNext:
264    """One link, bound in a function of its own.
265
266    Not a closure written inline in the loop above: a closure made in a loop
267    captures the variable and not its value, so every link would end up calling
268    the last interceptor. It is the oldest bug in Python and it would show up
269    here as interceptors that all appear to run and only one of which does.
270    """
271
272    async def step(context: PublishContext) -> PublishResult:
273        return await interceptor(context, rest)
274
275    return step
276
277
278def _consume_step(interceptor: ConsumeInterceptor, rest: ConsumeNext) -> ConsumeNext:
279    async def step(context: ConsumeContext) -> Ack:
280        return await interceptor(context, rest)
281
282    return step