acemq_amqp.sync

  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 same library, for programs that are not running an event loop.
 16
 17Plenty of Python is not asyncio and has no reason to become asyncio because it
 18sends a message. This is a blocking API for those programs::
 19
 20    with sync.connect("amqp://localhost/") as mq:
 21        mq.publisher(routing_key="orders").send({"id": 7})
 22
 23It is a facade rather than a second implementation. A loop runs on a thread of
 24its own and every call here is handed to it, so the envelope rules, the codec
 25negotiation and the retry engine are the ones in
 26:mod:`acemq_amqp.connection` — there is one set of retry arithmetic in this
 27library, and it is not this file's.
 28
 29Handlers run on a worker thread rather than on the loop. A blocking handler
 30called from the loop would stop every other consumer on the connection, and a
 31blocking handler is exactly what somebody using this API has: that is why they
 32are here.
 33"""
 34
 35from __future__ import annotations
 36
 37import asyncio
 38import threading
 39from collections.abc import Callable, Coroutine, Iterable, Mapping
 40from concurrent.futures import ThreadPoolExecutor
 41from datetime import timedelta
 42from typing import Any, TypeAlias, TypeVar
 43
 44from .ack import Ack
 45from .codec import Codec
 46from .connection import (
 47    DEFAULT_CONFIRM_TIMEOUT,
 48    DEFAULT_DRAIN_TIMEOUT,
 49    DEFAULT_MAX_OUTSTANDING_PUBLISHES,
 50    DEFAULT_PREFETCH,
 51    Connection,
 52    Consumer,
 53    Message,
 54    Publisher,
 55)
 56from .connection import connect as _connect_async
 57from .envelope import Envelope
 58from .retry import RetryPolicy
 59from .security import Security
 60from .topology import Topology
 61from .transport import PublishResult
 62
 63T = TypeVar("T")
 64
 65#: What a blocking handler is: a message in, a decision out.
 66SyncHandler: TypeAlias = Callable[[Message], Ack]
 67
 68
 69class _LoopThread:
 70    """An event loop on a thread, and the way to get work onto it.
 71
 72    One per connection rather than one per process, so two connections cannot
 73    starve each other and closing one does not take the other's loop with it.
 74    """
 75
 76    def __init__(self, name: str) -> None:
 77        self._loop = asyncio.new_event_loop()
 78        self._ready = threading.Event()
 79        self._thread = threading.Thread(target=self._run, name=name, daemon=True)
 80        self._thread.start()
 81        self._ready.wait()
 82
 83    def _run(self) -> None:
 84        asyncio.set_event_loop(self._loop)
 85        self._loop.call_soon(self._ready.set)
 86        self._loop.run_forever()
 87
 88    def run(self, work: Coroutine[Any, Any, T]) -> T:
 89        """Runs a coroutine on the loop and waits for its answer here."""
 90        return asyncio.run_coroutine_threadsafe(work, self._loop).result()
 91
 92    def stop(self) -> None:
 93        """Stops the loop and waits for its thread.
 94
 95        Joined rather than left to the daemon flag, so that a connection closed
 96        in a test is really gone by the time the next test starts rather than
 97        still holding a socket.
 98        """
 99        self._loop.call_soon_threadsafe(self._loop.stop)
100        self._thread.join()
101        self._loop.close()
102
103
104class SyncPublisher:
105    """A :class:`~acemq_amqp.Publisher` you can call without awaiting."""
106
107    def __init__(self, loop: _LoopThread, publisher: Publisher) -> None:
108        self._loop = loop
109        self._publisher = publisher
110
111    def send(
112        self,
113        payload: Any,
114        *,
115        envelope: Envelope | None = None,
116        routing_key: str | None = None,
117    ) -> PublishResult:
118        """Publishes one message and waits for the broker to answer.
119
120        :param payload: what to send
121        :param envelope: metadata to send instead of a fresh one
122        :param routing_key: what to publish under, overriding the publisher's
123        :returns: what the broker said
124        """
125        return self._loop.run(
126            self._publisher.send(payload, envelope=envelope, routing_key=routing_key)
127        )
128
129    def send_all(self, payloads: Iterable[Any]) -> list[PublishResult]:
130        """Publishes a batch and waits for every confirm.
131
132        Every message goes out before any confirm is awaited, which is the
133        whole point of it: a loop calling :meth:`send` pays a broker round trip
134        per message. See :meth:`acemq_amqp.Publisher.send_all` for what a
135        partial failure reports and why it is not atomic.
136
137        :param payloads: what to send, in order
138        :returns: what the broker said about each, in the order the payloads
139            were given
140        :raises PublishError: when any message was not confirmed
141        """
142        return self._loop.run(self._publisher.send_all(payloads))
143
144
145class SyncConsumer:
146    """A running subscription. Close it to stop."""
147
148    def __init__(
149        self, connection: SyncConnection, consumer: Consumer, workers: ThreadPoolExecutor
150    ) -> None:
151        self._connection = connection
152        self._consumer = consumer
153        self._workers = workers
154
155    @property
156    def queue(self) -> str:
157        """The queue this consumer reads."""
158        return self._consumer.queue
159
160    def close(self, *, timeout: float | None = DEFAULT_DRAIN_TIMEOUT) -> bool:
161        """Stops the consumer and waits, for up to ``timeout``, for the handlers
162        already running.
163
164        Delivery stops first, and a message that had been delivered but not
165        started is given back to the broker. A handler that finishes in time is
166        settled as usual — see :meth:`acemq_amqp.Consumer.close`.
167
168        **A handler still running at the deadline is abandoned, not stopped.**
169        Python cannot cancel a thread, so it keeps running until the handler
170        returns of its own accord, and whatever it returns is thrown away. Its
171        message is left unsettled — never acknowledged, never rejected, never
172        dead-lettered — so the broker hands it out again once the channel goes:
173        a handler that was cut off may have done its work, and the message may
174        be handled twice, which is the at-least-once promise and not a loss.
175        The settlement belongs to the consumer, not to the thread, so a thread
176        that finishes late has nothing to acknowledge with and cannot settle a
177        message on a channel that has closed or been replaced.
178
179        The abandoned thread is a normal, non-daemon thread: the interpreter
180        waits for it at exit. A handler that can hang should carry its own
181        timeout.
182
183        :param timeout: seconds to wait for every running handler together;
184            ``None`` waits for as long as they take
185        :returns: ``True`` if every handler finished, ``False`` if the time ran
186            out first and some were abandoned — the consumer is stopped either way
187        """
188        try:
189            return self._connection._loop.run(self._consumer.close(timeout=timeout))
190        finally:
191            # Not waited for: a thread still running here is one the deadline
192            # has already given up on.
193            self._workers.shutdown(wait=False, cancel_futures=True)
194
195    def __enter__(self) -> SyncConsumer:
196        return self
197
198    def __exit__(self, *_: object) -> None:
199        self.close()
200
201
202class SyncConnection:
203    """A connection to a broker, without an event loop in sight.
204
205    Everything on it blocks until the broker has answered, which is the point:
206    a program that is not running a loop should not have to start one to send a
207    message.
208    """
209
210    def __init__(self, loop: _LoopThread, connection: Connection) -> None:
211        self._loop = loop
212        self._connection = connection
213        self._closed = False
214        self._workers: list[ThreadPoolExecutor] = []
215
216    @property
217    def connection(self) -> Connection:
218        """The asyncio connection underneath, for code that has a loop after all."""
219        return self._connection
220
221    def declare(self, topology: Topology) -> None:
222        """Applies a topology to this connection's broker."""
223        self._loop.run(self._connection.declare(topology))
224
225    def publisher(
226        self,
227        exchange: str = "",
228        routing_key: str = "",
229        *,
230        codec: Codec | None = None,
231        persistent: bool = True,
232        mandatory: bool = False,
233    ) -> SyncPublisher:
234        """Builds a publisher for an exchange and routing key."""
235        return SyncPublisher(
236            self._loop,
237            self._connection.publisher(
238                exchange,
239                routing_key,
240                codec=codec,
241                persistent=persistent,
242                mandatory=mandatory,
243            ),
244        )
245
246    def consume(
247        self,
248        queue: str,
249        handler: SyncHandler,
250        *,
251        codec: Codec | None = None,
252        retry: RetryPolicy | None = None,
253        prefetch: int | None = None,
254        concurrency: int = 1,
255        tag: str = "",
256        args: Mapping[str, Any] | None = None,
257        declare: bool = True,
258    ) -> SyncConsumer:
259        """Reads messages from a queue until the returned consumer is closed.
260
261        The handler is called on a worker thread, not on the loop, so it may
262        block for as long as it needs to without stopping anything else. There
263        are ``concurrency`` threads and ``concurrency`` messages in flight, so
264        the number means the same thing here as it does on the async API.
265
266        :param queue: what to read
267        :param handler: what to do with a message
268        :param codec: a codec other than this connection's
269        :param retry: a policy other than this connection's
270        :param prefetch: how many unacknowledged messages to hold
271        :param concurrency: how many messages to work on at once
272        :param tag: what to call this consumer to the broker
273        :param args: broker-specific consumer arguments
274        :param declare: declare ``{queue}.dlq``, ``{queue}.parked`` and the
275            rungs before subscribing. On by default; see
276            :meth:`acemq_amqp.Connection.consume`
277        :returns: the running consumer
278        """
279        workers = ThreadPoolExecutor(
280            max_workers=concurrency, thread_name_prefix=f"acemq-{queue}"
281        )
282        self._workers.append(workers)
283
284        async def run_on_a_thread(message: Message) -> Ack:
285            loop = asyncio.get_running_loop()
286            return await loop.run_in_executor(workers, handler, message)
287
288        consumer = self._loop.run(
289            self._connection.consume(
290                queue,
291                run_on_a_thread,
292                codec=codec,
293                retry=retry,
294                prefetch=prefetch,
295                concurrency=concurrency,
296                tag=tag,
297                args=args,
298                declare=declare,
299            )
300        )
301        return SyncConsumer(self, consumer, workers)
302
303    def queue_exists(self, queue: str) -> bool:
304        """Whether a queue is on the broker, creating nothing."""
305        return self._loop.run(self._connection.queue_exists(queue))
306
307    def message_count(self, queue: str) -> int:
308        """How many messages are waiting on a queue."""
309        return self._loop.run(self._connection.message_count(queue))
310
311    def delete_queue(self, queue: str) -> None:
312        """Removes a queue and every message still on it."""
313        self._loop.run(self._connection.delete_queue(queue))
314
315    def close(self, *, timeout: float | None = DEFAULT_DRAIN_TIMEOUT) -> bool:
316        """Stops every consumer, releases the connection and stops the loop.
317
318        Every consumer's running handlers share one deadline, ``timeout``. A
319        handler still running at it is abandoned exactly as
320        :meth:`SyncConsumer.close` describes: its thread keeps running, its
321        result is discarded and its message is left unsettled for the broker
322        to redeliver.
323
324        :param timeout: seconds to wait for the running handlers; ``None``
325            waits for as long as they take
326        :returns: ``True`` if every handler finished, ``False`` if the time ran
327            out first. Closing again returns ``True``
328        """
329        if self._closed:
330            return True
331        self._closed = True
332        try:
333            return self._loop.run(self._connection.close(timeout=timeout))
334        finally:
335            self._loop.stop()
336            for workers in self._workers:
337                workers.shutdown(wait=False, cancel_futures=True)
338
339    def __enter__(self) -> SyncConnection:
340        return self
341
342    def __exit__(self, *_: object) -> None:
343        self.close()
344
345
346def connect(
347    url: str,
348    *,
349    codec: Codec | None = None,
350    origin: str | None = None,
351    retry: RetryPolicy | None = None,
352    prefetch: int = DEFAULT_PREFETCH,
353    max_outstanding_publishes: int = DEFAULT_MAX_OUTSTANDING_PUBLISHES,
354    confirm_timeout: timedelta = DEFAULT_CONFIRM_TIMEOUT,
355    security: Security | None = None,
356    **transport_options: Any,
357) -> SyncConnection:
358    """Opens a connection to a broker, blocking until it is open.
359
360    Takes the same arguments as :func:`acemq_amqp.connect` and returns the
361    blocking shape of the same thing — including ``security``, because a program
362    that is not running an event loop has exactly the same broker to reach::
363
364        with sync.connect(
365            "amqps://broker.internal:5671/",
366            security=Security(
367                certificate_authority="certs/ca.crt",
368                credentials=credentials_from_environment(),
369            ),
370        ) as mq:
371            ...
372
373    :param url: where the broker is
374    :param codec: what publishers and consumers use unless they say otherwise
375    :param origin: what to stamp on published messages
376    :param retry: what consumers use unless they say otherwise
377    :param prefetch: how many unacknowledged messages a consumer holds
378    :param max_outstanding_publishes: how many publishes may be waiting for the
379        broker at once, a thousand by default. See
380        :class:`acemq_amqp.Connection`
381    :param confirm_timeout: how long a publish waits for room before raising
382    :param security: how to verify the broker and who to log in as
383    :param transport_options: passed to the transport
384    :returns: the connection
385    """
386    loop = _LoopThread(name="acemq-loop")
387    try:
388        connection = loop.run(
389            _connect_async(
390                url,
391                codec=codec,
392                origin=origin,
393                retry=retry,
394                prefetch=prefetch,
395                max_outstanding_publishes=max_outstanding_publishes,
396                confirm_timeout=confirm_timeout,
397                security=security,
398                **transport_options,
399            )
400        )
401    except BaseException:
402        # The loop outlives a failed connect otherwise, and a thread nobody has
403        # a reference to is a thread nobody can stop.
404        loop.stop()
405        raise
406    return SyncConnection(loop, connection)