acemq_amqp.connection
Publishing and consuming, on top of a transport.
This is where the envelope, the codec and the retry policy meet a broker. It is
asyncio because a broker client is: every publish is a round trip and every
delivery arrives on its own, and a library that hid that behind blocking calls
would make one slow queue stall a whole process. acemq_amqp.sync is the
blocking API for programs that are not running a loop, and it is built on this
one rather than beside it — one implementation of the retry rules, not two.
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"""Publishing and consuming, on top of a transport. 16 17This is where the envelope, the codec and the retry policy meet a broker. It is 18asyncio because a broker client is: every publish is a round trip and every 19delivery arrives on its own, and a library that hid that behind blocking calls 20would make one slow queue stall a whole process. :mod:`acemq_amqp.sync` is the 21blocking API for programs that are not running a loop, and it is built on this 22one rather than beside it — one implementation of the retry rules, not two. 23""" 24 25from __future__ import annotations 26 27import asyncio 28import inspect 29import logging 30import socket 31import time 32import uuid 33from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence 34from dataclasses import dataclass 35from datetime import datetime, timedelta, timezone 36from typing import Any, TypeAlias 37from urllib.parse import urlsplit 38 39from . import naming 40from .ack import ( 41 OUTCOME_ACKED, 42 OUTCOME_DEAD_LETTERED, 43 OUTCOME_IN_PROGRESS, 44 OUTCOME_PARKED, 45 OUTCOME_REJECTED, 46 OUTCOME_RETRIED, 47 Ack, 48 Action, 49 FatalError, 50 Settlement, 51) 52from .codec import Codec, JsonCodec 53from .envelope import Envelope 54from .errors import AceMQError, PublishError, PublishingPausedError 55from .interceptors import ( 56 ConsumeContext, 57 ConsumeInterceptor, 58 PublishContext, 59 PublishInterceptor, 60 consume_chain, 61 publish_chain, 62) 63from .retry import ZERO, RetryPolicy, Wait, no_retry 64from .security import Security 65from .telemetry import ( 66 METRIC_CONSUME_ATTEMPTS, 67 METRIC_CONSUME_DURATION, 68 METRIC_CONSUME_IN_FLIGHT, 69 METRIC_CONSUME_TOTAL, 70 METRIC_DEAD_LETTERED_TOTAL, 71 METRIC_PUBLISH_DURATION, 72 METRIC_PUBLISH_TOTAL, 73 METRIC_RETRIED_TOTAL, 74 METRIC_RUNG_MISSING, 75 METRIC_SET_ASIDE_FAILED, 76 OUTCOME_CONFIRMED, 77 OUTCOME_FAILED, 78 OUTCOME_REFUSED, 79 OUTCOME_UNROUTABLE, 80 TAG_EXCHANGE, 81 TAG_OUTCOME, 82 TAG_QUEUE, 83 TAG_ROUTING_KEY, 84 TAG_TARGET, 85 HealthReport, 86 HealthStatus, 87 NullObserver, 88 Observer, 89) 90from .topology import Topology, declare_where_failures_go 91from .transport import ( 92 BlockedState, 93 ConsumeSpec, 94 Delivery, 95 MessageSource, 96 Outbound, 97 PublishResult, 98 QueueAdmin, 99 QueueSpec, 100 Subscription, 101 Transport, 102) 103 104log = logging.getLogger("acemq") 105 106#: How many unacknowledged messages a consumer holds unless it says otherwise. 107#: Enough that a handler is never waiting on the network, small enough that a 108#: consumer that stalls does not take a queue's worth of work down with it. 109DEFAULT_PREFETCH = 20 110 111#: How many publishes may be waiting for the broker at once, unless a connection 112#: says otherwise. 113#: 114#: Java's ``maxOutstandingPublishes`` and .NET's ``MaxOutstandingPublishes`` 115#: default to the same thousand, and the number is the same one for the same 116#: reason: enough to keep a broker busy, small enough that the memory it can cost 117#: is bounded and obvious. A publisher that wants more should say so rather than 118#: inherit it. 119DEFAULT_MAX_OUTSTANDING_PUBLISHES = 1000 120 121#: How long a publish waits for room when every permit is taken, unless a 122#: connection says otherwise. Java's ``confirmTimeout``, and the same ten seconds. 123DEFAULT_CONFIRM_TIMEOUT = timedelta(seconds=10) 124 125#: How long :meth:`Connection.health` gives the broker to answer its probe. 126#: 127#: Shorter than the five seconds :func:`~acemq_amqp.telemetry.aggregate_health` 128#: gives a whole set of checks, and deliberately: the connection's own answer for 129#: a broker that has stopped answering is the careful one — up when the broker 130#: has blocked this connection, down when it has simply gone — and an aggregate 131#: whose deadline expired first would replace it with a flat "did not answer 132#: within 5s". The inner deadline has to run out first for the outer one to have 133#: anything to report. 134DEFAULT_HEALTH_TIMEOUT = 3.0 135 136#: How long, in seconds, closing waits for handlers that are already running 137#: before it stops waiting. Twenty, which is the figure Java, .NET and Ruby use: 138#: long enough for an ordinary handler to finish, and inside the thirty seconds 139#: an orchestrator usually allows between asking a process to stop and killing 140#: it, so the connection is closed cleanly rather than torn down from outside. 141DEFAULT_DRAIN_TIMEOUT = 20.0 142 143#: What a health report says about a connection the broker has blocked. It is 144#: reported **up**, which is the one place this disagrees with a naive reading of 145#: the state: a blocked connection is the broker protecting itself from a disk or 146#: memory alarm, and an application that fails its own readiness check for it is 147#: one an orchestrator restarts into the same blocked broker, having thrown away 148#: whatever it was holding. Java's ``AceMqHealthIndicator`` and Go's health check 149#: make the same call for the same reason. 150#: 151#: **This is a fixed prefix, and it is the same sentence in every AceMQ 152#: library** — Go's ``blockedPrefix`` and Ruby's ``Health::BLOCKED`` are these 153#: exact words. An operator writes one alert rule that matches a blocked broker 154#: whatever language the service is in, so the wording is a contract rather than 155#: a phrasing. Anything the broker itself said is appended after a colon; this 156#: library cannot recover that reason (see :attr:`Connection.blocked_reason`), 157#: so on RabbitMQ the detail is the prefix alone. 158#: 159#: The reasoning above belongs in the documentation, not in the line somebody 160#: reads at three in the morning, which is why it is a comment here and not part 161#: of the string. 162BLOCKED_DETAIL = "the broker has blocked this connection; publishing is paused" 163 164 165@dataclass(frozen=True, slots=True) 166class Message: 167 """A delivery that has been decoded. 168 169 :param payload: the body, read through the codec 170 :param envelope: the metadata that travelled with it 171 :param routing_key: the key it arrived under 172 :param content_type: what the sender said the body was, or ``None`` 173 :param redelivered: the broker saying it has handed this one over before 174 :param body: the undecoded bytes, for a handler that wants to see them 175 :param reply_to: AMQP's own ``reply-to`` property, empty when the sender set 176 none. A responder reads it as the fallback behind the 177 ``acemq-reply-to`` header, which is what lets a Java or .NET caller — 178 neither of which writes the header — be answered by a Python responder 179 """ 180 181 payload: Any 182 envelope: Envelope 183 routing_key: str 184 content_type: str | None 185 redelivered: bool 186 body: bytes 187 reply_to: str = "" 188 189 190#: What a handler is. Returning an :class:`~acemq_amqp.Ack` is the decision. 191#: 192#: Both shapes are accepted because both are honest: a handler that talks to a 193#: database over asyncio is a coroutine, and a handler that only does arithmetic 194#: should not have to pretend to be one. 195Handler: TypeAlias = Callable[[Message], "Ack | Awaitable[Ack]"] 196 197#: A handler that is definitely a coroutine function, which every handler 198#: :mod:`acemq_amqp.patterns` builds is. 199#: 200#: It is a :data:`Handler` and can be passed anywhere one can. The narrower name 201#: exists so that a caller who wants to run a wrapped handler directly — a test, 202#: usually — can await it without mypy pointing out that the wider type might not 203#: be awaitable. 204AsyncHandler: TypeAlias = Callable[[Message], Awaitable[Ack]] 205 206 207class Publisher: 208 """Sends messages to one destination. 209 210 Cheap to keep and safe to share: build one per message type at start-up 211 rather than one per message. 212 """ 213 214 def __init__( 215 self, 216 connection: Connection, 217 exchange: str, 218 routing_key: str, 219 *, 220 codec: Codec | None = None, 221 persistent: bool = True, 222 mandatory: bool = False, 223 ) -> None: 224 self._connection = connection 225 self._exchange = exchange 226 self._routing_key = routing_key 227 self._codec = codec or connection.codec 228 self._persistent = persistent 229 self._mandatory = mandatory 230 231 async def send( 232 self, 233 payload: Any, 234 *, 235 envelope: Envelope | None = None, 236 routing_key: str | None = None, 237 reply_to: str = "", 238 ) -> PublishResult: 239 """Publishes one message. 240 241 The envelope is built here when none is given: an identifier, a 242 correlation defaulting to it, this connection's origin, and now. The 243 type falls back to the routing key, which is why a publisher bound to 244 ``order.placed`` needs no further configuration to say what it sends. 245 246 :param payload: what to send, encoded by this publisher's codec 247 :param envelope: metadata to send instead of a fresh one, for a message 248 derived from another 249 :param routing_key: what to publish under, for a publisher bound to an 250 exchange rather than to one key 251 :param reply_to: where an answer should come back to, written to AMQP's 252 own ``reply-to`` property. Only :mod:`~acemq_amqp.patterns` request 253 and reply sets it, and it sets the ``acemq-reply-to`` header to the 254 same value: the header is what survives a service that rebuilds the 255 message, the property is what the other four libraries read 256 :returns: what the broker said 257 :raises PublishError: when the publisher is mandatory and the message 258 reached no queue 259 """ 260 key = self._routing_key if routing_key is None else routing_key 261 outgoing = envelope or Envelope(origin=self._connection.origin) 262 if not outgoing.origin: 263 # A message derived from another — a reply, a pipeline hop, the next 264 # stop on a routing slip — is still published by this process, and an 265 # origin naming the service before it would be a lie in a log line 266 # somebody is going to trust. 267 outgoing = outgoing.with_(origin=self._connection.origin) 268 269 context = PublishContext( 270 exchange=self._exchange, 271 routing_key=key, 272 envelope=outgoing, 273 payload=payload, 274 persistent=self._persistent, 275 mandatory=self._mandatory, 276 reply_to=reply_to, 277 ) 278 # Read here rather than at construction, so an interceptor registered 279 # during start-up applies to the publishers that already exist. Building 280 # the chain per message costs a few closures and buys that. 281 return await publish_chain(self._connection.publish_interceptors, self._send)(context) 282 283 async def send_all(self, payloads: Iterable[Any]) -> list[PublishResult]: 284 """Publishes a batch and waits for every confirm. 285 286 What most bulk publishing actually wants: the throughput of pipelining 287 with the safety of having waited. Every message goes out before any 288 confirm is awaited, and only then are all of them checked together. 289 Awaiting each send in turn is a broker round trip per message — the 290 loop a caller could have written for themselves, and none of the 291 throughput a batch exists for. 292 293 The results come back in the order the payloads were given, whatever 294 order the broker answers in. 295 296 **This is not atomic.** AMQP has no such thing: there is no way to 297 publish a hundred messages such that all or none arrive, and a library 298 that offered one would be lying. A failure does not abandon the rest 299 either — every send is awaited whatever an earlier one did, so a batch 300 that half succeeded can say how many arrived. That count is the point: 301 a caller told only "it failed" resends the messages the broker already 302 has. 303 304 How wide the batch gets on the wire is capped, and by the connection 305 rather than by this method: 306 :attr:`Connection.max_outstanding_publishes` is a thousand by default, 307 and the thousand-and-first task waits for one of the first thousand to 308 be answered before its message is written. A list of a million is 309 therefore a million tasks and a thousand messages in flight rather than 310 a million, which is backpressure instead of a memory leak — but a 311 million tasks is still a million tasks, so chunk what comes out of a 312 database rather than handing a cursor's worth of rows to one call. 313 314 :param payloads: what to send, in order, each encoded by this 315 publisher's codec 316 :returns: what the broker said about each, in the order the payloads 317 were given 318 :raises PublishError: when any message was not confirmed. Its message 319 names how many failed and how many did not, and ``__cause__`` 320 carries the first failure in payload order — which is not 321 necessarily the first one the broker answered. A batch wide enough 322 that the broker stops keeping up fails this way too, with the first 323 failure saying so 324 """ 325 batch = list(payloads) 326 # Everything is started before anything is awaited. A task per payload, 327 # all of them created before the gather below, is what puts the whole 328 # batch on the wire while the broker is still working through the 329 # first confirm. 330 sends = [asyncio.create_task(self.send(payload)) for payload in batch] 331 # return_exceptions, and not for tidiness: without it the first failure 332 # comes out of the gather while the rest of the batch is still in 333 # flight, which loses the count this method exists to report and leaves 334 # the others' exceptions unretrieved for asyncio to complain about 335 # later. 336 settled = await asyncio.gather(*sends, return_exceptions=True) 337 338 results: list[PublishResult] = [] 339 first_failure: BaseException | None = None 340 failed = 0 341 for outcome in settled: 342 if isinstance(outcome, BaseException): 343 failed += 1 344 if first_failure is None: 345 first_failure = outcome 346 else: 347 results.append(outcome) 348 349 if first_failure is not None: 350 # Named rather than printed. A failure with nothing in its message — 351 # the CancelledError every task gets when the connection closes 352 # underneath the batch is the common one — renders as an empty 353 # string, and the sentence then ends "The first failure was: " and 354 # stops, promising an explanation it does not give. The class name 355 # is not much, but it is the difference between "the sends were 356 # cancelled" and nothing at all. 357 described = str(first_failure) or type(first_failure).__name__ 358 # The counts matter. A batch that half succeeded is the ordinary 359 # outcome of a broker problem partway through, and this sentence is 360 # word for word the one Java and .NET raise, so that an operator 361 # reading a log has one thing to recognise rather than three. 362 raise PublishError( 363 "", 364 self._exchange, 365 self._routing_key, 366 described, 367 summary=( 368 f"{failed} of {len(batch)} messages were not confirmed;" 369 f" {len(results)} were. The first failure was: {described}" 370 ), 371 ) from first_failure 372 return results 373 374 async def _send(self, context: PublishContext) -> PublishResult: 375 """The innermost work: encode what the interceptors left, and send it.""" 376 observer = self._connection.observer 377 # ``routing.key`` and not ``key``: Java and .NET both tag a publish with 378 # the fully-qualified name, and a dashboard that groups by it should 379 # read the same in all five languages. 380 labels = {TAG_EXCHANGE: context.exchange, TAG_ROUTING_KEY: context.routing_key} 381 # Started before the encode, because everything from here on is time the 382 # caller spends inside send(): the codec, the wait for a permit and the 383 # broker's confirm are all things that make a publish slow, and a timer 384 # that began after the encode would hide a slow codec completely. 385 started = time.monotonic() 386 387 def record(outcome: str) -> None: 388 """The pair, always together and always with the same labels. 389 390 Two calls rather than one because the interface has two methods, and 391 in this order because a dashboard dividing the duration's count into 392 the total should never see the total lead it. 393 """ 394 observer.observe( 395 METRIC_PUBLISH_DURATION, 396 time.monotonic() - started, 397 {**labels, TAG_OUTCOME: outcome}, 398 ) 399 observer.count(METRIC_PUBLISH_TOTAL, 1, {**labels, TAG_OUTCOME: outcome}) 400 401 # Encoded here rather than before the chain ran, so an interceptor can 402 # change the payload as well as its metadata. A codec that has already 403 # run leaves an interceptor with bytes and nothing it can do to them. 404 try: 405 body = self._codec.encode(context.payload) 406 result = await self._connection.publish_raw( 407 context.exchange, 408 context.routing_key, 409 Outbound( 410 body=body, 411 content_type=self._codec.content_type, 412 message_id=context.envelope.id, 413 reply_to=context.reply_to, 414 headers=context.envelope.to_headers(routing_key=context.routing_key), 415 persistent=context.persistent, 416 mandatory=context.mandatory, 417 ), 418 ) 419 except PublishingPausedError: 420 # Declined before anything was written: nothing lost, so not 421 # ``failed``, which is the outcome an alert on loss watches. 422 record(OUTCOME_REFUSED) 423 raise 424 except Exception: 425 record(OUTCOME_FAILED) 426 raise 427 428 # Raised rather than left in the result, because a caller who does not 429 # read the result would otherwise carry on believing the message went 430 # somewhere. Unroutable is the quietest failure AMQP has: the publish 431 # succeeds, the consumer waits, and nothing anywhere says why. 432 if context.mandatory and not result.routed: 433 record(OUTCOME_UNROUTABLE) 434 raise PublishError( 435 context.envelope.id, 436 context.exchange, 437 context.routing_key, 438 result.return_reason or "no queue is bound to receive it", 439 unroutable=True, 440 ) 441 442 record(OUTCOME_CONFIRMED) 443 return result 444 445 446class _HopRefusedError(Exception): 447 """The broker refused a republish the consumer made on the user's behalf.""" 448 449 450class Consumer: 451 """A running subscription. Close it to stop. 452 453 One is built by :meth:`Connection.consume` rather than directly, because it 454 has to be started before it is any use and a half-built consumer that looks 455 finished is a thing somebody will hold on to. 456 457 Starting one declares the queues it will need when a message fails — its 458 dead-letter queue, its parking lot and its rungs — unless it was told not 459 to. See :func:`acemq_amqp.topology.declare_where_failures_go`. 460 """ 461 462 def __init__( 463 self, 464 connection: Connection, 465 queue: str, 466 handler: Handler, 467 *, 468 codec: Codec, 469 retry: RetryPolicy, 470 concurrency: int, 471 declare: bool = True, 472 ) -> None: 473 if concurrency < 1: 474 raise ValueError(f"acemq: concurrency must be at least 1, got {concurrency}") 475 self._connection = connection 476 self._queue = queue 477 self._handler = handler 478 self._codec = codec 479 self._retry = retry 480 self._concurrency = concurrency 481 self._declare = declare 482 self._work: asyncio.Queue[Delivery | None] = asyncio.Queue() 483 self._workers: list[asyncio.Task[None]] = [] 484 self._subscription: Subscription | None = None 485 self._in_flight = 0 486 self._closed = False 487 488 @property 489 def queue(self) -> str: 490 """The queue this consumer reads.""" 491 return self._queue 492 493 @property 494 def closed(self) -> bool: 495 """Whether this consumer has been stopped.""" 496 return self._closed 497 498 @property 499 def running(self) -> bool: 500 """Whether this consumer is subscribed and still has workers to hand a 501 delivery to. 502 503 What a health check reads. A consumer whose workers have all finished 504 without it being closed is a consumer the broker is still sending 505 messages to and nothing is reading — which looks identical to a quiet 506 queue from outside, and is the failure worth telling a probe about. 507 """ 508 return ( 509 not self._closed 510 and self._subscription is not None 511 and any(not worker.done() for worker in self._workers) 512 ) 513 514 @property 515 def in_flight(self) -> int: 516 """How many messages this consumer is working on right now.""" 517 return self._in_flight 518 519 @property 520 def _labels(self) -> dict[str, str]: 521 return {TAG_QUEUE: self._queue} 522 523 @property 524 def _observer(self) -> Observer: 525 return self._connection.observer 526 527 async def _start(self, prefetch: int, tag: str, args: Mapping[str, Any]) -> None: 528 if self._declare: 529 # Before the subscription rather than after it, so that the first 530 # message this consumer gives up on already has somewhere to go. A 531 # broker that refuses one of these declarations is a consumer that 532 # does not start, which is the honest failure: it would otherwise 533 # start and lose its first dead letter. 534 await declare_where_failures_go( 535 self._connection.transport, self._queue, self._retry 536 ) 537 538 self._workers = [ 539 asyncio.create_task(self._work_loop(), name=f"acemq-consumer-{self._queue}-{n}") 540 for n in range(self._concurrency) 541 ] 542 try: 543 self._subscription = await self._connection.transport.consume( 544 self._queue, 545 ConsumeSpec(prefetch=prefetch, tag=tag, args=dict(args)), 546 self._deliver, 547 ) 548 except BaseException: 549 # Nothing was subscribed, so nothing will arrive; the workers would 550 # otherwise sit on an empty queue for the life of the process. 551 for _ in self._workers: 552 self._work.put_nowait(None) 553 await asyncio.gather(*self._workers) 554 raise 555 556 async def _deliver(self, delivery: Delivery) -> None: 557 """Hands a delivery to the workers. 558 559 Queued rather than handled here, so that ``concurrency`` means what it 560 says: aio-pika starts a task per delivery and does not wait for the last 561 one, so without a queue of our own a prefetch of twenty would run twenty 562 handlers at once whatever the caller asked for. The queue needs no bound 563 because prefetch is one: the broker will not send a twenty-first message 564 until one of the twenty has been settled. 565 """ 566 self._work.put_nowait(delivery) 567 568 async def _work_loop(self) -> None: 569 while True: 570 delivery = await self._work.get() 571 if delivery is None: 572 return 573 try: 574 await self._handle(delivery) 575 except Exception: 576 # A failure here is this library's, not the handler's, and the 577 # message is left unsettled on purpose: the broker will hand it 578 # back rather than it disappearing because our own code broke. 579 log.exception("acemq: could not settle a delivery from %s", self._queue) 580 581 async def _handle(self, delivery: Delivery) -> None: 582 envelope = Envelope.from_headers(delivery.headers, delivery.routing_key) 583 # No counter for "a message arrived": every delivery is counted once 584 # when it is settled, on ``acemq.consume.total`` with the outcome, and 585 # the sum across the outcomes is how many arrived. One counter that 586 # leads the others by however many messages are in flight is a counter 587 # that makes an operator wonder which one is lying. 588 # 589 # The attempt is a distribution rather than a counter, and it is recorded 590 # here on arrival rather than at the settlement: it is a fact about the 591 # delivery that is true before the handler runs, and recording it after 592 # would drop every message this consumer is still working on when the 593 # process stops. A rising distribution is a dependency in trouble, and it 594 # says so before anything else does — the dead letters only move once the 595 # attempts run out. 596 self._observer.observe( 597 METRIC_CONSUME_ATTEMPTS, float(envelope.attempt), self._labels 598 ) 599 self._in_flight += 1 600 self._observer.gauge(METRIC_CONSUME_IN_FLIGHT, self._in_flight, self._labels) 601 try: 602 await self._handle_one(delivery, envelope) 603 finally: 604 self._in_flight -= 1 605 self._observer.gauge(METRIC_CONSUME_IN_FLIGHT, self._in_flight, self._labels) 606 607 async def _handle_one(self, delivery: Delivery, envelope: Envelope) -> None: 608 try: 609 payload = self._codec.decode(delivery.body, delivery.content_type) 610 except Exception as failure: 611 # A body that will not decode decodes no better next time, so it 612 # does not go round the retry schedule until it ages out. 613 # 614 # Parked rather than dead-lettered: a message that failed five times 615 # and a message nothing could read are different problems with 616 # different answers — one is usually the world, the other is usually 617 # a producer — and whoever drains the dead letters should not have 618 # to sort them by hand. 619 # 620 # Counted as a settled delivery here rather than in ``_carry_out``, 621 # which this path never reaches: no handler ran, so there is no 622 # duration to record, but the message did arrive and was dealt with 623 # and a total that missed it would not add up. 624 self._observer.count( 625 METRIC_CONSUME_TOTAL, 1, {**self._labels, TAG_OUTCOME: OUTCOME_PARKED} 626 ) 627 await self._park( 628 delivery, envelope, f"could not be decoded: {_describe(failure)}" 629 ) 630 return 631 632 context = ConsumeContext( 633 queue=self._queue, 634 envelope=envelope, 635 payload=payload, 636 body=delivery.body, 637 content_type=delivery.content_type, 638 routing_key=delivery.routing_key, 639 redelivered=delivery.redelivered, 640 # This delivery is being driven by a consumer, so how it ends will 641 # be reported. An interceptor that wants to know can wait for it. 642 reports_settlement=True, 643 reply_to=delivery.reply_to, 644 ) 645 646 started = time.monotonic() 647 try: 648 decision = await consume_chain( 649 self._connection.consume_interceptors, self._run_handler 650 )(context) 651 except Exception as failure: 652 # An exception is how Python says a thing failed, so a handler that 653 # raises is asking for the retry policy rather than confessing a 654 # bug — which is what Java does with the same situation, and what a 655 # handler written without reading the documentation will expect. 656 # Raising FatalError is how it says the opposite, and the retry path 657 # below reads that mark before it reads the request. 658 # 659 # An interceptor that raises lands here too, and deliberately: a 660 # refused message is retried and then dead-lettered rather than 661 # acknowledged as though something had processed it. 662 decision = Ack(Action.RETRY, failure) 663 664 # Timed around the interceptors as well as the handler, because what an 665 # operator wants to know is how long a message takes to deal with, and 666 # an interceptor that opens a transaction is part of dealing with it. 667 # Stopped here and recorded a few lines below rather than in a 668 # ``finally``, because the duration is tagged with the outcome and the 669 # outcome is not known until ``_decide`` has run: how long a message 670 # took and what happened to it are one question, and a p99 that mixes 671 # the messages that worked with the ones that timed out answers neither 672 # half of it. 673 elapsed = time.monotonic() - started 674 675 # Whatever the interceptors left, rather than what arrived: one that 676 # rewrote the envelope on the way in meant that rewrite for the retry 677 # and the dead letter as much as for the handler. 678 envelope = context.envelope 679 680 if isinstance(decision, Ack): 681 settlement, wait = self._decide(envelope, decision) 682 else: 683 settlement, wait = ( 684 Settlement( 685 OUTCOME_DEAD_LETTERED, 686 reason=( 687 f"the handler returned {type(decision).__name__} instead of an Ack" 688 ), 689 ), 690 None, 691 ) 692 693 self._observer.observe( 694 METRIC_CONSUME_DURATION, 695 elapsed, 696 {**self._labels, TAG_OUTCOME: settlement.outcome}, 697 ) 698 699 # Announced before it is carried out, and so before a consumer-side 700 # retry sleeps out its backoff. A listener told afterwards would be 701 # describing the delivery minutes after the decision, and the span it is 702 # writing on would have stayed open for a wait that is not work. 703 context.settled(settlement) 704 await self._carry_out(delivery, envelope, settlement, wait) 705 706 async def _run_handler(self, context: ConsumeContext) -> Ack: 707 """The innermost work: build the message the interceptors left, and run 708 the handler on it. 709 710 The message is built here rather than before the chain, so that an 711 interceptor which changed the envelope or the payload changed what the 712 handler is given rather than a copy nobody reads. 713 """ 714 message = Message( 715 payload=context.payload, 716 envelope=context.envelope, 717 routing_key=context.routing_key, 718 content_type=context.content_type, 719 redelivered=context.redelivered, 720 body=context.body, 721 reply_to=context.reply_to, 722 ) 723 returned = self._handler(message) 724 # Not an Ack is a mistake worth reporting, but it is not this method's 725 # to report: it is returned as it stands and _handle dead-letters it 726 # with a reason, which is where every other unhandleable delivery goes. 727 # An interceptor between here and there sees the same thing the handler 728 # returned, which is the honest thing to show it. 729 decision: Ack = await returned if inspect.isawaitable(returned) else returned 730 return decision 731 732 def _decide(self, envelope: Envelope, decision: Ack) -> tuple[Settlement, Wait | None]: 733 """What is going to happen to this delivery, before anything happens to 734 it. 735 736 Worked out on its own, and without touching the broker, so that it can 737 be announced first. The gap this closes is a small one to describe and a 738 wide one to fall into: a handler asking for a retry it has no attempts 739 left for is dead-lettered, and anything reading the handler's answer 740 instead of this one records a retry that never happened. 741 742 :param envelope: the message's metadata, for the attempt and the age 743 :param decision: what the handler answered 744 :returns: the settlement, and the wait when there is another attempt — 745 which the settlement does not carry, because whether a wait is spent 746 here or on a rung queue is this consumer's business and not a 747 listener's 748 """ 749 if decision.action is Action.ACCEPT: 750 return Settlement(OUTCOME_ACKED), None 751 752 if decision.action is Action.REJECT: 753 return ( 754 Settlement( 755 OUTCOME_REJECTED, 756 reason=f"the handler rejected it: {_describe(decision.error)}", 757 ), 758 None, 759 ) 760 761 if decision.action is Action.PARK: 762 # The handler read the message and found nothing it could read. The 763 # engine reaches the same conclusion for a body its codec refuses, 764 # and both end up in the same queue for the same reason: a message 765 # nobody can read is a producer problem, and it is only findable if 766 # it is not filed with the messages that merely ran out of luck. 767 return ( 768 Settlement( 769 OUTCOME_PARKED, 770 reason=f"the handler parked it: {_describe(decision.error)}", 771 ), 772 None, 773 ) 774 775 if decision.action is Action.IN_PROGRESS: 776 # Somebody else holds the message. Not a failure of this one, so the 777 # retry policy is not asked: it can neither spend an attempt nor run 778 # out and dead-letter a message nobody has failed to handle. 779 delay = decision.delay if decision.delay is not None else ZERO 780 return Settlement(OUTCOME_IN_PROGRESS, delay=delay), Wait(delay, in_broker=False) 781 782 if isinstance(decision.error, FatalError): 783 # The handler asked for a retry but marked the reason as one that 784 # will not change. Honouring the mark rather than the request is the 785 # point of having it. 786 return ( 787 Settlement( 788 OUTCOME_DEAD_LETTERED, 789 reason=( 790 "the handler reported an unprocessable message: " 791 f"{_describe(decision.error)}" 792 ), 793 ), 794 None, 795 ) 796 797 wait = self._retry.next_wait(envelope.attempt, envelope.age) 798 if wait is None: 799 return ( 800 Settlement( 801 OUTCOME_DEAD_LETTERED, 802 reason=f"{self._exhausted(envelope)}: {_describe(decision.error)}", 803 ), 804 None, 805 ) 806 807 return Settlement(OUTCOME_RETRIED, delay=wait.delay), wait 808 809 async def _carry_out( 810 self, 811 delivery: Delivery, 812 envelope: Envelope, 813 settlement: Settlement, 814 wait: Wait | None, 815 ) -> None: 816 """Does what :meth:`_decide` decided, and counts it as decided. 817 818 Every counter here is chosen from the :class:`~acemq_amqp.ack.Settlement` 819 and never from the handler's :class:`~acemq_amqp.ack.Ack`, which is the 820 same rule the delivery's span follows. A handler asking for a retry it 821 has no attempts left for is dead-lettered, and a counter that read the 822 request rather than the answer would report a retry for a message 823 nothing will ever try again — so the dead letters would be undercounted 824 by exactly the messages an operator most wants to find. 825 826 One ``acemq.consume.total`` per delivery, carrying the outcome, and then 827 whichever of the standalone counters applies. ``retried`` and 828 ``dead_lettered`` are both an outcome here and a counter of their own, 829 which is what Java does and is not a redundancy: the outcome says what 830 was decided about a delivery, and the counter says how many messages are 831 going round again or have been set aside, which is the number an alert 832 is written against. 833 """ 834 self._observer.count( 835 METRIC_CONSUME_TOTAL, 1, {**self._labels, TAG_OUTCOME: settlement.outcome} 836 ) 837 838 if settlement.outcome == OUTCOME_ACKED: 839 await delivery.ack() 840 return 841 842 if settlement.outcome == OUTCOME_PARKED: 843 # ``_park`` counts it, exactly as it does for a body that would not 844 # decode. There is one parked counter and one parked queue whether 845 # the codec or the handler was the one that could not read it. 846 await self._park(delivery, envelope, settlement.reason or "") 847 return 848 849 if wait is None: 850 await self._dead_letter(delivery, envelope, settlement.reason or "") 851 return 852 853 if settlement.outcome == OUTCOME_IN_PROGRESS: 854 await self._put_back(delivery, envelope, wait.delay) 855 return 856 857 if wait.in_broker and await self._retry_in_broker(delivery, envelope, wait.delay): 858 return 859 860 if wait.delay > ZERO: 861 # Waiting here holds the delivery, and so holds one of this 862 # consumer's prefetch slots. For a wait of a few seconds that is the 863 # cheaper of the two costs; for a longer one it is not, which is why 864 # the policy has a threshold and the long waits went to a rung queue 865 # a few lines above. 866 await asyncio.sleep(wait.delay.total_seconds()) 867 868 await self._retry_again(delivery, envelope, wait.delay) 869 870 def _exhausted(self, envelope: Envelope) -> str: 871 """Why there is no next attempt, in words an operator can act on.""" 872 if envelope.attempt >= self._retry.max_attempts: 873 attempts = self._retry.max_attempts 874 return f"exhausted {attempts} attempt{'' if attempts == 1 else 's'}" 875 return f"exceeded the maximum message age of {self._retry.max_message_age}" 876 877 async def _retry_in_broker( 878 self, delivery: Delivery, envelope: Envelope, delay: timedelta 879 ) -> bool: 880 """Parks the message on ``{queue}.retry.{delay}`` and lets the broker 881 return it. 882 883 The rung's ``x-message-ttl`` is the delay and its dead-letter target is 884 this queue, so the wait costs nothing here: no delivery held, no prefetch 885 slot spent, and — the reason it exists — nothing lost when this process 886 restarts halfway through. A consumer sleeping on a five-minute backoff 887 that dies at minute one does not resume at minute one; the broker 888 redelivers the unacknowledged message immediately, and the policy that 889 said five minutes delivers in none. 890 891 The obvious simplification is one rung queue and a per-message 892 ``expiration`` instead of several with fixed TTLs. It does not work. 893 RabbitMQ only ever expires the message at the *head* of a queue, so a 894 message with a ten-minute TTL at the front holds back every thirty-second 895 one behind it, and the delays that come out bear no resemblance to the 896 ones that went in. Never set a per-message TTL for this. 897 898 The attempt advances here exactly as it does on an immediate retry: the 899 counter belongs to the message, and a message that has been round the 900 broker is no less on its second attempt than one that waited here. 901 902 :returns: whether the rung took it. ``False`` means no such queue, and 903 the caller should fall back to waiting here 904 """ 905 rung = naming.retry_queue(self._queue, delay) 906 next_attempt = envelope.with_(attempt=envelope.attempt + 1) 907 try: 908 routed = await self._republish(delivery, rung, next_attempt) 909 except _HopRefusedError: 910 # The rung is there and would not take it. Requeued at once, with 911 # nothing else on this consumer settled, rather than waited out 912 # here: the broker is the one saying no, and holding the delivery 913 # for the whole delay would stall this consumer for nothing. 914 self._observer.count(METRIC_RETRIED_TOTAL, 1, {**self._labels, "where": "requeued"}) 915 await delivery.nack(True) 916 return True 917 if not routed: 918 # Kept even though a consumer declares its own rungs when it starts, 919 # because the rung can still be missing: a consumer started with 920 # ``declare=False`` declares nothing, and a rung deleted under a 921 # running consumer is gone whoever declared it. What has changed is 922 # what it means — it used to be an ordinary topology mistake and is 923 # now either a deliberate opt-out or somebody removing queues. 924 # 925 # Degraded rather than fatal: the message is still deliverable, and 926 # waiting for it here is what this library did before there were 927 # rungs. Counted as well as logged, because this is the one failure 928 # here that nothing else shows: the message is still retried and the 929 # wait still happens, so a dashboard reads as normal while the reason 930 # the rung exists is quietly gone. 931 self._observer.count(METRIC_RUNG_MISSING, 1, {**self._labels, "rung": rung}) 932 log.error( 933 "acemq: %s is not on the broker, so %s will wait %s in this consumer " 934 "instead; declare it with Topology().queue(%r, retry=policy), or let " 935 "the consumer declare it by leaving declare=True", 936 rung, 937 envelope.id, 938 delay, 939 self._queue, 940 ) 941 return False 942 943 self._observer.count(METRIC_RETRIED_TOTAL, 1, {**self._labels, "where": "broker"}) 944 log.info( 945 "acemq: retrying %s from %s, attempt %d of %d, after %s in %s", 946 envelope.id, 947 self._queue, 948 next_attempt.attempt, 949 self._retry.max_attempts, 950 delay, 951 rung, 952 ) 953 await delivery.ack() 954 return True 955 956 async def _retry_again( 957 self, delivery: Delivery, envelope: Envelope, delay: timedelta 958 ) -> None: 959 """Puts the message back on its own queue, one attempt further on. 960 961 Republished rather than requeued, because a requeue returns the bytes 962 the broker was given: the attempt header would still read what the 963 publisher wrote however many times the message had come round, and the 964 count would live only in this process's memory, which is the one place 965 it is lost when the process that has been failing restarts. 966 967 The cost is that the message goes to the back of the queue rather than 968 the front, so a retry is no longer in order with its neighbours. For a 969 message that has already failed once, that is the better trade. 970 """ 971 next_attempt = envelope.with_(attempt=envelope.attempt + 1) 972 try: 973 delivered = await self._republish(delivery, self._queue, next_attempt) 974 except _HopRefusedError: 975 delivered = False 976 if not delivered: 977 # The queue we are consuming has gone. Requeued rather than 978 # acknowledged, because dropping it here would lose a message over a 979 # broker change nobody told this consumer about. 980 # 981 # Counted as a retry all the same, and it is one: the settlement 982 # said ``retried``, the span says ``retried``, and the broker is 983 # about to hand the message back. Leaving it uncounted was the one 984 # place where the counters and the span disagreed about the same 985 # delivery. The label says which path it took, because a requeue 986 # keeps the attempt header it arrived with and the other two do not. 987 self._observer.count(METRIC_RETRIED_TOTAL, 1, {**self._labels, "where": "requeued"}) 988 log.error( 989 "acemq: cannot republish %s onto %s for attempt %d; returning it to the broker", 990 envelope.id, 991 self._queue, 992 next_attempt.attempt, 993 ) 994 await delivery.nack(True) 995 return 996 997 self._observer.count(METRIC_RETRIED_TOTAL, 1, {**self._labels, "where": "consumer"}) 998 log.info( 999 "acemq: retrying %s from %s, attempt %d of %d, after %s", 1000 envelope.id, 1001 self._queue, 1002 next_attempt.attempt, 1003 self._retry.max_attempts, 1004 delay, 1005 ) 1006 await delivery.ack() 1007 1008 async def _put_back( 1009 self, delivery: Delivery, envelope: Envelope, delay: timedelta 1010 ) -> None: 1011 """Returns a message somebody else is working on, attempt unchanged. 1012 1013 :meth:`_retry_again` without the increment: the attempt counts this 1014 message's failures, and a claim held elsewhere is not one of them. 1015 """ 1016 if delay > ZERO: 1017 await asyncio.sleep(delay.total_seconds()) 1018 try: 1019 delivered = await self._republish(delivery, self._queue, envelope) 1020 except _HopRefusedError: 1021 delivered = False 1022 if delivered: 1023 await delivery.ack() 1024 return 1025 log.error( 1026 "acemq: cannot put %s back onto %s; returning it to the broker", 1027 envelope.id, 1028 self._queue, 1029 ) 1030 await delivery.nack(True) 1031 1032 async def _park(self, delivery: Delivery, envelope: Envelope, reason: str) -> None: 1033 """Sends the message to ``{queue}.parked`` with the reason attached. 1034 1035 Where a message goes when it never reached the handler at all. Somebody 1036 has to look at it, and what they need to know first is that it was 1037 unreadable rather than unlucky. 1038 """ 1039 # The same counter a dead-lettering moves, separated by the outcome 1040 # rather than by a metric of its own — which is what Java settled on, and 1041 # what makes "how much is this queue giving up on" one number that can 1042 # then be split into the two problems it is made of. 1043 self._observer.count( 1044 METRIC_DEAD_LETTERED_TOTAL, 1, {**self._labels, TAG_OUTCOME: OUTCOME_PARKED} 1045 ) 1046 await self._set_aside(delivery, envelope, naming.parked_queue(self._queue), reason) 1047 1048 async def _dead_letter(self, delivery: Delivery, envelope: Envelope, reason: str) -> None: 1049 """Sends the message to ``{queue}.dlq`` with the reason attached, and 1050 acknowledges the original. 1051 1052 Acknowledging a message that failed looks wrong and is what makes this 1053 reliable: the message has already been safely republished somewhere 1054 else, so acknowledging the original is removing the copy that has been 1055 dealt with. Rejecting it instead would either requeue it into a hot loop 1056 or, with a dead-letter exchange configured on the queue, send it 1057 somewhere this consumer did not choose and without the reason. 1058 1059 The reason travels as an envelope field, so a consumer of the 1060 dead-letter queue reads it back through the API rather than having to 1061 know the wire header name. 1062 """ 1063 self._observer.count( 1064 METRIC_DEAD_LETTERED_TOTAL, 1065 1, 1066 {**self._labels, TAG_OUTCOME: OUTCOME_DEAD_LETTERED}, 1067 ) 1068 await self._set_aside( 1069 delivery, envelope, naming.dead_letter_queue(self._queue), reason 1070 ) 1071 1072 async def _set_aside( 1073 self, delivery: Delivery, envelope: Envelope, target: str, reason: str 1074 ) -> None: 1075 """Republishes to ``target`` with the reason, then acknowledges.""" 1076 failed = envelope.with_(error=reason) 1077 try: 1078 delivered = await self._republish(delivery, target, failed) 1079 except _HopRefusedError: 1080 delivered = False 1081 if not delivered: 1082 # Rejected rather than acknowledged: without a dead-letter queue to 1083 # put it in, the broker's own dead-lettering is the last thing left 1084 # between this message and nothing. 1085 self._observer.count( 1086 METRIC_SET_ASIDE_FAILED, 1, {**self._labels, TAG_TARGET: target} 1087 ) 1088 log.error( 1089 "acemq: cannot move %s to %s (%s); rejecting it to the broker instead", 1090 envelope.id, 1091 target, 1092 reason, 1093 ) 1094 await delivery.nack(False) 1095 return 1096 1097 log.warning( 1098 "acemq: set aside %s from %s after %d attempts, into %s: %s", 1099 envelope.id, 1100 self._queue, 1101 envelope.attempt, 1102 target, 1103 reason, 1104 ) 1105 await delivery.ack() 1106 1107 async def _republish(self, delivery: Delivery, queue: str, envelope: Envelope) -> bool: 1108 """Sends the original bytes to a queue by name, and says whether they 1109 arrived. 1110 1111 Through the default exchange, which routes to the queue whose name 1112 matches the routing key, and mandatory so that a queue that is not there 1113 is an answer rather than a silence. The body goes back exactly as it 1114 came: re-encoding through a class that has since changed would replace 1115 what was committed with something else. 1116 """ 1117 try: 1118 result = await self._connection.publish_raw( 1119 "", 1120 queue, 1121 Outbound( 1122 body=delivery.body, 1123 content_type=delivery.content_type or "", 1124 message_id=envelope.id, 1125 headers=envelope.to_headers(routing_key=delivery.routing_key), 1126 persistent=True, 1127 mandatory=True, 1128 ), 1129 ) 1130 except PublishError as refused: 1131 # The broker nacked it, or would not take it while blocked. Raised 1132 # as the caller's to settle, because what to do differs: a refused 1133 # rung means requeue now, where a missing rung means wait here. 1134 # Escaping further left the delivery unsettled for the life of the 1135 # channel, which is what Ruby did before 0.7.12. 1136 log.error("acemq: the broker refused %s onto %s: %s", envelope.id, queue, refused) 1137 raise _HopRefusedError(str(refused)) from refused 1138 return result.routed 1139 1140 async def close(self, *, timeout: float | None = DEFAULT_DRAIN_TIMEOUT) -> bool: 1141 """Stops the consumer and waits, for up to ``timeout``, for the handlers 1142 already running. 1143 1144 A message being worked on when this is called is finished and settled, 1145 rather than abandoned for the broker to hand to somebody else. A message 1146 that had been delivered but not started is given back instead: it is the 1147 broker's to hand to another consumer, and holding it here only to work 1148 through a retry delay would make closing take as long as the schedule. 1149 1150 A handler still running when the time is up is cancelled, and its message 1151 is left unsettled — neither acknowledged nor rejected — so the broker 1152 hands it out again once the channel goes. Shutdown is never the reason a 1153 message is lost or dead-lettered. The same happens when the caller 1154 cancels closing, ``asyncio.wait_for`` included, and that cancellation is 1155 raised again once the consumer has been cleaned up. 1156 1157 The subscription is released last, after everything has been settled, 1158 because a settlement travels on the channel its delivery arrived on. 1159 1160 :param timeout: seconds to wait for every running handler together; 1161 ``None`` waits for as long as they take 1162 :returns: ``True`` if every handler finished, ``False`` if the time ran 1163 out first and some were cut off — the consumer is stopped either way 1164 """ 1165 if self._closed: 1166 return True 1167 self._closed = True 1168 1169 running: set[asyncio.Task[None]] = set() 1170 try: 1171 if self._subscription is not None: 1172 await self._subscription.stop() 1173 1174 while True: 1175 try: 1176 pending = self._work.get_nowait() 1177 except asyncio.QueueEmpty: 1178 break 1179 if pending is not None: 1180 await pending.nack(True) 1181 1182 for _ in self._workers: 1183 self._work.put_nowait(None) 1184 if self._workers: 1185 _, running = await asyncio.wait(self._workers, timeout=timeout) 1186 finally: 1187 # Reached with workers still running for two reasons: the deadline 1188 # passed, or the caller cancelled us. Either way they are cancelled 1189 # before the channel goes, so nothing settles a message on a channel 1190 # that is closing — and a cancellation that came from the caller is 1191 # not caught here, so it carries on out once this has run. 1192 stragglers = [worker for worker in self._workers if not worker.done()] 1193 for worker in stragglers: 1194 worker.cancel() 1195 if stragglers: 1196 await asyncio.wait(stragglers) 1197 if self._subscription is not None: 1198 await self._subscription.close() 1199 self._connection._untrack(self) 1200 1201 for worker in self._workers: 1202 # A worker that was already cancelled — a task group unwinding, a 1203 # shutdown that cancelled the loop's tasks before closing the 1204 # connection — is not news to report to a caller who is merely 1205 # closing. One that died of its own accord is. 1206 failure = None if worker.cancelled() else worker.exception() 1207 if failure is not None: 1208 raise failure 1209 return not running 1210 1211 async def __aenter__(self) -> Consumer: 1212 return self 1213 1214 async def __aexit__(self, *_: object) -> None: 1215 # Not returned: a true value from here tells the interpreter that the 1216 # exception leaving the block was handled. 1217 await self.close() 1218 1219 1220class Connection: 1221 """A connection to a broker, and the defaults everything on it inherits. 1222 1223 Built by :func:`connect` in the ordinary case. Taking a transport directly 1224 is for a fake in a test, or for a connection made with settings a URL cannot 1225 carry. 1226 1227 :param transport: how to reach the broker 1228 :param codec: what publishers and consumers use unless they say otherwise 1229 :param origin: what to stamp on published messages, conventionally 1230 ``service@host`` 1231 :param retry: what consumers use unless they say otherwise. The default is 1232 one delivery and no second chance, so a handler that calls 1233 ``retry()`` on a connection with no policy dead-letters the message and 1234 says so in the reason. That is deliberate: the alternative default is an 1235 immediate requeue, which is a hot loop against a broker that nobody 1236 asked for 1237 :param prefetch: how many unacknowledged messages a consumer holds 1238 :param max_outstanding_publishes: how many publishes may be waiting for the 1239 broker at once. The ceiling that turns a memory leak into backpressure: 1240 without one, a caller publishing faster than the broker answers 1241 accumulates unconfirmed messages until the process dies, which looks 1242 like throughput right up to the moment it does not. A thousand by 1243 default, which is what Java's ``maxOutstandingPublishes`` and .NET's 1244 ``MaxOutstandingPublishes`` default to 1245 :param confirm_timeout: how long a publish waits for one of those permits, 1246 and then for the broker's confirm, before giving up. Ten seconds by 1247 default, the same as Java's ``confirmTimeout``. Reaching it raises 1248 rather than waiting on, because a publisher stalled for ever behind a 1249 broker that has stopped answering is the failure this bound exists to make visible 1250 :param on_publish: cross-cutting behaviour to wrap every publish on this 1251 connection in, outermost first. See :mod:`acemq_amqp.interceptors` 1252 :param on_consume: the same around every handler. 1253 :meth:`intercept_publish` and :meth:`intercept_consume` add more later 1254 :param observer: where the numbers go. Nothing by default, so a service 1255 that wants none pays for none. See :mod:`acemq_amqp.telemetry` 1256 """ 1257 1258 def __init__( 1259 self, 1260 transport: Transport, 1261 *, 1262 codec: Codec | None = None, 1263 origin: str | None = None, 1264 retry: RetryPolicy | None = None, 1265 prefetch: int = DEFAULT_PREFETCH, 1266 max_outstanding_publishes: int = DEFAULT_MAX_OUTSTANDING_PUBLISHES, 1267 confirm_timeout: timedelta = DEFAULT_CONFIRM_TIMEOUT, 1268 on_publish: Sequence[PublishInterceptor] = (), 1269 on_consume: Sequence[ConsumeInterceptor] = (), 1270 observer: Observer | None = None, 1271 ) -> None: 1272 if prefetch < 0: 1273 raise ValueError(f"acemq: prefetch must not be negative, got {prefetch}") 1274 if max_outstanding_publishes < 1: 1275 raise ValueError( 1276 "acemq: max_outstanding_publishes must be at least 1, got " 1277 f"{max_outstanding_publishes}" 1278 ) 1279 if confirm_timeout <= timedelta(0): 1280 raise ValueError( 1281 f"acemq: confirm_timeout must be positive, got {confirm_timeout}" 1282 ) 1283 self._transport = transport 1284 self._observer: Observer = observer or NullObserver() 1285 self._codec = codec or JsonCodec() 1286 self._origin = origin or default_origin() 1287 self._retry = retry or no_retry() 1288 self._prefetch = prefetch 1289 self._max_outstanding = max_outstanding_publishes 1290 self._confirm_timeout = confirm_timeout 1291 # Built here rather than on first use. An asyncio.Semaphore has bound no 1292 # loop at construction since 3.10, so it is safe to make one outside a 1293 # running loop, and making it here means every publish on this connection 1294 # shares the one counter however many tasks are publishing. 1295 self._outstanding = asyncio.Semaphore(max_outstanding_publishes) 1296 self._in_flight_publishes = 0 1297 self._consumers: list[Consumer] = [] 1298 self._publishing: list[PublishInterceptor] = list(on_publish) 1299 self._consuming: list[ConsumeInterceptor] = list(on_consume) 1300 self._closed = False 1301 1302 @property 1303 def transport(self) -> Transport: 1304 """The transport underneath, for anything this library does not wrap.""" 1305 return self._transport 1306 1307 @property 1308 def codec(self) -> Codec: 1309 """The codec publishers and consumers use unless told otherwise.""" 1310 return self._codec 1311 1312 @property 1313 def origin(self) -> str: 1314 """What published messages are stamped with.""" 1315 return self._origin 1316 1317 @property 1318 def retry(self) -> RetryPolicy: 1319 """The retry policy consumers use unless told otherwise.""" 1320 return self._retry 1321 1322 @property 1323 def observer(self) -> Observer: 1324 """Where this connection's numbers go.""" 1325 return self._observer 1326 1327 @property 1328 def max_outstanding_publishes(self) -> int: 1329 """How many publishes may be waiting for the broker at once.""" 1330 return self._max_outstanding 1331 1332 @property 1333 def confirm_timeout(self) -> timedelta: 1334 """How long a publish waits for room before giving up.""" 1335 return self._confirm_timeout 1336 1337 @property 1338 def outstanding_publishes(self) -> int: 1339 """How many publishes are waiting for the broker right now. 1340 1341 What a stalled publisher looks like from outside: a number pinned at 1342 :attr:`max_outstanding_publishes` is a broker that has stopped answering, 1343 and it says so before the first :class:`~acemq_amqp.PublishError` does. 1344 """ 1345 # Counted here rather than read off the semaphore, which keeps its count 1346 # private. Raised after the permit is taken and lowered before it is 1347 # given back, both inside publish_raw, so it never reads high. 1348 return self._in_flight_publishes 1349 1350 @property 1351 def consumers(self) -> tuple[Consumer, ...]: 1352 """The consumers running on this connection. 1353 1354 A snapshot, and public because a health check that has to say whether 1355 the consumers are alive should not need private access to find them — 1356 the same rule an interceptor follows. 1357 """ 1358 return tuple(self._consumers) 1359 1360 @property 1361 def closed(self) -> bool: 1362 """Whether this connection has been closed.""" 1363 return self._closed 1364 1365 @property 1366 def blocked(self) -> bool | None: 1367 """Whether the broker has blocked this connection, or ``None`` when the 1368 transport cannot be asked. 1369 1370 RabbitMQ blocks a connection when it is low on memory or disk, and while 1371 it lasts it stops reading that connection's socket: everything sent — 1372 a publish, a declaration, :meth:`health`'s own probe — waits rather than 1373 failing. It is the broker protecting itself and it is not an error, so 1374 it is worth telling apart from the broker having gone away, which looks 1375 identical from outside and wants the opposite response. 1376 1377 ``None`` means the question could not be asked at all, which is a 1378 different fact from ``False`` and is reported as one: a transport that 1379 does not implement 1380 :class:`~acemq_amqp.transport.BlockedState`, or one whose client has 1381 moved the state out of reach. See :attr:`blocked_reason` for why there 1382 may be a boolean and no reason to go with it. 1383 """ 1384 transport = self._transport 1385 return transport.blocked if isinstance(transport, BlockedState) else None 1386 1387 @property 1388 def blocked_reason(self) -> str | None: 1389 """What the broker said when it blocked this connection, if anything. 1390 1391 ``None`` when the connection is not blocked, and also ``None`` when the 1392 transport is blocked but cannot say why — which is what the RabbitMQ 1393 transport answers, because aio-pika reads RabbitMQ's reason and does not 1394 keep it. Reporting the block without the reason is the honest half; 1395 inventing a reason would read exactly like one the broker sent. 1396 """ 1397 transport = self._transport 1398 return transport.blocked_reason if isinstance(transport, BlockedState) else None 1399 1400 async def health(self, timeout: float = DEFAULT_HEALTH_TIMEOUT) -> HealthReport: 1401 """Whether this connection can reach its broker, and its consumers run. 1402 1403 The broker half is a declaration of a temporary queue, which is the 1404 cheapest thing AMQP offers that actually proves the connection works. A 1405 TCP connection that is open but wedged — the broker paused, the network 1406 black-holing — answers a socket-level check exactly as a healthy one 1407 does, right up until something is asked of it. 1408 1409 The consumer half is the one a probe usually wants and nothing outside 1410 can see. A consumer whose workers have died without it being closed is 1411 one the broker is still sending messages to and nothing is reading, and 1412 from outside that is indistinguishable from a quiet queue. 1413 1414 **A blocked connection is up, with the reason.** The broker blocks a 1415 connection when it is low on disk or memory, and an application that 1416 fails its own readiness check for it is one an orchestrator restarts 1417 into the same blocked broker, having thrown away whatever it was 1418 holding. ``parts["blocked"]`` carries the state either way, and is 1419 ``None`` when the transport could not be asked — see :attr:`blocked`. 1420 1421 **The probe is bounded.** A blocked broker stops reading the socket, so 1422 the round trip this makes is precisely the thing that does not come back 1423 when something is wrong; without a deadline of its own a health check 1424 hangs, and a readiness endpoint that hangs takes the instance out of 1425 rotation with no report at all. A probe that runs out of time is 1426 abandoned rather than waited for, because a request that cannot be 1427 answered cannot reliably be cancelled either. 1428 1429 It costs a round trip, so it is not something to call per request. Wire 1430 it to a readiness probe and let the probe's interval decide how often. 1431 1432 :param timeout: how long to give the broker's answer, three seconds by 1433 default. Shorter than the deadline 1434 :func:`~acemq_amqp.telemetry.aggregate_health` puts around a whole 1435 set of checks, so that this report rather than that one is what an 1436 operator reads when a broker stops answering 1437 :returns: the report. Degraded means working but worth an alert 1438 """ 1439 checked = datetime.now(timezone.utc) 1440 consumers = self.consumers 1441 stalled = [consumer.queue for consumer in consumers if not consumer.running] 1442 blocked = self.blocked 1443 parts: dict[str, Any] = { 1444 "consumers": len(consumers), 1445 "in-flight": sum(consumer.in_flight for consumer in consumers), 1446 "blocked": blocked, 1447 } 1448 reason = self.blocked_reason 1449 if reason: 1450 parts["blocked-reason"] = reason 1451 1452 if self._closed: 1453 return HealthReport( 1454 HealthStatus.DOWN, "the connection has been closed", checked, parts 1455 ) 1456 1457 if blocked: 1458 # Asked before the probe rather than after it, because the probe is 1459 # what a blocked broker will not answer: a round trip started here 1460 # would spend the whole deadline arriving at the answer this already 1461 # has. 1462 return self._blocked_report(checked, parts, stalled, reason) 1463 1464 # Asking whether a queue nothing has ever declared exists, rather than 1465 # declaring a temporary one. It is the same round trip and it proves the 1466 # same thing, and it creates nothing at all: an exclusive queue is only 1467 # released when the channel that declared it closes, so a probe that 1468 # declared one would leave a queue on the broker for the life of the 1469 # connection and a new one behind every restart. 1470 probe = f"acemq-health-{uuid.uuid4().hex}" 1471 started = time.monotonic() 1472 try: 1473 await self._probe_within(probe, timeout) 1474 except asyncio.TimeoutError: 1475 # A round trip that does not come back is the ordinary way a block 1476 # shows itself, and the notification may have arrived while this was 1477 # waiting. Asked again rather than assumed, because "the broker is 1478 # protecting itself" and "the broker has gone" are the same silence. 1479 blocked = self.blocked 1480 parts["blocked"] = blocked 1481 reason = self.blocked_reason 1482 if reason: 1483 parts["blocked-reason"] = reason 1484 if blocked: 1485 return self._blocked_report(checked, parts, stalled, reason) 1486 return HealthReport( 1487 HealthStatus.DOWN, 1488 f"the broker did not answer within {timeout:g}s", 1489 checked, 1490 parts, 1491 ) 1492 except Exception as failure: 1493 parts["error"] = _describe(failure) 1494 return HealthReport( 1495 HealthStatus.DOWN, f"the broker did not answer: {failure}", checked, parts 1496 ) 1497 parts["round-trip"] = round(time.monotonic() - started, 4) 1498 1499 if stalled: 1500 # Degraded rather than down: the connection works, and a replacement 1501 # instance would almost certainly stall the same way. Worth waking 1502 # somebody; not worth taking this one out of rotation. 1503 return HealthReport( 1504 HealthStatus.DEGRADED, 1505 _stalled_detail(stalled), 1506 checked, 1507 {**parts, "stalled": sorted(stalled)}, 1508 ) 1509 return HealthReport(HealthStatus.UP, "", checked, parts) 1510 1511 def _blocked_report( 1512 self, 1513 checked: datetime, 1514 parts: dict[str, Any], 1515 stalled: list[str], 1516 reason: str | None, 1517 ) -> HealthReport: 1518 """What a blocked connection reports: up, with the reason. 1519 1520 Up for the block itself, and still degraded when a consumer of this 1521 process has stopped reading — that is this instance's own failure rather 1522 than the broker's, it outlives the alarm, and a block is not a reason to 1523 stop saying so. 1524 """ 1525 detail = self._blocked_detail(reason) 1526 if stalled: 1527 return HealthReport( 1528 HealthStatus.DEGRADED, 1529 f"{_stalled_detail(stalled)}; and {detail}", 1530 checked, 1531 {**parts, "stalled": sorted(stalled)}, 1532 ) 1533 return HealthReport(HealthStatus.UP, detail, checked, parts) 1534 1535 @staticmethod 1536 def _blocked_detail(reason: str | None) -> str: 1537 """The blocked sentence, with the broker's reason when there is one. 1538 1539 The reason follows a colon, which is how Go and Ruby write it too — 1540 ``blockedPrefix + ": " + reason`` and ``"#{BLOCKED}: #{reason}"``. One 1541 shape across the family means one alert rule reads a blocked broker 1542 whatever language the service happens to be written in. 1543 """ 1544 return BLOCKED_DETAIL if not reason else f"{BLOCKED_DETAIL}: {reason}" 1545 1546 async def _probe_within(self, name: str, timeout: float) -> None: 1547 """One round trip to the broker, given ``timeout`` to come back. 1548 1549 The probe runs as a task that is abandoned rather than awaited when the 1550 deadline passes, which is the difference between a bounded check and one 1551 that merely looks bounded: cancelling a request to a broker that has 1552 stopped reading its socket means waiting for a cancellation that travels 1553 the same way the request did. 1554 1555 :raises asyncio.TimeoutError: when the broker did not answer in time 1556 """ 1557 probe = asyncio.ensure_future(self._probe(name)) 1558 try: 1559 done, _ = await asyncio.wait({probe}, timeout=timeout) 1560 except asyncio.CancelledError: 1561 # The caller gave up on the whole check. Waiting on a task is not 1562 # cancelling it, so it is cancelled here rather than left running 1563 # against a broker nobody is listening to any more. 1564 probe.cancel() 1565 probe.add_done_callback(_retrieved) 1566 raise 1567 if not done: 1568 probe.cancel() 1569 # Whatever it ends as is nobody's news now, and an abandoned task 1570 # whose exception is never retrieved is a warning at collection time 1571 # about a failure that has already been reported as a timeout. 1572 probe.add_done_callback(_retrieved) 1573 raise asyncio.TimeoutError 1574 # Raises what the probe raised, which is what the caller reports. 1575 probe.result() 1576 1577 async def _probe(self, name: str) -> None: 1578 """One round trip to the broker that leaves nothing behind. 1579 1580 A transport that cannot be asked whether a queue exists gets the next 1581 cheapest thing: a temporary queue, removed again straight away. Both 1582 prove the same thing, which is that the connection answers rather than 1583 merely being open. 1584 1585 The temporary queue is classic, and has to be. It is exclusive and 1586 auto-deleting so that it goes when the channel does, and RabbitMQ 1587 refuses a quorum queue that is either — a health check that could not be 1588 declared would report a healthy broker as down. 1589 """ 1590 if isinstance(self._transport, QueueAdmin): 1591 await self._transport.queue_exists(name) 1592 return 1593 await self._transport.declare_queue( 1594 name, QueueSpec(durable=False, auto_delete=True, exclusive=True) 1595 ) 1596 1597 @property 1598 def publish_interceptors(self) -> tuple[PublishInterceptor, ...]: 1599 """What wraps every publish, outermost first. 1600 1601 A snapshot rather than the list itself, so a publish in progress cannot 1602 be handed a chain that is being edited underneath it. 1603 """ 1604 return tuple(self._publishing) 1605 1606 @property 1607 def consume_interceptors(self) -> tuple[ConsumeInterceptor, ...]: 1608 """What wraps every handler, outermost first.""" 1609 return tuple(self._consuming) 1610 1611 def intercept_publish(self, interceptor: PublishInterceptor) -> Connection: 1612 """Wraps every publish on this connection in ``interceptor``. 1613 1614 Registered ones run in the order they were added, so this one is 1615 innermost until the next is added. Read at publish time rather than 1616 copied into each publisher, so an interceptor registered during start-up 1617 applies to publishers that already exist — which is what makes it usable 1618 from a framework's wiring rather than only from the line that connects. 1619 1620 Adding at start-up is still what to do. A message already in flight will 1621 not see it, and an interceptor whose whole job is to stamp every message 1622 is not doing it if the first few went without. 1623 1624 :param interceptor: what to wrap the publish in. See 1625 :mod:`acemq_amqp.interceptors` 1626 :returns: this connection, so registrations read as one expression 1627 """ 1628 self._publishing.append(interceptor) 1629 return self 1630 1631 def intercept_consume(self, interceptor: ConsumeInterceptor) -> Connection: 1632 """Wraps every handler on this connection in ``interceptor``. 1633 1634 :param interceptor: what to wrap the handler in 1635 :returns: this connection 1636 """ 1637 self._consuming.append(interceptor) 1638 return self 1639 1640 async def declare(self, topology: Topology) -> None: 1641 """Applies a topology to this connection's broker. 1642 1643 :param topology: what the broker should have 1644 """ 1645 await topology.apply(self._transport) 1646 1647 def publisher( 1648 self, 1649 exchange: str = "", 1650 routing_key: str = "", 1651 *, 1652 codec: Codec | None = None, 1653 persistent: bool = True, 1654 mandatory: bool = False, 1655 ) -> Publisher: 1656 """Builds a publisher for an exchange and routing key. 1657 1658 Publish straight to a queue by leaving the exchange empty: the default 1659 exchange routes to the queue whose name matches the routing key. 1660 1661 :param exchange: where to publish, empty for the default exchange 1662 :param routing_key: what to publish under, or a queue name 1663 :param codec: a codec other than this connection's 1664 :param persistent: ask the broker to write these messages to disk 1665 :param mandatory: fail when a message reaches no queue at all, rather 1666 than letting the broker drop it silently 1667 :returns: the publisher 1668 """ 1669 return Publisher( 1670 self, 1671 exchange, 1672 routing_key, 1673 codec=codec, 1674 persistent=persistent, 1675 mandatory=mandatory, 1676 ) 1677 1678 async def consume( 1679 self, 1680 queue: str, 1681 handler: Handler, 1682 *, 1683 codec: Codec | None = None, 1684 retry: RetryPolicy | None = None, 1685 prefetch: int | None = None, 1686 concurrency: int = 1, 1687 tag: str = "", 1688 args: Mapping[str, Any] | None = None, 1689 declare: bool = True, 1690 ) -> Consumer: 1691 """Reads messages from a queue until the returned consumer is closed. 1692 1693 Before it subscribes it declares the queues it will need when a message 1694 fails: ``{queue}.dlq``, ``{queue}.parked``, ``acemq.dlx`` and, when the 1695 retry policy has waits the broker holds, ``acemq.retry`` and the rungs. 1696 See :func:`acemq_amqp.topology.declare_where_failures_go` for exactly 1697 what and why. The source queue is not among them, and neither is 1698 anything a producer needs, so this is not a replacement for declaring a 1699 topology — it is the floor underneath a service that was deployed 1700 without one. 1701 1702 :param queue: what to read 1703 :param handler: what to do with a message; returning an 1704 :class:`~acemq_amqp.Ack` is the decision 1705 :param codec: a codec other than this connection's 1706 :param retry: a policy other than this connection's 1707 :param prefetch: how many unacknowledged messages to hold 1708 :param concurrency: how many messages to work on at once. One by 1709 default, which keeps a queue's messages in order; raising it trades 1710 that order for throughput, which is the right trade for handlers 1711 that spend their time waiting on something else 1712 :param tag: what to call this consumer to the broker 1713 :param args: broker-specific consumer arguments 1714 :param declare: declare the queues above before subscribing. On by 1715 default, because the alternative default loses messages silently. 1716 Turn it off for a login with no ``configure`` permission on the 1717 vhost, which is refused by the broker rather than ignored, and for a 1718 tool draining a queue it does not own — a one-off reader of 1719 ``orders.dlq`` has no business creating ``orders.dlq.dlq``. A 1720 consumer started this way still publishes to those queues, so 1721 somebody else has to have declared them 1722 :returns: the running consumer 1723 """ 1724 if self._closed: 1725 raise AceMQError("acemq: this connection is closed") 1726 1727 consumer = Consumer( 1728 self, 1729 queue, 1730 handler, 1731 codec=codec or self._codec, 1732 retry=retry or self._retry, 1733 concurrency=concurrency, 1734 declare=declare, 1735 ) 1736 self._consumers.append(consumer) 1737 try: 1738 await consumer._start( 1739 self._prefetch if prefetch is None else prefetch, 1740 tag, 1741 args or {}, 1742 ) 1743 except BaseException: 1744 self._untrack(consumer) 1745 raise 1746 return consumer 1747 1748 async def publish_raw( 1749 self, exchange: str, routing_key: str, message: Outbound 1750 ) -> PublishResult: 1751 """Sends bytes that are already encoded, with headers already rendered. 1752 1753 For the retry and dead-letter hops, and for anything else replaying a 1754 message recorded earlier: the payload's class may not exist any more, 1755 and re-encoding it would produce different bytes from the ones that were 1756 committed. Ordinary publishing goes through :class:`Publisher`. 1757 1758 **This is the one place the outstanding-publish bound is taken.** Every 1759 publish in the library comes through here — a :class:`Publisher`, a 1760 retry rung, a dead letter, a replay — so one permit counts them all, and 1761 :meth:`Publisher.send_all` cannot put more messages on the wire at once 1762 than :attr:`max_outstanding_publishes` allows however long the list it 1763 was given is. 1764 1765 :param exchange: where to publish, empty for the default exchange 1766 :param routing_key: what to publish under 1767 :param message: the message, encoded 1768 :returns: what the broker said 1769 :raises PublishingPausedError: when the broker has blocked this 1770 connection. Nothing was sent; retry once it unblocks 1771 :raises PublishError: when every permit is taken and none came free 1772 within :attr:`confirm_timeout`, or when the broker did not confirm 1773 this publish within it. The second may or may not have arrived 1774 """ 1775 # Refused rather than queued, before a permit or a byte is spent. A 1776 # blocked broker stops reading the socket, so a publish sent now sits in 1777 # the client's write buffer with no confirm coming until the alarm 1778 # clears, then times out as a failure that may or may not have arrived. 1779 # Go and .NET refuse at the same point for the same reason. ``None`` 1780 # means the state could not be read and is not treated as blocked. 1781 if self.blocked is True: 1782 reason = self.blocked_reason 1783 raise PublishingPausedError( 1784 message.message_id, 1785 exchange, 1786 routing_key, 1787 ( 1788 f"{self._blocked_detail(reason)}. Nothing was sent; back off and " 1789 "retry once the broker unblocks" 1790 ), 1791 ) 1792 1793 # Taken before the message is written and not after, which is the whole 1794 # point: a bound applied afterwards has already let the message into 1795 # memory. The permit is given back in the ``finally`` below on every 1796 # path there is — the publish returning, the publish raising, and this 1797 # task being cancelled while the broker is still thinking — because a 1798 # permit lost on any one of those is a connection that publishes a 1799 # thousand more messages and then stops for ever. 1800 try: 1801 await asyncio.wait_for( 1802 self._outstanding.acquire(), self._confirm_timeout.total_seconds() 1803 ) 1804 except asyncio.TimeoutError as stalled: 1805 # Said rather than waited out. A publisher parked behind a broker 1806 # that has stopped answering is indistinguishable from a quiet 1807 # service, and this sentence is the one Java raises word for word so 1808 # that an operator reading a log has one thing to recognise. 1809 raise PublishError( 1810 message.message_id, 1811 exchange, 1812 routing_key, 1813 ( 1814 f"{self._max_outstanding} publishes are already waiting for a " 1815 f"confirm and none completed within {self._confirm_timeout}. " 1816 "The broker is not keeping up; publish more slowly rather than " 1817 "buffering more." 1818 ), 1819 ) from stalled 1820 1821 self._in_flight_publishes += 1 1822 try: 1823 # Bounded as well, by the same timeout. RabbitMQ says a connection is 1824 # blocked only once it publishes, so the first publish after an alarm 1825 # is written and then held with no confirm until the alarm clears -- 1826 # minutes, or never. Said here as a failure, not a refusal: it was 1827 # written, so it may have arrived. The publish after it sees the 1828 # block and is refused above. .NET bounds the same wait the same way. 1829 result = await asyncio.wait_for( 1830 self._transport.publish(exchange, routing_key, message), 1831 self._confirm_timeout.total_seconds(), 1832 ) 1833 except asyncio.TimeoutError as unconfirmed: 1834 raise PublishError( 1835 message.message_id, 1836 exchange, 1837 routing_key, 1838 ( 1839 f"the broker did not confirm this publish within " 1840 f"{self._confirm_timeout}. It may or may not have been stored, " 1841 "so treat it as neither sent nor lost" 1842 ), 1843 ) from unconfirmed 1844 finally: 1845 self._in_flight_publishes -= 1 1846 self._outstanding.release() 1847 1848 # A nack is the broker declining to keep a message it was asked to take, and 1849 # it is raised here rather than returned. 1850 # 1851 # It used to be returned, and `confirmed` was read nowhere in this library 1852 # except a telemetry label -- so a caller who did not inspect the result 1853 # carried on believing the message had gone somewhere, and Java, .NET, Go and 1854 # Ruby all raised on the same event. Worse than the wrong belief: the outbox 1855 # relay removes a record because the publish did not raise, so a nack deleted 1856 # a record for a message the broker had refused. At-most-once, in the pattern 1857 # whose whole purpose is not losing anything. 1858 # 1859 # Checked here because this is the one place every publish in the library 1860 # comes through -- a Publisher, a retry rung, a dead letter, a replay, the 1861 # outbox -- so one check covers all of them, and anything that settles or 1862 # deletes on the strength of a publish returning now hears the refusal. 1863 # 1864 # Distinct from unroutable, which the Publisher path reports separately and 1865 # which sets `unroutable`: the broker could route this one and would not keep 1866 # it, and the two want different responses. 1867 # 1868 # So a returned message is passed back rather than raised here. The 1869 # RabbitMQ transport reports a Basic.Return as not confirmed as well as 1870 # not routed, and raising it as a nack hid ``routed`` from every caller 1871 # that reads it: a mandatory Publisher said "refused" rather than 1872 # "reached no queue", and the retry and dead-letter hops never reached 1873 # the fallbacks written for a queue that is not there. 1874 if not result.confirmed and result.routed: 1875 raise PublishError( 1876 message.message_id, 1877 exchange, 1878 routing_key, 1879 ( 1880 "the broker refused to confirm it. The message reached the broker " 1881 "and the broker has not taken responsibility for it, so it must be " 1882 "treated as not sent" 1883 ), 1884 ) 1885 1886 return result 1887 1888 async def pull(self, queue: str) -> Delivery | None: 1889 """Takes one message off a queue, or ``None`` when there is none waiting. 1890 1891 For tools rather than for services: it is a round trip per message where 1892 :meth:`consume` gets a stream. What it can do that a consumer cannot is 1893 say that a queue is empty, which is what anything working through a 1894 backlog and then stopping — a replay, a drain — has to know. 1895 1896 The delivery comes back unsettled, so the caller decides whether it is 1897 gone or goes back. 1898 1899 :param queue: what to read from 1900 :returns: the delivery, or ``None`` 1901 """ 1902 return await self._source().pull(queue) 1903 1904 async def queue_exists(self, queue: str) -> bool: 1905 """Whether a queue is on the broker, creating nothing.""" 1906 return await self._admin().queue_exists(queue) 1907 1908 async def message_count(self, queue: str) -> int: 1909 """How many messages are waiting on a queue.""" 1910 return await self._admin().message_count(queue) 1911 1912 async def delete_queue(self, queue: str) -> None: 1913 """Removes a queue and every message still on it.""" 1914 await self._admin().delete_queue(queue) 1915 1916 def _admin(self) -> QueueAdmin: 1917 if not isinstance(self._transport, QueueAdmin): 1918 raise AceMQError( 1919 f"acemq: the {type(self._transport).__name__} transport cannot manage queues" 1920 ) 1921 return self._transport 1922 1923 def _source(self) -> MessageSource: 1924 if not isinstance(self._transport, MessageSource): 1925 raise AceMQError( 1926 f"acemq: the {type(self._transport).__name__} transport cannot be " 1927 "asked for one message at a time" 1928 ) 1929 return self._transport 1930 1931 def _untrack(self, consumer: Consumer) -> None: 1932 if consumer in self._consumers: 1933 self._consumers.remove(consumer) 1934 1935 async def close(self, *, timeout: float | None = DEFAULT_DRAIN_TIMEOUT) -> bool: 1936 """Stops every consumer on this connection and releases it. 1937 1938 Consumers first, and their handlers given up to ``timeout`` to finish, 1939 so a message being worked on when this is called is settled rather than 1940 returned to the queue for somebody else to redo. The bound is one 1941 deadline for every handler of every consumer, not one each: they are 1942 all drained at once. A handler still running at the deadline is 1943 cancelled and its message left unsettled, for the broker to redeliver — 1944 see :meth:`Consumer.close`. 1945 1946 The connection is released whatever happens, including when the caller 1947 cancels closing; that cancellation is raised again afterwards. 1948 1949 :param timeout: seconds to wait for the running handlers; ``None`` 1950 waits for as long as they take 1951 :returns: ``True`` if every handler finished, ``False`` if the time ran 1952 out first 1953 """ 1954 if self._closed: 1955 return True 1956 self._closed = True 1957 try: 1958 drained = await asyncio.gather( 1959 *(consumer.close(timeout=timeout) for consumer in list(self._consumers)), 1960 return_exceptions=True, 1961 ) 1962 finally: 1963 await self._transport.close() 1964 for outcome in drained: 1965 if isinstance(outcome, BaseException): 1966 raise outcome 1967 return all(drained) 1968 1969 async def __aenter__(self) -> Connection: 1970 return self 1971 1972 async def __aexit__(self, *_: object) -> None: 1973 await self.close() 1974 1975 1976@dataclass(frozen=True, slots=True) 1977class BrokerHealth: 1978 """A connection as a :class:`~acemq_amqp.telemetry.HealthCheck`. 1979 1980 So that a broker check and an application's own — a database, a downstream 1981 service — can be handed to 1982 :func:`~acemq_amqp.telemetry.aggregate_health` together, which is what a 1983 readiness endpoint actually needs. 1984 1985 A blocked broker comes back through here as **up, with the reason**, and 1986 nothing in between folds it into anything else: the aggregate takes the 1987 worst status it is given, so a check that reported back pressure as degraded 1988 would quietly degrade a report that had decided otherwise. The deadline is 1989 the same argument — :attr:`timeout` is shorter than the aggregate's, so that 1990 a broker which has stopped answering is described by this check rather than 1991 by the aggregate giving up on it. 1992 1993 :param connection: what to ask 1994 :param label: what to call it in a combined report 1995 :param timeout: how long to give the broker's answer. See 1996 :meth:`Connection.health` 1997 """ 1998 1999 connection: Connection 2000 label: str = "broker" 2001 timeout: float = DEFAULT_HEALTH_TIMEOUT 2002 2003 @property 2004 def name(self) -> str: 2005 return self.label 2006 2007 async def check(self) -> HealthReport: 2008 return await self.connection.health(self.timeout) 2009 2010 2011async def connect( 2012 url: str, 2013 *, 2014 codec: Codec | None = None, 2015 origin: str | None = None, 2016 retry: RetryPolicy | None = None, 2017 prefetch: int = DEFAULT_PREFETCH, 2018 max_outstanding_publishes: int = DEFAULT_MAX_OUTSTANDING_PUBLISHES, 2019 confirm_timeout: timedelta = DEFAULT_CONFIRM_TIMEOUT, 2020 security: Security | None = None, 2021 on_publish: Sequence[PublishInterceptor] = (), 2022 on_consume: Sequence[ConsumeInterceptor] = (), 2023 observer: Observer | None = None, 2024 **transport_options: Any, 2025) -> Connection: 2026 """Opens a connection to a broker. 2027 2028 The URL's scheme picks the transport, and says whether the connection is 2029 encrypted: ``amqps://`` is, ``amqp://`` is not. Both need aio-pika, which is 2030 why it is imported here rather than by the package: a program that only 2031 reads AceMQ envelopes should not be made to install an AMQP client to import 2032 this library. 2033 2034 An ``amqps://`` URL is verified against the machine's trust store and will 2035 not speak anything older than TLS 1.2, with or without a ``security``. Pass 2036 one to name a different authority, to present a client certificate, or to 2037 supply the login separately from the URL — which is how a password stays out 2038 of a connection string that ends up in a log. 2039 2040 :param url: where the broker is 2041 :param codec: what publishers and consumers use unless they say otherwise 2042 :param origin: what to stamp on published messages 2043 :param retry: what consumers use unless they say otherwise 2044 :param prefetch: how many unacknowledged messages a consumer holds 2045 :param max_outstanding_publishes: how many publishes may be waiting for the 2046 broker at once, a thousand by default. See :class:`Connection` 2047 :param confirm_timeout: how long a publish waits for room before raising 2048 :param security: how to verify the broker and who to log in as. See 2049 :class:`~acemq_amqp.Security` 2050 :param on_publish: what to wrap every publish in, outermost first. See 2051 :mod:`acemq_amqp.interceptors` 2052 :param on_consume: what to wrap every handler in 2053 :param observer: where the numbers go. See :mod:`acemq_amqp.telemetry` 2054 :param transport_options: passed to the transport, which for RabbitMQ is 2055 :func:`aio_pika.connect_robust` 2056 :returns: the connection 2057 :raises SecurityError: when the security settings cannot be honoured, before 2058 anything is dialled 2059 """ 2060 scheme = urlsplit(url).scheme 2061 if scheme not in ("amqp", "amqps"): 2062 raise AceMQError( 2063 f"acemq: no transport knows how to reach {scheme or url!r}; " 2064 "this library speaks amqp:// and amqps://" 2065 ) 2066 2067 # Resolved here rather than in the transport, so that a missing certificate 2068 # or an unset password variable is an exception from connect() with the 2069 # setting named in it, rather than a handshake failure from somewhere in 2070 # aio-pika that says only that the socket closed. 2071 settings = security or Security() 2072 url = settings.applied_to(url) 2073 # The caller's own options win: transport_options is the escape hatch for 2074 # everything this library does not model, and a Security built from defaults 2075 # should not quietly overrule an ssl_context somebody passed by hand. 2076 transport_options = {**settings.transport_options(url), **transport_options} 2077 2078 try: 2079 from .rabbitmq import RabbitMQTransport 2080 except ImportError as missing: # pragma: no cover - depends on how it was installed 2081 raise AceMQError( 2082 "acemq: reaching a broker needs aio-pika, which is an optional extra. " 2083 'Install it with pip install "acemq-amqp[rabbitmq]"' 2084 ) from missing 2085 2086 transport = await RabbitMQTransport.connect(url, **transport_options) 2087 return Connection( 2088 transport, 2089 codec=codec, 2090 origin=origin, 2091 retry=retry, 2092 prefetch=prefetch, 2093 max_outstanding_publishes=max_outstanding_publishes, 2094 confirm_timeout=confirm_timeout, 2095 on_publish=on_publish, 2096 on_consume=on_consume, 2097 observer=observer, 2098 ) 2099 2100 2101def default_origin() -> str: 2102 """``acemq@{hostname}``, which names the machine but not the service. 2103 2104 Enough to tell two pods apart in a log, which is what an origin is for, and 2105 deliberately not a guess at the service name: a wrong service name in a 2106 message that has already been published is worse than an honest hostname. 2107 """ 2108 return f"acemq@{socket.gethostname()}" 2109 2110 2111def _stalled_detail(stalled: list[str]) -> str: 2112 """The sentence naming the consumers that have stopped reading.""" 2113 return "these consumers have stopped reading: " + ", ".join(sorted(stalled)) 2114 2115 2116def _retrieved(task: asyncio.Task[None]) -> None: 2117 """Reads an abandoned task's outcome so asyncio does not complain about it.""" 2118 if not task.cancelled(): 2119 task.exception() 2120 2121 2122def _describe(failure: BaseException | None) -> str: 2123 """A failure as the sentence that goes in ``x-acemq-error``.""" 2124 if failure is None: 2125 return "no reason given" 2126 text = str(failure) 2127 return type(failure).__name__ + (f": {text}" if text else "")
log =
<Logger acemq (WARNING)>
DEFAULT_PREFETCH =
20
DEFAULT_MAX_OUTSTANDING_PUBLISHES =
1000
DEFAULT_CONFIRM_TIMEOUT =
datetime.timedelta(seconds=10)
DEFAULT_HEALTH_TIMEOUT =
3.0
DEFAULT_DRAIN_TIMEOUT =
20.0
BLOCKED_DETAIL =
'the broker has blocked this connection; publishing is paused'
@dataclass(frozen=True, slots=True)
class
Message:
166@dataclass(frozen=True, slots=True) 167class Message: 168 """A delivery that has been decoded. 169 170 :param payload: the body, read through the codec 171 :param envelope: the metadata that travelled with it 172 :param routing_key: the key it arrived under 173 :param content_type: what the sender said the body was, or ``None`` 174 :param redelivered: the broker saying it has handed this one over before 175 :param body: the undecoded bytes, for a handler that wants to see them 176 :param reply_to: AMQP's own ``reply-to`` property, empty when the sender set 177 none. A responder reads it as the fallback behind the 178 ``acemq-reply-to`` header, which is what lets a Java or .NET caller — 179 neither of which writes the header — be answered by a Python responder 180 """ 181 182 payload: Any 183 envelope: Envelope 184 routing_key: str 185 content_type: str | None 186 redelivered: bool 187 body: bytes 188 reply_to: str = ""
A delivery that has been decoded.
Parameters
- payload: the body, read through the codec
- envelope: the metadata that travelled with it
- routing_key: the key it arrived under
- content_type: what the sender said the body was, or
None - redelivered: the broker saying it has handed this one over before
- body: the undecoded bytes, for a handler that wants to see them
- reply_to: AMQP's own
reply-toproperty, empty when the sender set none. A responder reads it as the fallback behind theacemq-reply-toheader, which is what lets a Java or .NET caller — neither of which writes the header — be answered by a Python responder
Message( payload: Any, envelope: acemq_amqp.Envelope, routing_key: str, content_type: str | None, redelivered: bool, body: bytes, reply_to: str = '')
envelope: acemq_amqp.Envelope
class
Publisher:
208class Publisher: 209 """Sends messages to one destination. 210 211 Cheap to keep and safe to share: build one per message type at start-up 212 rather than one per message. 213 """ 214 215 def __init__( 216 self, 217 connection: Connection, 218 exchange: str, 219 routing_key: str, 220 *, 221 codec: Codec | None = None, 222 persistent: bool = True, 223 mandatory: bool = False, 224 ) -> None: 225 self._connection = connection 226 self._exchange = exchange 227 self._routing_key = routing_key 228 self._codec = codec or connection.codec 229 self._persistent = persistent 230 self._mandatory = mandatory 231 232 async def send( 233 self, 234 payload: Any, 235 *, 236 envelope: Envelope | None = None, 237 routing_key: str | None = None, 238 reply_to: str = "", 239 ) -> PublishResult: 240 """Publishes one message. 241 242 The envelope is built here when none is given: an identifier, a 243 correlation defaulting to it, this connection's origin, and now. The 244 type falls back to the routing key, which is why a publisher bound to 245 ``order.placed`` needs no further configuration to say what it sends. 246 247 :param payload: what to send, encoded by this publisher's codec 248 :param envelope: metadata to send instead of a fresh one, for a message 249 derived from another 250 :param routing_key: what to publish under, for a publisher bound to an 251 exchange rather than to one key 252 :param reply_to: where an answer should come back to, written to AMQP's 253 own ``reply-to`` property. Only :mod:`~acemq_amqp.patterns` request 254 and reply sets it, and it sets the ``acemq-reply-to`` header to the 255 same value: the header is what survives a service that rebuilds the 256 message, the property is what the other four libraries read 257 :returns: what the broker said 258 :raises PublishError: when the publisher is mandatory and the message 259 reached no queue 260 """ 261 key = self._routing_key if routing_key is None else routing_key 262 outgoing = envelope or Envelope(origin=self._connection.origin) 263 if not outgoing.origin: 264 # A message derived from another — a reply, a pipeline hop, the next 265 # stop on a routing slip — is still published by this process, and an 266 # origin naming the service before it would be a lie in a log line 267 # somebody is going to trust. 268 outgoing = outgoing.with_(origin=self._connection.origin) 269 270 context = PublishContext( 271 exchange=self._exchange, 272 routing_key=key, 273 envelope=outgoing, 274 payload=payload, 275 persistent=self._persistent, 276 mandatory=self._mandatory, 277 reply_to=reply_to, 278 ) 279 # Read here rather than at construction, so an interceptor registered 280 # during start-up applies to the publishers that already exist. Building 281 # the chain per message costs a few closures and buys that. 282 return await publish_chain(self._connection.publish_interceptors, self._send)(context) 283 284 async def send_all(self, payloads: Iterable[Any]) -> list[PublishResult]: 285 """Publishes a batch and waits for every confirm. 286 287 What most bulk publishing actually wants: the throughput of pipelining 288 with the safety of having waited. Every message goes out before any 289 confirm is awaited, and only then are all of them checked together. 290 Awaiting each send in turn is a broker round trip per message — the 291 loop a caller could have written for themselves, and none of the 292 throughput a batch exists for. 293 294 The results come back in the order the payloads were given, whatever 295 order the broker answers in. 296 297 **This is not atomic.** AMQP has no such thing: there is no way to 298 publish a hundred messages such that all or none arrive, and a library 299 that offered one would be lying. A failure does not abandon the rest 300 either — every send is awaited whatever an earlier one did, so a batch 301 that half succeeded can say how many arrived. That count is the point: 302 a caller told only "it failed" resends the messages the broker already 303 has. 304 305 How wide the batch gets on the wire is capped, and by the connection 306 rather than by this method: 307 :attr:`Connection.max_outstanding_publishes` is a thousand by default, 308 and the thousand-and-first task waits for one of the first thousand to 309 be answered before its message is written. A list of a million is 310 therefore a million tasks and a thousand messages in flight rather than 311 a million, which is backpressure instead of a memory leak — but a 312 million tasks is still a million tasks, so chunk what comes out of a 313 database rather than handing a cursor's worth of rows to one call. 314 315 :param payloads: what to send, in order, each encoded by this 316 publisher's codec 317 :returns: what the broker said about each, in the order the payloads 318 were given 319 :raises PublishError: when any message was not confirmed. Its message 320 names how many failed and how many did not, and ``__cause__`` 321 carries the first failure in payload order — which is not 322 necessarily the first one the broker answered. A batch wide enough 323 that the broker stops keeping up fails this way too, with the first 324 failure saying so 325 """ 326 batch = list(payloads) 327 # Everything is started before anything is awaited. A task per payload, 328 # all of them created before the gather below, is what puts the whole 329 # batch on the wire while the broker is still working through the 330 # first confirm. 331 sends = [asyncio.create_task(self.send(payload)) for payload in batch] 332 # return_exceptions, and not for tidiness: without it the first failure 333 # comes out of the gather while the rest of the batch is still in 334 # flight, which loses the count this method exists to report and leaves 335 # the others' exceptions unretrieved for asyncio to complain about 336 # later. 337 settled = await asyncio.gather(*sends, return_exceptions=True) 338 339 results: list[PublishResult] = [] 340 first_failure: BaseException | None = None 341 failed = 0 342 for outcome in settled: 343 if isinstance(outcome, BaseException): 344 failed += 1 345 if first_failure is None: 346 first_failure = outcome 347 else: 348 results.append(outcome) 349 350 if first_failure is not None: 351 # Named rather than printed. A failure with nothing in its message — 352 # the CancelledError every task gets when the connection closes 353 # underneath the batch is the common one — renders as an empty 354 # string, and the sentence then ends "The first failure was: " and 355 # stops, promising an explanation it does not give. The class name 356 # is not much, but it is the difference between "the sends were 357 # cancelled" and nothing at all. 358 described = str(first_failure) or type(first_failure).__name__ 359 # The counts matter. A batch that half succeeded is the ordinary 360 # outcome of a broker problem partway through, and this sentence is 361 # word for word the one Java and .NET raise, so that an operator 362 # reading a log has one thing to recognise rather than three. 363 raise PublishError( 364 "", 365 self._exchange, 366 self._routing_key, 367 described, 368 summary=( 369 f"{failed} of {len(batch)} messages were not confirmed;" 370 f" {len(results)} were. The first failure was: {described}" 371 ), 372 ) from first_failure 373 return results 374 375 async def _send(self, context: PublishContext) -> PublishResult: 376 """The innermost work: encode what the interceptors left, and send it.""" 377 observer = self._connection.observer 378 # ``routing.key`` and not ``key``: Java and .NET both tag a publish with 379 # the fully-qualified name, and a dashboard that groups by it should 380 # read the same in all five languages. 381 labels = {TAG_EXCHANGE: context.exchange, TAG_ROUTING_KEY: context.routing_key} 382 # Started before the encode, because everything from here on is time the 383 # caller spends inside send(): the codec, the wait for a permit and the 384 # broker's confirm are all things that make a publish slow, and a timer 385 # that began after the encode would hide a slow codec completely. 386 started = time.monotonic() 387 388 def record(outcome: str) -> None: 389 """The pair, always together and always with the same labels. 390 391 Two calls rather than one because the interface has two methods, and 392 in this order because a dashboard dividing the duration's count into 393 the total should never see the total lead it. 394 """ 395 observer.observe( 396 METRIC_PUBLISH_DURATION, 397 time.monotonic() - started, 398 {**labels, TAG_OUTCOME: outcome}, 399 ) 400 observer.count(METRIC_PUBLISH_TOTAL, 1, {**labels, TAG_OUTCOME: outcome}) 401 402 # Encoded here rather than before the chain ran, so an interceptor can 403 # change the payload as well as its metadata. A codec that has already 404 # run leaves an interceptor with bytes and nothing it can do to them. 405 try: 406 body = self._codec.encode(context.payload) 407 result = await self._connection.publish_raw( 408 context.exchange, 409 context.routing_key, 410 Outbound( 411 body=body, 412 content_type=self._codec.content_type, 413 message_id=context.envelope.id, 414 reply_to=context.reply_to, 415 headers=context.envelope.to_headers(routing_key=context.routing_key), 416 persistent=context.persistent, 417 mandatory=context.mandatory, 418 ), 419 ) 420 except PublishingPausedError: 421 # Declined before anything was written: nothing lost, so not 422 # ``failed``, which is the outcome an alert on loss watches. 423 record(OUTCOME_REFUSED) 424 raise 425 except Exception: 426 record(OUTCOME_FAILED) 427 raise 428 429 # Raised rather than left in the result, because a caller who does not 430 # read the result would otherwise carry on believing the message went 431 # somewhere. Unroutable is the quietest failure AMQP has: the publish 432 # succeeds, the consumer waits, and nothing anywhere says why. 433 if context.mandatory and not result.routed: 434 record(OUTCOME_UNROUTABLE) 435 raise PublishError( 436 context.envelope.id, 437 context.exchange, 438 context.routing_key, 439 result.return_reason or "no queue is bound to receive it", 440 unroutable=True, 441 ) 442 443 record(OUTCOME_CONFIRMED) 444 return result
Sends messages to one destination.
Cheap to keep and safe to share: build one per message type at start-up rather than one per message.
Publisher( connection: Connection, exchange: str, routing_key: str, *, codec: acemq_amqp.Codec | None = None, persistent: bool = True, mandatory: bool = False)
215 def __init__( 216 self, 217 connection: Connection, 218 exchange: str, 219 routing_key: str, 220 *, 221 codec: Codec | None = None, 222 persistent: bool = True, 223 mandatory: bool = False, 224 ) -> None: 225 self._connection = connection 226 self._exchange = exchange 227 self._routing_key = routing_key 228 self._codec = codec or connection.codec 229 self._persistent = persistent 230 self._mandatory = mandatory
async def
send( self, payload: Any, *, envelope: acemq_amqp.Envelope | None = None, routing_key: str | None = None, reply_to: str = '') -> acemq_amqp.PublishResult:
232 async def send( 233 self, 234 payload: Any, 235 *, 236 envelope: Envelope | None = None, 237 routing_key: str | None = None, 238 reply_to: str = "", 239 ) -> PublishResult: 240 """Publishes one message. 241 242 The envelope is built here when none is given: an identifier, a 243 correlation defaulting to it, this connection's origin, and now. The 244 type falls back to the routing key, which is why a publisher bound to 245 ``order.placed`` needs no further configuration to say what it sends. 246 247 :param payload: what to send, encoded by this publisher's codec 248 :param envelope: metadata to send instead of a fresh one, for a message 249 derived from another 250 :param routing_key: what to publish under, for a publisher bound to an 251 exchange rather than to one key 252 :param reply_to: where an answer should come back to, written to AMQP's 253 own ``reply-to`` property. Only :mod:`~acemq_amqp.patterns` request 254 and reply sets it, and it sets the ``acemq-reply-to`` header to the 255 same value: the header is what survives a service that rebuilds the 256 message, the property is what the other four libraries read 257 :returns: what the broker said 258 :raises PublishError: when the publisher is mandatory and the message 259 reached no queue 260 """ 261 key = self._routing_key if routing_key is None else routing_key 262 outgoing = envelope or Envelope(origin=self._connection.origin) 263 if not outgoing.origin: 264 # A message derived from another — a reply, a pipeline hop, the next 265 # stop on a routing slip — is still published by this process, and an 266 # origin naming the service before it would be a lie in a log line 267 # somebody is going to trust. 268 outgoing = outgoing.with_(origin=self._connection.origin) 269 270 context = PublishContext( 271 exchange=self._exchange, 272 routing_key=key, 273 envelope=outgoing, 274 payload=payload, 275 persistent=self._persistent, 276 mandatory=self._mandatory, 277 reply_to=reply_to, 278 ) 279 # Read here rather than at construction, so an interceptor registered 280 # during start-up applies to the publishers that already exist. Building 281 # the chain per message costs a few closures and buys that. 282 return await publish_chain(self._connection.publish_interceptors, self._send)(context)
Publishes one message.
The envelope is built here when none is given: an identifier, a
correlation defaulting to it, this connection's origin, and now. The
type falls back to the routing key, which is why a publisher bound to
order.placed needs no further configuration to say what it sends.
Parameters
- payload: what to send, encoded by this publisher's codec
- envelope: metadata to send instead of a fresh one, for a message derived from another
- routing_key: what to publish under, for a publisher bound to an exchange rather than to one key
- reply_to: where an answer should come back to, written to AMQP's
own
reply-toproperty. Only~acemq_amqp.patternsrequest and reply sets it, and it sets theacemq-reply-toheader to the same value: the header is what survives a service that rebuilds the message, the property is what the other four libraries read :returns: what the broker said
Raises
- PublishError: when the publisher is mandatory and the message reached no queue
284 async def send_all(self, payloads: Iterable[Any]) -> list[PublishResult]: 285 """Publishes a batch and waits for every confirm. 286 287 What most bulk publishing actually wants: the throughput of pipelining 288 with the safety of having waited. Every message goes out before any 289 confirm is awaited, and only then are all of them checked together. 290 Awaiting each send in turn is a broker round trip per message — the 291 loop a caller could have written for themselves, and none of the 292 throughput a batch exists for. 293 294 The results come back in the order the payloads were given, whatever 295 order the broker answers in. 296 297 **This is not atomic.** AMQP has no such thing: there is no way to 298 publish a hundred messages such that all or none arrive, and a library 299 that offered one would be lying. A failure does not abandon the rest 300 either — every send is awaited whatever an earlier one did, so a batch 301 that half succeeded can say how many arrived. That count is the point: 302 a caller told only "it failed" resends the messages the broker already 303 has. 304 305 How wide the batch gets on the wire is capped, and by the connection 306 rather than by this method: 307 :attr:`Connection.max_outstanding_publishes` is a thousand by default, 308 and the thousand-and-first task waits for one of the first thousand to 309 be answered before its message is written. A list of a million is 310 therefore a million tasks and a thousand messages in flight rather than 311 a million, which is backpressure instead of a memory leak — but a 312 million tasks is still a million tasks, so chunk what comes out of a 313 database rather than handing a cursor's worth of rows to one call. 314 315 :param payloads: what to send, in order, each encoded by this 316 publisher's codec 317 :returns: what the broker said about each, in the order the payloads 318 were given 319 :raises PublishError: when any message was not confirmed. Its message 320 names how many failed and how many did not, and ``__cause__`` 321 carries the first failure in payload order — which is not 322 necessarily the first one the broker answered. A batch wide enough 323 that the broker stops keeping up fails this way too, with the first 324 failure saying so 325 """ 326 batch = list(payloads) 327 # Everything is started before anything is awaited. A task per payload, 328 # all of them created before the gather below, is what puts the whole 329 # batch on the wire while the broker is still working through the 330 # first confirm. 331 sends = [asyncio.create_task(self.send(payload)) for payload in batch] 332 # return_exceptions, and not for tidiness: without it the first failure 333 # comes out of the gather while the rest of the batch is still in 334 # flight, which loses the count this method exists to report and leaves 335 # the others' exceptions unretrieved for asyncio to complain about 336 # later. 337 settled = await asyncio.gather(*sends, return_exceptions=True) 338 339 results: list[PublishResult] = [] 340 first_failure: BaseException | None = None 341 failed = 0 342 for outcome in settled: 343 if isinstance(outcome, BaseException): 344 failed += 1 345 if first_failure is None: 346 first_failure = outcome 347 else: 348 results.append(outcome) 349 350 if first_failure is not None: 351 # Named rather than printed. A failure with nothing in its message — 352 # the CancelledError every task gets when the connection closes 353 # underneath the batch is the common one — renders as an empty 354 # string, and the sentence then ends "The first failure was: " and 355 # stops, promising an explanation it does not give. The class name 356 # is not much, but it is the difference between "the sends were 357 # cancelled" and nothing at all. 358 described = str(first_failure) or type(first_failure).__name__ 359 # The counts matter. A batch that half succeeded is the ordinary 360 # outcome of a broker problem partway through, and this sentence is 361 # word for word the one Java and .NET raise, so that an operator 362 # reading a log has one thing to recognise rather than three. 363 raise PublishError( 364 "", 365 self._exchange, 366 self._routing_key, 367 described, 368 summary=( 369 f"{failed} of {len(batch)} messages were not confirmed;" 370 f" {len(results)} were. The first failure was: {described}" 371 ), 372 ) from first_failure 373 return results
Publishes a batch and waits for every confirm.
What most bulk publishing actually wants: the throughput of pipelining with the safety of having waited. Every message goes out before any confirm is awaited, and only then are all of them checked together. Awaiting each send in turn is a broker round trip per message — the loop a caller could have written for themselves, and none of the throughput a batch exists for.
The results come back in the order the payloads were given, whatever order the broker answers in.
This is not atomic. AMQP has no such thing: there is no way to publish a hundred messages such that all or none arrive, and a library that offered one would be lying. A failure does not abandon the rest either — every send is awaited whatever an earlier one did, so a batch that half succeeded can say how many arrived. That count is the point: a caller told only "it failed" resends the messages the broker already has.
How wide the batch gets on the wire is capped, and by the connection
rather than by this method:
Connection.max_outstanding_publishes is a thousand by default,
and the thousand-and-first task waits for one of the first thousand to
be answered before its message is written. A list of a million is
therefore a million tasks and a thousand messages in flight rather than
a million, which is backpressure instead of a memory leak — but a
million tasks is still a million tasks, so chunk what comes out of a
database rather than handing a cursor's worth of rows to one call.
Parameters
- payloads: what to send, in order, each encoded by this publisher's codec :returns: what the broker said about each, in the order the payloads were given
Raises
- PublishError: when any message was not confirmed. Its message
names how many failed and how many did not, and
__cause__carries the first failure in payload order — which is not necessarily the first one the broker answered. A batch wide enough that the broker stops keeping up fails this way too, with the first failure saying so
class
Consumer:
451class Consumer: 452 """A running subscription. Close it to stop. 453 454 One is built by :meth:`Connection.consume` rather than directly, because it 455 has to be started before it is any use and a half-built consumer that looks 456 finished is a thing somebody will hold on to. 457 458 Starting one declares the queues it will need when a message fails — its 459 dead-letter queue, its parking lot and its rungs — unless it was told not 460 to. See :func:`acemq_amqp.topology.declare_where_failures_go`. 461 """ 462 463 def __init__( 464 self, 465 connection: Connection, 466 queue: str, 467 handler: Handler, 468 *, 469 codec: Codec, 470 retry: RetryPolicy, 471 concurrency: int, 472 declare: bool = True, 473 ) -> None: 474 if concurrency < 1: 475 raise ValueError(f"acemq: concurrency must be at least 1, got {concurrency}") 476 self._connection = connection 477 self._queue = queue 478 self._handler = handler 479 self._codec = codec 480 self._retry = retry 481 self._concurrency = concurrency 482 self._declare = declare 483 self._work: asyncio.Queue[Delivery | None] = asyncio.Queue() 484 self._workers: list[asyncio.Task[None]] = [] 485 self._subscription: Subscription | None = None 486 self._in_flight = 0 487 self._closed = False 488 489 @property 490 def queue(self) -> str: 491 """The queue this consumer reads.""" 492 return self._queue 493 494 @property 495 def closed(self) -> bool: 496 """Whether this consumer has been stopped.""" 497 return self._closed 498 499 @property 500 def running(self) -> bool: 501 """Whether this consumer is subscribed and still has workers to hand a 502 delivery to. 503 504 What a health check reads. A consumer whose workers have all finished 505 without it being closed is a consumer the broker is still sending 506 messages to and nothing is reading — which looks identical to a quiet 507 queue from outside, and is the failure worth telling a probe about. 508 """ 509 return ( 510 not self._closed 511 and self._subscription is not None 512 and any(not worker.done() for worker in self._workers) 513 ) 514 515 @property 516 def in_flight(self) -> int: 517 """How many messages this consumer is working on right now.""" 518 return self._in_flight 519 520 @property 521 def _labels(self) -> dict[str, str]: 522 return {TAG_QUEUE: self._queue} 523 524 @property 525 def _observer(self) -> Observer: 526 return self._connection.observer 527 528 async def _start(self, prefetch: int, tag: str, args: Mapping[str, Any]) -> None: 529 if self._declare: 530 # Before the subscription rather than after it, so that the first 531 # message this consumer gives up on already has somewhere to go. A 532 # broker that refuses one of these declarations is a consumer that 533 # does not start, which is the honest failure: it would otherwise 534 # start and lose its first dead letter. 535 await declare_where_failures_go( 536 self._connection.transport, self._queue, self._retry 537 ) 538 539 self._workers = [ 540 asyncio.create_task(self._work_loop(), name=f"acemq-consumer-{self._queue}-{n}") 541 for n in range(self._concurrency) 542 ] 543 try: 544 self._subscription = await self._connection.transport.consume( 545 self._queue, 546 ConsumeSpec(prefetch=prefetch, tag=tag, args=dict(args)), 547 self._deliver, 548 ) 549 except BaseException: 550 # Nothing was subscribed, so nothing will arrive; the workers would 551 # otherwise sit on an empty queue for the life of the process. 552 for _ in self._workers: 553 self._work.put_nowait(None) 554 await asyncio.gather(*self._workers) 555 raise 556 557 async def _deliver(self, delivery: Delivery) -> None: 558 """Hands a delivery to the workers. 559 560 Queued rather than handled here, so that ``concurrency`` means what it 561 says: aio-pika starts a task per delivery and does not wait for the last 562 one, so without a queue of our own a prefetch of twenty would run twenty 563 handlers at once whatever the caller asked for. The queue needs no bound 564 because prefetch is one: the broker will not send a twenty-first message 565 until one of the twenty has been settled. 566 """ 567 self._work.put_nowait(delivery) 568 569 async def _work_loop(self) -> None: 570 while True: 571 delivery = await self._work.get() 572 if delivery is None: 573 return 574 try: 575 await self._handle(delivery) 576 except Exception: 577 # A failure here is this library's, not the handler's, and the 578 # message is left unsettled on purpose: the broker will hand it 579 # back rather than it disappearing because our own code broke. 580 log.exception("acemq: could not settle a delivery from %s", self._queue) 581 582 async def _handle(self, delivery: Delivery) -> None: 583 envelope = Envelope.from_headers(delivery.headers, delivery.routing_key) 584 # No counter for "a message arrived": every delivery is counted once 585 # when it is settled, on ``acemq.consume.total`` with the outcome, and 586 # the sum across the outcomes is how many arrived. One counter that 587 # leads the others by however many messages are in flight is a counter 588 # that makes an operator wonder which one is lying. 589 # 590 # The attempt is a distribution rather than a counter, and it is recorded 591 # here on arrival rather than at the settlement: it is a fact about the 592 # delivery that is true before the handler runs, and recording it after 593 # would drop every message this consumer is still working on when the 594 # process stops. A rising distribution is a dependency in trouble, and it 595 # says so before anything else does — the dead letters only move once the 596 # attempts run out. 597 self._observer.observe( 598 METRIC_CONSUME_ATTEMPTS, float(envelope.attempt), self._labels 599 ) 600 self._in_flight += 1 601 self._observer.gauge(METRIC_CONSUME_IN_FLIGHT, self._in_flight, self._labels) 602 try: 603 await self._handle_one(delivery, envelope) 604 finally: 605 self._in_flight -= 1 606 self._observer.gauge(METRIC_CONSUME_IN_FLIGHT, self._in_flight, self._labels) 607 608 async def _handle_one(self, delivery: Delivery, envelope: Envelope) -> None: 609 try: 610 payload = self._codec.decode(delivery.body, delivery.content_type) 611 except Exception as failure: 612 # A body that will not decode decodes no better next time, so it 613 # does not go round the retry schedule until it ages out. 614 # 615 # Parked rather than dead-lettered: a message that failed five times 616 # and a message nothing could read are different problems with 617 # different answers — one is usually the world, the other is usually 618 # a producer — and whoever drains the dead letters should not have 619 # to sort them by hand. 620 # 621 # Counted as a settled delivery here rather than in ``_carry_out``, 622 # which this path never reaches: no handler ran, so there is no 623 # duration to record, but the message did arrive and was dealt with 624 # and a total that missed it would not add up. 625 self._observer.count( 626 METRIC_CONSUME_TOTAL, 1, {**self._labels, TAG_OUTCOME: OUTCOME_PARKED} 627 ) 628 await self._park( 629 delivery, envelope, f"could not be decoded: {_describe(failure)}" 630 ) 631 return 632 633 context = ConsumeContext( 634 queue=self._queue, 635 envelope=envelope, 636 payload=payload, 637 body=delivery.body, 638 content_type=delivery.content_type, 639 routing_key=delivery.routing_key, 640 redelivered=delivery.redelivered, 641 # This delivery is being driven by a consumer, so how it ends will 642 # be reported. An interceptor that wants to know can wait for it. 643 reports_settlement=True, 644 reply_to=delivery.reply_to, 645 ) 646 647 started = time.monotonic() 648 try: 649 decision = await consume_chain( 650 self._connection.consume_interceptors, self._run_handler 651 )(context) 652 except Exception as failure: 653 # An exception is how Python says a thing failed, so a handler that 654 # raises is asking for the retry policy rather than confessing a 655 # bug — which is what Java does with the same situation, and what a 656 # handler written without reading the documentation will expect. 657 # Raising FatalError is how it says the opposite, and the retry path 658 # below reads that mark before it reads the request. 659 # 660 # An interceptor that raises lands here too, and deliberately: a 661 # refused message is retried and then dead-lettered rather than 662 # acknowledged as though something had processed it. 663 decision = Ack(Action.RETRY, failure) 664 665 # Timed around the interceptors as well as the handler, because what an 666 # operator wants to know is how long a message takes to deal with, and 667 # an interceptor that opens a transaction is part of dealing with it. 668 # Stopped here and recorded a few lines below rather than in a 669 # ``finally``, because the duration is tagged with the outcome and the 670 # outcome is not known until ``_decide`` has run: how long a message 671 # took and what happened to it are one question, and a p99 that mixes 672 # the messages that worked with the ones that timed out answers neither 673 # half of it. 674 elapsed = time.monotonic() - started 675 676 # Whatever the interceptors left, rather than what arrived: one that 677 # rewrote the envelope on the way in meant that rewrite for the retry 678 # and the dead letter as much as for the handler. 679 envelope = context.envelope 680 681 if isinstance(decision, Ack): 682 settlement, wait = self._decide(envelope, decision) 683 else: 684 settlement, wait = ( 685 Settlement( 686 OUTCOME_DEAD_LETTERED, 687 reason=( 688 f"the handler returned {type(decision).__name__} instead of an Ack" 689 ), 690 ), 691 None, 692 ) 693 694 self._observer.observe( 695 METRIC_CONSUME_DURATION, 696 elapsed, 697 {**self._labels, TAG_OUTCOME: settlement.outcome}, 698 ) 699 700 # Announced before it is carried out, and so before a consumer-side 701 # retry sleeps out its backoff. A listener told afterwards would be 702 # describing the delivery minutes after the decision, and the span it is 703 # writing on would have stayed open for a wait that is not work. 704 context.settled(settlement) 705 await self._carry_out(delivery, envelope, settlement, wait) 706 707 async def _run_handler(self, context: ConsumeContext) -> Ack: 708 """The innermost work: build the message the interceptors left, and run 709 the handler on it. 710 711 The message is built here rather than before the chain, so that an 712 interceptor which changed the envelope or the payload changed what the 713 handler is given rather than a copy nobody reads. 714 """ 715 message = Message( 716 payload=context.payload, 717 envelope=context.envelope, 718 routing_key=context.routing_key, 719 content_type=context.content_type, 720 redelivered=context.redelivered, 721 body=context.body, 722 reply_to=context.reply_to, 723 ) 724 returned = self._handler(message) 725 # Not an Ack is a mistake worth reporting, but it is not this method's 726 # to report: it is returned as it stands and _handle dead-letters it 727 # with a reason, which is where every other unhandleable delivery goes. 728 # An interceptor between here and there sees the same thing the handler 729 # returned, which is the honest thing to show it. 730 decision: Ack = await returned if inspect.isawaitable(returned) else returned 731 return decision 732 733 def _decide(self, envelope: Envelope, decision: Ack) -> tuple[Settlement, Wait | None]: 734 """What is going to happen to this delivery, before anything happens to 735 it. 736 737 Worked out on its own, and without touching the broker, so that it can 738 be announced first. The gap this closes is a small one to describe and a 739 wide one to fall into: a handler asking for a retry it has no attempts 740 left for is dead-lettered, and anything reading the handler's answer 741 instead of this one records a retry that never happened. 742 743 :param envelope: the message's metadata, for the attempt and the age 744 :param decision: what the handler answered 745 :returns: the settlement, and the wait when there is another attempt — 746 which the settlement does not carry, because whether a wait is spent 747 here or on a rung queue is this consumer's business and not a 748 listener's 749 """ 750 if decision.action is Action.ACCEPT: 751 return Settlement(OUTCOME_ACKED), None 752 753 if decision.action is Action.REJECT: 754 return ( 755 Settlement( 756 OUTCOME_REJECTED, 757 reason=f"the handler rejected it: {_describe(decision.error)}", 758 ), 759 None, 760 ) 761 762 if decision.action is Action.PARK: 763 # The handler read the message and found nothing it could read. The 764 # engine reaches the same conclusion for a body its codec refuses, 765 # and both end up in the same queue for the same reason: a message 766 # nobody can read is a producer problem, and it is only findable if 767 # it is not filed with the messages that merely ran out of luck. 768 return ( 769 Settlement( 770 OUTCOME_PARKED, 771 reason=f"the handler parked it: {_describe(decision.error)}", 772 ), 773 None, 774 ) 775 776 if decision.action is Action.IN_PROGRESS: 777 # Somebody else holds the message. Not a failure of this one, so the 778 # retry policy is not asked: it can neither spend an attempt nor run 779 # out and dead-letter a message nobody has failed to handle. 780 delay = decision.delay if decision.delay is not None else ZERO 781 return Settlement(OUTCOME_IN_PROGRESS, delay=delay), Wait(delay, in_broker=False) 782 783 if isinstance(decision.error, FatalError): 784 # The handler asked for a retry but marked the reason as one that 785 # will not change. Honouring the mark rather than the request is the 786 # point of having it. 787 return ( 788 Settlement( 789 OUTCOME_DEAD_LETTERED, 790 reason=( 791 "the handler reported an unprocessable message: " 792 f"{_describe(decision.error)}" 793 ), 794 ), 795 None, 796 ) 797 798 wait = self._retry.next_wait(envelope.attempt, envelope.age) 799 if wait is None: 800 return ( 801 Settlement( 802 OUTCOME_DEAD_LETTERED, 803 reason=f"{self._exhausted(envelope)}: {_describe(decision.error)}", 804 ), 805 None, 806 ) 807 808 return Settlement(OUTCOME_RETRIED, delay=wait.delay), wait 809 810 async def _carry_out( 811 self, 812 delivery: Delivery, 813 envelope: Envelope, 814 settlement: Settlement, 815 wait: Wait | None, 816 ) -> None: 817 """Does what :meth:`_decide` decided, and counts it as decided. 818 819 Every counter here is chosen from the :class:`~acemq_amqp.ack.Settlement` 820 and never from the handler's :class:`~acemq_amqp.ack.Ack`, which is the 821 same rule the delivery's span follows. A handler asking for a retry it 822 has no attempts left for is dead-lettered, and a counter that read the 823 request rather than the answer would report a retry for a message 824 nothing will ever try again — so the dead letters would be undercounted 825 by exactly the messages an operator most wants to find. 826 827 One ``acemq.consume.total`` per delivery, carrying the outcome, and then 828 whichever of the standalone counters applies. ``retried`` and 829 ``dead_lettered`` are both an outcome here and a counter of their own, 830 which is what Java does and is not a redundancy: the outcome says what 831 was decided about a delivery, and the counter says how many messages are 832 going round again or have been set aside, which is the number an alert 833 is written against. 834 """ 835 self._observer.count( 836 METRIC_CONSUME_TOTAL, 1, {**self._labels, TAG_OUTCOME: settlement.outcome} 837 ) 838 839 if settlement.outcome == OUTCOME_ACKED: 840 await delivery.ack() 841 return 842 843 if settlement.outcome == OUTCOME_PARKED: 844 # ``_park`` counts it, exactly as it does for a body that would not 845 # decode. There is one parked counter and one parked queue whether 846 # the codec or the handler was the one that could not read it. 847 await self._park(delivery, envelope, settlement.reason or "") 848 return 849 850 if wait is None: 851 await self._dead_letter(delivery, envelope, settlement.reason or "") 852 return 853 854 if settlement.outcome == OUTCOME_IN_PROGRESS: 855 await self._put_back(delivery, envelope, wait.delay) 856 return 857 858 if wait.in_broker and await self._retry_in_broker(delivery, envelope, wait.delay): 859 return 860 861 if wait.delay > ZERO: 862 # Waiting here holds the delivery, and so holds one of this 863 # consumer's prefetch slots. For a wait of a few seconds that is the 864 # cheaper of the two costs; for a longer one it is not, which is why 865 # the policy has a threshold and the long waits went to a rung queue 866 # a few lines above. 867 await asyncio.sleep(wait.delay.total_seconds()) 868 869 await self._retry_again(delivery, envelope, wait.delay) 870 871 def _exhausted(self, envelope: Envelope) -> str: 872 """Why there is no next attempt, in words an operator can act on.""" 873 if envelope.attempt >= self._retry.max_attempts: 874 attempts = self._retry.max_attempts 875 return f"exhausted {attempts} attempt{'' if attempts == 1 else 's'}" 876 return f"exceeded the maximum message age of {self._retry.max_message_age}" 877 878 async def _retry_in_broker( 879 self, delivery: Delivery, envelope: Envelope, delay: timedelta 880 ) -> bool: 881 """Parks the message on ``{queue}.retry.{delay}`` and lets the broker 882 return it. 883 884 The rung's ``x-message-ttl`` is the delay and its dead-letter target is 885 this queue, so the wait costs nothing here: no delivery held, no prefetch 886 slot spent, and — the reason it exists — nothing lost when this process 887 restarts halfway through. A consumer sleeping on a five-minute backoff 888 that dies at minute one does not resume at minute one; the broker 889 redelivers the unacknowledged message immediately, and the policy that 890 said five minutes delivers in none. 891 892 The obvious simplification is one rung queue and a per-message 893 ``expiration`` instead of several with fixed TTLs. It does not work. 894 RabbitMQ only ever expires the message at the *head* of a queue, so a 895 message with a ten-minute TTL at the front holds back every thirty-second 896 one behind it, and the delays that come out bear no resemblance to the 897 ones that went in. Never set a per-message TTL for this. 898 899 The attempt advances here exactly as it does on an immediate retry: the 900 counter belongs to the message, and a message that has been round the 901 broker is no less on its second attempt than one that waited here. 902 903 :returns: whether the rung took it. ``False`` means no such queue, and 904 the caller should fall back to waiting here 905 """ 906 rung = naming.retry_queue(self._queue, delay) 907 next_attempt = envelope.with_(attempt=envelope.attempt + 1) 908 try: 909 routed = await self._republish(delivery, rung, next_attempt) 910 except _HopRefusedError: 911 # The rung is there and would not take it. Requeued at once, with 912 # nothing else on this consumer settled, rather than waited out 913 # here: the broker is the one saying no, and holding the delivery 914 # for the whole delay would stall this consumer for nothing. 915 self._observer.count(METRIC_RETRIED_TOTAL, 1, {**self._labels, "where": "requeued"}) 916 await delivery.nack(True) 917 return True 918 if not routed: 919 # Kept even though a consumer declares its own rungs when it starts, 920 # because the rung can still be missing: a consumer started with 921 # ``declare=False`` declares nothing, and a rung deleted under a 922 # running consumer is gone whoever declared it. What has changed is 923 # what it means — it used to be an ordinary topology mistake and is 924 # now either a deliberate opt-out or somebody removing queues. 925 # 926 # Degraded rather than fatal: the message is still deliverable, and 927 # waiting for it here is what this library did before there were 928 # rungs. Counted as well as logged, because this is the one failure 929 # here that nothing else shows: the message is still retried and the 930 # wait still happens, so a dashboard reads as normal while the reason 931 # the rung exists is quietly gone. 932 self._observer.count(METRIC_RUNG_MISSING, 1, {**self._labels, "rung": rung}) 933 log.error( 934 "acemq: %s is not on the broker, so %s will wait %s in this consumer " 935 "instead; declare it with Topology().queue(%r, retry=policy), or let " 936 "the consumer declare it by leaving declare=True", 937 rung, 938 envelope.id, 939 delay, 940 self._queue, 941 ) 942 return False 943 944 self._observer.count(METRIC_RETRIED_TOTAL, 1, {**self._labels, "where": "broker"}) 945 log.info( 946 "acemq: retrying %s from %s, attempt %d of %d, after %s in %s", 947 envelope.id, 948 self._queue, 949 next_attempt.attempt, 950 self._retry.max_attempts, 951 delay, 952 rung, 953 ) 954 await delivery.ack() 955 return True 956 957 async def _retry_again( 958 self, delivery: Delivery, envelope: Envelope, delay: timedelta 959 ) -> None: 960 """Puts the message back on its own queue, one attempt further on. 961 962 Republished rather than requeued, because a requeue returns the bytes 963 the broker was given: the attempt header would still read what the 964 publisher wrote however many times the message had come round, and the 965 count would live only in this process's memory, which is the one place 966 it is lost when the process that has been failing restarts. 967 968 The cost is that the message goes to the back of the queue rather than 969 the front, so a retry is no longer in order with its neighbours. For a 970 message that has already failed once, that is the better trade. 971 """ 972 next_attempt = envelope.with_(attempt=envelope.attempt + 1) 973 try: 974 delivered = await self._republish(delivery, self._queue, next_attempt) 975 except _HopRefusedError: 976 delivered = False 977 if not delivered: 978 # The queue we are consuming has gone. Requeued rather than 979 # acknowledged, because dropping it here would lose a message over a 980 # broker change nobody told this consumer about. 981 # 982 # Counted as a retry all the same, and it is one: the settlement 983 # said ``retried``, the span says ``retried``, and the broker is 984 # about to hand the message back. Leaving it uncounted was the one 985 # place where the counters and the span disagreed about the same 986 # delivery. The label says which path it took, because a requeue 987 # keeps the attempt header it arrived with and the other two do not. 988 self._observer.count(METRIC_RETRIED_TOTAL, 1, {**self._labels, "where": "requeued"}) 989 log.error( 990 "acemq: cannot republish %s onto %s for attempt %d; returning it to the broker", 991 envelope.id, 992 self._queue, 993 next_attempt.attempt, 994 ) 995 await delivery.nack(True) 996 return 997 998 self._observer.count(METRIC_RETRIED_TOTAL, 1, {**self._labels, "where": "consumer"}) 999 log.info( 1000 "acemq: retrying %s from %s, attempt %d of %d, after %s", 1001 envelope.id, 1002 self._queue, 1003 next_attempt.attempt, 1004 self._retry.max_attempts, 1005 delay, 1006 ) 1007 await delivery.ack() 1008 1009 async def _put_back( 1010 self, delivery: Delivery, envelope: Envelope, delay: timedelta 1011 ) -> None: 1012 """Returns a message somebody else is working on, attempt unchanged. 1013 1014 :meth:`_retry_again` without the increment: the attempt counts this 1015 message's failures, and a claim held elsewhere is not one of them. 1016 """ 1017 if delay > ZERO: 1018 await asyncio.sleep(delay.total_seconds()) 1019 try: 1020 delivered = await self._republish(delivery, self._queue, envelope) 1021 except _HopRefusedError: 1022 delivered = False 1023 if delivered: 1024 await delivery.ack() 1025 return 1026 log.error( 1027 "acemq: cannot put %s back onto %s; returning it to the broker", 1028 envelope.id, 1029 self._queue, 1030 ) 1031 await delivery.nack(True) 1032 1033 async def _park(self, delivery: Delivery, envelope: Envelope, reason: str) -> None: 1034 """Sends the message to ``{queue}.parked`` with the reason attached. 1035 1036 Where a message goes when it never reached the handler at all. Somebody 1037 has to look at it, and what they need to know first is that it was 1038 unreadable rather than unlucky. 1039 """ 1040 # The same counter a dead-lettering moves, separated by the outcome 1041 # rather than by a metric of its own — which is what Java settled on, and 1042 # what makes "how much is this queue giving up on" one number that can 1043 # then be split into the two problems it is made of. 1044 self._observer.count( 1045 METRIC_DEAD_LETTERED_TOTAL, 1, {**self._labels, TAG_OUTCOME: OUTCOME_PARKED} 1046 ) 1047 await self._set_aside(delivery, envelope, naming.parked_queue(self._queue), reason) 1048 1049 async def _dead_letter(self, delivery: Delivery, envelope: Envelope, reason: str) -> None: 1050 """Sends the message to ``{queue}.dlq`` with the reason attached, and 1051 acknowledges the original. 1052 1053 Acknowledging a message that failed looks wrong and is what makes this 1054 reliable: the message has already been safely republished somewhere 1055 else, so acknowledging the original is removing the copy that has been 1056 dealt with. Rejecting it instead would either requeue it into a hot loop 1057 or, with a dead-letter exchange configured on the queue, send it 1058 somewhere this consumer did not choose and without the reason. 1059 1060 The reason travels as an envelope field, so a consumer of the 1061 dead-letter queue reads it back through the API rather than having to 1062 know the wire header name. 1063 """ 1064 self._observer.count( 1065 METRIC_DEAD_LETTERED_TOTAL, 1066 1, 1067 {**self._labels, TAG_OUTCOME: OUTCOME_DEAD_LETTERED}, 1068 ) 1069 await self._set_aside( 1070 delivery, envelope, naming.dead_letter_queue(self._queue), reason 1071 ) 1072 1073 async def _set_aside( 1074 self, delivery: Delivery, envelope: Envelope, target: str, reason: str 1075 ) -> None: 1076 """Republishes to ``target`` with the reason, then acknowledges.""" 1077 failed = envelope.with_(error=reason) 1078 try: 1079 delivered = await self._republish(delivery, target, failed) 1080 except _HopRefusedError: 1081 delivered = False 1082 if not delivered: 1083 # Rejected rather than acknowledged: without a dead-letter queue to 1084 # put it in, the broker's own dead-lettering is the last thing left 1085 # between this message and nothing. 1086 self._observer.count( 1087 METRIC_SET_ASIDE_FAILED, 1, {**self._labels, TAG_TARGET: target} 1088 ) 1089 log.error( 1090 "acemq: cannot move %s to %s (%s); rejecting it to the broker instead", 1091 envelope.id, 1092 target, 1093 reason, 1094 ) 1095 await delivery.nack(False) 1096 return 1097 1098 log.warning( 1099 "acemq: set aside %s from %s after %d attempts, into %s: %s", 1100 envelope.id, 1101 self._queue, 1102 envelope.attempt, 1103 target, 1104 reason, 1105 ) 1106 await delivery.ack() 1107 1108 async def _republish(self, delivery: Delivery, queue: str, envelope: Envelope) -> bool: 1109 """Sends the original bytes to a queue by name, and says whether they 1110 arrived. 1111 1112 Through the default exchange, which routes to the queue whose name 1113 matches the routing key, and mandatory so that a queue that is not there 1114 is an answer rather than a silence. The body goes back exactly as it 1115 came: re-encoding through a class that has since changed would replace 1116 what was committed with something else. 1117 """ 1118 try: 1119 result = await self._connection.publish_raw( 1120 "", 1121 queue, 1122 Outbound( 1123 body=delivery.body, 1124 content_type=delivery.content_type or "", 1125 message_id=envelope.id, 1126 headers=envelope.to_headers(routing_key=delivery.routing_key), 1127 persistent=True, 1128 mandatory=True, 1129 ), 1130 ) 1131 except PublishError as refused: 1132 # The broker nacked it, or would not take it while blocked. Raised 1133 # as the caller's to settle, because what to do differs: a refused 1134 # rung means requeue now, where a missing rung means wait here. 1135 # Escaping further left the delivery unsettled for the life of the 1136 # channel, which is what Ruby did before 0.7.12. 1137 log.error("acemq: the broker refused %s onto %s: %s", envelope.id, queue, refused) 1138 raise _HopRefusedError(str(refused)) from refused 1139 return result.routed 1140 1141 async def close(self, *, timeout: float | None = DEFAULT_DRAIN_TIMEOUT) -> bool: 1142 """Stops the consumer and waits, for up to ``timeout``, for the handlers 1143 already running. 1144 1145 A message being worked on when this is called is finished and settled, 1146 rather than abandoned for the broker to hand to somebody else. A message 1147 that had been delivered but not started is given back instead: it is the 1148 broker's to hand to another consumer, and holding it here only to work 1149 through a retry delay would make closing take as long as the schedule. 1150 1151 A handler still running when the time is up is cancelled, and its message 1152 is left unsettled — neither acknowledged nor rejected — so the broker 1153 hands it out again once the channel goes. Shutdown is never the reason a 1154 message is lost or dead-lettered. The same happens when the caller 1155 cancels closing, ``asyncio.wait_for`` included, and that cancellation is 1156 raised again once the consumer has been cleaned up. 1157 1158 The subscription is released last, after everything has been settled, 1159 because a settlement travels on the channel its delivery arrived on. 1160 1161 :param timeout: seconds to wait for every running handler together; 1162 ``None`` waits for as long as they take 1163 :returns: ``True`` if every handler finished, ``False`` if the time ran 1164 out first and some were cut off — the consumer is stopped either way 1165 """ 1166 if self._closed: 1167 return True 1168 self._closed = True 1169 1170 running: set[asyncio.Task[None]] = set() 1171 try: 1172 if self._subscription is not None: 1173 await self._subscription.stop() 1174 1175 while True: 1176 try: 1177 pending = self._work.get_nowait() 1178 except asyncio.QueueEmpty: 1179 break 1180 if pending is not None: 1181 await pending.nack(True) 1182 1183 for _ in self._workers: 1184 self._work.put_nowait(None) 1185 if self._workers: 1186 _, running = await asyncio.wait(self._workers, timeout=timeout) 1187 finally: 1188 # Reached with workers still running for two reasons: the deadline 1189 # passed, or the caller cancelled us. Either way they are cancelled 1190 # before the channel goes, so nothing settles a message on a channel 1191 # that is closing — and a cancellation that came from the caller is 1192 # not caught here, so it carries on out once this has run. 1193 stragglers = [worker for worker in self._workers if not worker.done()] 1194 for worker in stragglers: 1195 worker.cancel() 1196 if stragglers: 1197 await asyncio.wait(stragglers) 1198 if self._subscription is not None: 1199 await self._subscription.close() 1200 self._connection._untrack(self) 1201 1202 for worker in self._workers: 1203 # A worker that was already cancelled — a task group unwinding, a 1204 # shutdown that cancelled the loop's tasks before closing the 1205 # connection — is not news to report to a caller who is merely 1206 # closing. One that died of its own accord is. 1207 failure = None if worker.cancelled() else worker.exception() 1208 if failure is not None: 1209 raise failure 1210 return not running 1211 1212 async def __aenter__(self) -> Consumer: 1213 return self 1214 1215 async def __aexit__(self, *_: object) -> None: 1216 # Not returned: a true value from here tells the interpreter that the 1217 # exception leaving the block was handled. 1218 await self.close()
A running subscription. Close it to stop.
One is built by Connection.consume() rather than directly, because it
has to be started before it is any use and a half-built consumer that looks
finished is a thing somebody will hold on to.
Starting one declares the queues it will need when a message fails — its
dead-letter queue, its parking lot and its rungs — unless it was told not
to. See acemq_amqp.topology.declare_where_failures_go().
Consumer( connection: Connection, queue: str, handler: Callable[Message, acemq_amqp.Ack | Awaitable[acemq_amqp.Ack]], *, codec: acemq_amqp.Codec, retry: acemq_amqp.RetryPolicy, concurrency: int, declare: bool = True)
463 def __init__( 464 self, 465 connection: Connection, 466 queue: str, 467 handler: Handler, 468 *, 469 codec: Codec, 470 retry: RetryPolicy, 471 concurrency: int, 472 declare: bool = True, 473 ) -> None: 474 if concurrency < 1: 475 raise ValueError(f"acemq: concurrency must be at least 1, got {concurrency}") 476 self._connection = connection 477 self._queue = queue 478 self._handler = handler 479 self._codec = codec 480 self._retry = retry 481 self._concurrency = concurrency 482 self._declare = declare 483 self._work: asyncio.Queue[Delivery | None] = asyncio.Queue() 484 self._workers: list[asyncio.Task[None]] = [] 485 self._subscription: Subscription | None = None 486 self._in_flight = 0 487 self._closed = False
queue: str
489 @property 490 def queue(self) -> str: 491 """The queue this consumer reads.""" 492 return self._queue
The queue this consumer reads.
closed: bool
494 @property 495 def closed(self) -> bool: 496 """Whether this consumer has been stopped.""" 497 return self._closed
Whether this consumer has been stopped.
running: bool
499 @property 500 def running(self) -> bool: 501 """Whether this consumer is subscribed and still has workers to hand a 502 delivery to. 503 504 What a health check reads. A consumer whose workers have all finished 505 without it being closed is a consumer the broker is still sending 506 messages to and nothing is reading — which looks identical to a quiet 507 queue from outside, and is the failure worth telling a probe about. 508 """ 509 return ( 510 not self._closed 511 and self._subscription is not None 512 and any(not worker.done() for worker in self._workers) 513 )
Whether this consumer is subscribed and still has workers to hand a delivery to.
What a health check reads. A consumer whose workers have all finished without it being closed is a consumer the broker is still sending messages to and nothing is reading — which looks identical to a quiet queue from outside, and is the failure worth telling a probe about.
in_flight: int
515 @property 516 def in_flight(self) -> int: 517 """How many messages this consumer is working on right now.""" 518 return self._in_flight
How many messages this consumer is working on right now.
async def
close(self, *, timeout: float | None = 20.0) -> bool:
1141 async def close(self, *, timeout: float | None = DEFAULT_DRAIN_TIMEOUT) -> bool: 1142 """Stops the consumer and waits, for up to ``timeout``, for the handlers 1143 already running. 1144 1145 A message being worked on when this is called is finished and settled, 1146 rather than abandoned for the broker to hand to somebody else. A message 1147 that had been delivered but not started is given back instead: it is the 1148 broker's to hand to another consumer, and holding it here only to work 1149 through a retry delay would make closing take as long as the schedule. 1150 1151 A handler still running when the time is up is cancelled, and its message 1152 is left unsettled — neither acknowledged nor rejected — so the broker 1153 hands it out again once the channel goes. Shutdown is never the reason a 1154 message is lost or dead-lettered. The same happens when the caller 1155 cancels closing, ``asyncio.wait_for`` included, and that cancellation is 1156 raised again once the consumer has been cleaned up. 1157 1158 The subscription is released last, after everything has been settled, 1159 because a settlement travels on the channel its delivery arrived on. 1160 1161 :param timeout: seconds to wait for every running handler together; 1162 ``None`` waits for as long as they take 1163 :returns: ``True`` if every handler finished, ``False`` if the time ran 1164 out first and some were cut off — the consumer is stopped either way 1165 """ 1166 if self._closed: 1167 return True 1168 self._closed = True 1169 1170 running: set[asyncio.Task[None]] = set() 1171 try: 1172 if self._subscription is not None: 1173 await self._subscription.stop() 1174 1175 while True: 1176 try: 1177 pending = self._work.get_nowait() 1178 except asyncio.QueueEmpty: 1179 break 1180 if pending is not None: 1181 await pending.nack(True) 1182 1183 for _ in self._workers: 1184 self._work.put_nowait(None) 1185 if self._workers: 1186 _, running = await asyncio.wait(self._workers, timeout=timeout) 1187 finally: 1188 # Reached with workers still running for two reasons: the deadline 1189 # passed, or the caller cancelled us. Either way they are cancelled 1190 # before the channel goes, so nothing settles a message on a channel 1191 # that is closing — and a cancellation that came from the caller is 1192 # not caught here, so it carries on out once this has run. 1193 stragglers = [worker for worker in self._workers if not worker.done()] 1194 for worker in stragglers: 1195 worker.cancel() 1196 if stragglers: 1197 await asyncio.wait(stragglers) 1198 if self._subscription is not None: 1199 await self._subscription.close() 1200 self._connection._untrack(self) 1201 1202 for worker in self._workers: 1203 # A worker that was already cancelled — a task group unwinding, a 1204 # shutdown that cancelled the loop's tasks before closing the 1205 # connection — is not news to report to a caller who is merely 1206 # closing. One that died of its own accord is. 1207 failure = None if worker.cancelled() else worker.exception() 1208 if failure is not None: 1209 raise failure 1210 return not running
Stops the consumer and waits, for up to timeout, for the handlers
already running.
A message being worked on when this is called is finished and settled, rather than abandoned for the broker to hand to somebody else. A message that had been delivered but not started is given back instead: it is the broker's to hand to another consumer, and holding it here only to work through a retry delay would make closing take as long as the schedule.
A handler still running when the time is up is cancelled, and its message
is left unsettled — neither acknowledged nor rejected — so the broker
hands it out again once the channel goes. Shutdown is never the reason a
message is lost or dead-lettered. The same happens when the caller
cancels closing, asyncio.wait_for included, and that cancellation is
raised again once the consumer has been cleaned up.
The subscription is released last, after everything has been settled, because a settlement travels on the channel its delivery arrived on.
Parameters
- timeout: seconds to wait for every running handler together;
Nonewaits for as long as they take :returns:Trueif every handler finished,Falseif the time ran out first and some were cut off — the consumer is stopped either way
class
Connection:
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()
A connection to a broker, and the defaults everything on it inherits.
Built by connect() in the ordinary case. Taking a transport directly
is for a fake in a test, or for a connection made with settings a URL cannot
carry.
Parameters
- transport: how to reach the broker
- codec: what publishers and consumers use unless they say otherwise
- origin: what to stamp on published messages, conventionally
service@host - retry: what consumers use unless they say otherwise. The default is
one delivery and no second chance, so a handler that calls
retry()on a connection with no policy dead-letters the message and says so in the reason. That is deliberate: the alternative default is an immediate requeue, which is a hot loop against a broker that nobody asked for - prefetch: how many unacknowledged messages a consumer holds
- max_outstanding_publishes: how many publishes may be waiting for the
broker at once. The ceiling that turns a memory leak into backpressure:
without one, a caller publishing faster than the broker answers
accumulates unconfirmed messages until the process dies, which looks
like throughput right up to the moment it does not. A thousand by
default, which is what Java's
maxOutstandingPublishesand .NET'sMaxOutstandingPublishesdefault to - confirm_timeout: how long a publish waits for one of those permits,
and then for the broker's confirm, before giving up. Ten seconds by
default, the same as Java's
confirmTimeout. Reaching it raises rather than waiting on, because a publisher stalled for ever behind a broker that has stopped answering is the failure this bound exists to make visible - on_publish: cross-cutting behaviour to wrap every publish on this
connection in, outermost first. See
acemq_amqp.interceptors - on_consume: the same around every handler.
intercept_publish()andintercept_consume()add more later - observer: where the numbers go. Nothing by default, so a service
that wants none pays for none. See
acemq_amqp.telemetry
Connection( transport: acemq_amqp.Transport, *, codec: acemq_amqp.Codec | None = None, origin: str | None = None, retry: acemq_amqp.RetryPolicy | None = None, prefetch: int = 20, max_outstanding_publishes: int = 1000, confirm_timeout: datetime.timedelta = datetime.timedelta(seconds=10), on_publish: Sequence[Callable[[acemq_amqp.PublishContext, Callable[[acemq_amqp.PublishContext], Awaitable[acemq_amqp.PublishResult]]], Awaitable[acemq_amqp.PublishResult]]] = (), on_consume: Sequence[Callable[[acemq_amqp.ConsumeContext, Callable[[acemq_amqp.ConsumeContext], Awaitable[acemq_amqp.Ack]]], Awaitable[acemq_amqp.Ack]]] = (), observer: acemq_amqp.Observer | None = None)
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
transport: acemq_amqp.Transport
1303 @property 1304 def transport(self) -> Transport: 1305 """The transport underneath, for anything this library does not wrap.""" 1306 return self._transport
The transport underneath, for anything this library does not wrap.
codec: acemq_amqp.Codec
1308 @property 1309 def codec(self) -> Codec: 1310 """The codec publishers and consumers use unless told otherwise.""" 1311 return self._codec
The codec publishers and consumers use unless told otherwise.
origin: str
1313 @property 1314 def origin(self) -> str: 1315 """What published messages are stamped with.""" 1316 return self._origin
What published messages are stamped with.
retry: acemq_amqp.RetryPolicy
1318 @property 1319 def retry(self) -> RetryPolicy: 1320 """The retry policy consumers use unless told otherwise.""" 1321 return self._retry
The retry policy consumers use unless told otherwise.
observer: acemq_amqp.Observer
1323 @property 1324 def observer(self) -> Observer: 1325 """Where this connection's numbers go.""" 1326 return self._observer
Where this connection's numbers go.
max_outstanding_publishes: int
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
How many publishes may be waiting for the broker at once.
confirm_timeout: datetime.timedelta
1333 @property 1334 def confirm_timeout(self) -> timedelta: 1335 """How long a publish waits for room before giving up.""" 1336 return self._confirm_timeout
How long a publish waits for room before giving up.
outstanding_publishes: int
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
How many publishes are waiting for the broker right now.
What a stalled publisher looks like from outside: a number pinned at
max_outstanding_publishes is a broker that has stopped answering,
and it says so before the first ~acemq_amqp.PublishError does.
consumers: tuple[Consumer, ...]
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)
The consumers running on this connection.
A snapshot, and public because a health check that has to say whether the consumers are alive should not need private access to find them — the same rule an interceptor follows.
closed: bool
1361 @property 1362 def closed(self) -> bool: 1363 """Whether this connection has been closed.""" 1364 return self._closed
Whether this connection has been closed.
blocked: bool | None
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
Whether the broker has blocked this connection, or None when the
transport cannot be asked.
RabbitMQ blocks a connection when it is low on memory or disk, and while
it lasts it stops reading that connection's socket: everything sent —
a publish, a declaration, health()'s own probe — waits rather than
failing. It is the broker protecting itself and it is not an error, so
it is worth telling apart from the broker having gone away, which looks
identical from outside and wants the opposite response.
None means the question could not be asked at all, which is a
different fact from False and is reported as one: a transport that
does not implement
~acemq_amqp.transport.BlockedState, or one whose client has
moved the state out of reach. See blocked_reason for why there
may be a boolean and no reason to go with it.
blocked_reason: str | None
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
What the broker said when it blocked this connection, if anything.
None when the connection is not blocked, and also None when the
transport is blocked but cannot say why — which is what the RabbitMQ
transport answers, because aio-pika reads RabbitMQ's reason and does not
keep it. Reporting the block without the reason is the honest half;
inventing a reason would read exactly like one the broker sent.
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)
Whether this connection can reach its broker, and its consumers run.
The broker half is a declaration of a temporary queue, which is the cheapest thing AMQP offers that actually proves the connection works. A TCP connection that is open but wedged — the broker paused, the network black-holing — answers a socket-level check exactly as a healthy one does, right up until something is asked of it.
The consumer half is the one a probe usually wants and nothing outside can see. A consumer whose workers have died without it being closed is one the broker is still sending messages to and nothing is reading, and from outside that is indistinguishable from a quiet queue.
A blocked connection is up, with the reason. The broker blocks a
connection when it is low on disk or memory, and an application that
fails its own readiness check for it is one an orchestrator restarts
into the same blocked broker, having thrown away whatever it was
holding. parts["blocked"] carries the state either way, and is
None when the transport could not be asked — see blocked.
The probe is bounded. A blocked broker stops reading the socket, so the round trip this makes is precisely the thing that does not come back when something is wrong; without a deadline of its own a health check hangs, and a readiness endpoint that hangs takes the instance out of rotation with no report at all. A probe that runs out of time is abandoned rather than waited for, because a request that cannot be answered cannot reliably be cancelled either.
It costs a round trip, so it is not something to call per request. Wire it to a readiness probe and let the probe's interval decide how often.
Parameters
- timeout: how long to give the broker's answer, three seconds by
default. Shorter than the deadline
~acemq_amqp.telemetry.aggregate_health()puts around a whole set of checks, so that this report rather than that one is what an operator reads when a broker stops answering :returns: the report. Degraded means working but worth an alert
publish_interceptors: tuple[Callable[[acemq_amqp.PublishContext, Callable[[acemq_amqp.PublishContext], Awaitable[acemq_amqp.PublishResult]]], Awaitable[acemq_amqp.PublishResult]], ...]
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)
What wraps every publish, outermost first.
A snapshot rather than the list itself, so a publish in progress cannot be handed a chain that is being edited underneath it.
consume_interceptors: tuple[Callable[[acemq_amqp.ConsumeContext, Callable[[acemq_amqp.ConsumeContext], Awaitable[acemq_amqp.Ack]]], Awaitable[acemq_amqp.Ack]], ...]
1607 @property 1608 def consume_interceptors(self) -> tuple[ConsumeInterceptor, ...]: 1609 """What wraps every handler, outermost first.""" 1610 return tuple(self._consuming)
What wraps every handler, outermost first.
def
intercept_publish( self, interceptor: Callable[[acemq_amqp.PublishContext, Callable[[acemq_amqp.PublishContext], Awaitable[acemq_amqp.PublishResult]]], Awaitable[acemq_amqp.PublishResult]]) -> Connection:
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
Wraps every publish on this connection in interceptor.
Registered ones run in the order they were added, so this one is innermost until the next is added. Read at publish time rather than copied into each publisher, so an interceptor registered during start-up applies to publishers that already exist — which is what makes it usable from a framework's wiring rather than only from the line that connects.
Adding at start-up is still what to do. A message already in flight will not see it, and an interceptor whose whole job is to stamp every message is not doing it if the first few went without.
Parameters
- interceptor: what to wrap the publish in. See
acemq_amqp.interceptors:returns: this connection, so registrations read as one expression
def
intercept_consume( self, interceptor: Callable[[acemq_amqp.ConsumeContext, Callable[[acemq_amqp.ConsumeContext], Awaitable[acemq_amqp.Ack]]], Awaitable[acemq_amqp.Ack]]) -> Connection:
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
Wraps every handler on this connection in interceptor.
Parameters
- interceptor: what to wrap the handler in :returns: this connection
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)
Applies a topology to this connection's broker.
Parameters
- topology: what the broker should have
def
publisher( self, exchange: str = '', routing_key: str = '', *, codec: acemq_amqp.Codec | None = None, persistent: bool = True, mandatory: bool = False) -> Publisher:
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 )
Builds a publisher for an exchange and routing key.
Publish straight to a queue by leaving the exchange empty: the default exchange routes to the queue whose name matches the routing key.
Parameters
- exchange: where to publish, empty for the default exchange
- routing_key: what to publish under, or a queue name
- codec: a codec other than this connection's
- persistent: ask the broker to write these messages to disk
- mandatory: fail when a message reaches no queue at all, rather than letting the broker drop it silently :returns: the publisher
async def
consume( self, queue: str, handler: Callable[Message, acemq_amqp.Ack | Awaitable[acemq_amqp.Ack]], *, codec: acemq_amqp.Codec | None = None, retry: acemq_amqp.RetryPolicy | None = None, prefetch: int | None = None, concurrency: int = 1, tag: str = '', args: Mapping[str, typing.Any] | None = None, declare: bool = True) -> Consumer:
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
Reads messages from a queue until the returned consumer is closed.
Before it subscribes it declares the queues it will need when a message
fails: {queue}.dlq, {queue}.parked, acemq.dlx and, when the
retry policy has waits the broker holds, acemq.retry and the rungs.
See acemq_amqp.topology.declare_where_failures_go() for exactly
what and why. The source queue is not among them, and neither is
anything a producer needs, so this is not a replacement for declaring a
topology — it is the floor underneath a service that was deployed
without one.
Parameters
- queue: what to read
- handler: what to do with a message; returning an
~acemq_amqp.Ackis the decision - codec: a codec other than this connection's
- retry: a policy other than this connection's
- prefetch: how many unacknowledged messages to hold
- concurrency: how many messages to work on at once. One by default, which keeps a queue's messages in order; raising it trades that order for throughput, which is the right trade for handlers that spend their time waiting on something else
- tag: what to call this consumer to the broker
- args: broker-specific consumer arguments
- declare: declare the queues above before subscribing. On by
default, because the alternative default loses messages silently.
Turn it off for a login with no
configurepermission on the vhost, which is refused by the broker rather than ignored, and for a tool draining a queue it does not own — a one-off reader oforders.dlqhas no business creatingorders.dlq.dlq. A consumer started this way still publishes to those queues, so somebody else has to have declared them :returns: the running consumer
async def
publish_raw( self, exchange: str, routing_key: str, message: acemq_amqp.Outbound) -> acemq_amqp.PublishResult:
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
Sends bytes that are already encoded, with headers already rendered.
For the retry and dead-letter hops, and for anything else replaying a
message recorded earlier: the payload's class may not exist any more,
and re-encoding it would produce different bytes from the ones that were
committed. Ordinary publishing goes through Publisher.
This is the one place the outstanding-publish bound is taken. Every
publish in the library comes through here — a Publisher, a
retry rung, a dead letter, a replay — so one permit counts them all, and
Publisher.send_all() cannot put more messages on the wire at once
than max_outstanding_publishes allows however long the list it
was given is.
Parameters
- exchange: where to publish, empty for the default exchange
- routing_key: what to publish under
- message: the message, encoded :returns: what the broker said
Raises
- PublishingPausedError: when the broker has blocked this connection. Nothing was sent; retry once it unblocks
- PublishError: when every permit is taken and none came free
within
confirm_timeout, or when the broker did not confirm this publish within it. The second may or may not have arrived
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)
Takes one message off a queue, or None when there is none waiting.
For tools rather than for services: it is a round trip per message where
consume() gets a stream. What it can do that a consumer cannot is
say that a queue is empty, which is what anything working through a
backlog and then stopping — a replay, a drain — has to know.
The delivery comes back unsettled, so the caller decides whether it is gone or goes back.
Parameters
- queue: what to read from
:returns: the delivery, or
None
async def
queue_exists(self, queue: str) -> bool:
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)
Whether a queue is on the broker, creating nothing.
async def
message_count(self, queue: str) -> int:
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)
How many messages are waiting on a queue.
async def
delete_queue(self, queue: str) -> None:
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)
Removes a queue and every message still on it.
async def
close(self, *, timeout: float | None = 20.0) -> bool:
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)
Stops every consumer on this connection and releases it.
Consumers first, and their handlers given up to timeout to finish,
so a message being worked on when this is called is settled rather than
returned to the queue for somebody else to redo. The bound is one
deadline for every handler of every consumer, not one each: they are
all drained at once. A handler still running at the deadline is
cancelled and its message left unsettled, for the broker to redeliver —
see Consumer.close().
The connection is released whatever happens, including when the caller cancels closing; that cancellation is raised again afterwards.
Parameters
- timeout: seconds to wait for the running handlers;
Nonewaits for as long as they take :returns:Trueif every handler finished,Falseif the time ran out first
@dataclass(frozen=True, slots=True)
class
BrokerHealth:
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)
A connection as a ~acemq_amqp.telemetry.HealthCheck.
So that a broker check and an application's own — a database, a downstream
service — can be handed to
~acemq_amqp.telemetry.aggregate_health() together, which is what a
readiness endpoint actually needs.
A blocked broker comes back through here as up, with the reason, and
nothing in between folds it into anything else: the aggregate takes the
worst status it is given, so a check that reported back pressure as degraded
would quietly degrade a report that had decided otherwise. The deadline is
the same argument — timeout is shorter than the aggregate's, so that
a broker which has stopped answering is described by this check rather than
by the aggregate giving up on it.
Parameters
- connection: what to ask
- label: what to call it in a combined report
- timeout: how long to give the broker's answer. See
Connection.health()
BrokerHealth( connection: Connection, label: str = 'broker', timeout: float = 3.0)
connection: Connection
async def
connect( url: str, *, codec: acemq_amqp.Codec | None = None, origin: str | None = None, retry: acemq_amqp.RetryPolicy | None = None, prefetch: int = 20, max_outstanding_publishes: int = 1000, confirm_timeout: datetime.timedelta = datetime.timedelta(seconds=10), security: acemq_amqp.Security | None = None, on_publish: Sequence[Callable[[acemq_amqp.PublishContext, Callable[[acemq_amqp.PublishContext], Awaitable[acemq_amqp.PublishResult]]], Awaitable[acemq_amqp.PublishResult]]] = (), on_consume: Sequence[Callable[[acemq_amqp.ConsumeContext, Callable[[acemq_amqp.ConsumeContext], Awaitable[acemq_amqp.Ack]]], Awaitable[acemq_amqp.Ack]]] = (), observer: acemq_amqp.Observer | None = None, **transport_options: Any) -> Connection:
2012async def connect( 2013 url: str, 2014 *, 2015 codec: Codec | None = None, 2016 origin: str | None = None, 2017 retry: RetryPolicy | None = None, 2018 prefetch: int = DEFAULT_PREFETCH, 2019 max_outstanding_publishes: int = DEFAULT_MAX_OUTSTANDING_PUBLISHES, 2020 confirm_timeout: timedelta = DEFAULT_CONFIRM_TIMEOUT, 2021 security: Security | None = None, 2022 on_publish: Sequence[PublishInterceptor] = (), 2023 on_consume: Sequence[ConsumeInterceptor] = (), 2024 observer: Observer | None = None, 2025 **transport_options: Any, 2026) -> Connection: 2027 """Opens a connection to a broker. 2028 2029 The URL's scheme picks the transport, and says whether the connection is 2030 encrypted: ``amqps://`` is, ``amqp://`` is not. Both need aio-pika, which is 2031 why it is imported here rather than by the package: a program that only 2032 reads AceMQ envelopes should not be made to install an AMQP client to import 2033 this library. 2034 2035 An ``amqps://`` URL is verified against the machine's trust store and will 2036 not speak anything older than TLS 1.2, with or without a ``security``. Pass 2037 one to name a different authority, to present a client certificate, or to 2038 supply the login separately from the URL — which is how a password stays out 2039 of a connection string that ends up in a log. 2040 2041 :param url: where the broker is 2042 :param codec: what publishers and consumers use unless they say otherwise 2043 :param origin: what to stamp on published messages 2044 :param retry: what consumers use unless they say otherwise 2045 :param prefetch: how many unacknowledged messages a consumer holds 2046 :param max_outstanding_publishes: how many publishes may be waiting for the 2047 broker at once, a thousand by default. See :class:`Connection` 2048 :param confirm_timeout: how long a publish waits for room before raising 2049 :param security: how to verify the broker and who to log in as. See 2050 :class:`~acemq_amqp.Security` 2051 :param on_publish: what to wrap every publish in, outermost first. See 2052 :mod:`acemq_amqp.interceptors` 2053 :param on_consume: what to wrap every handler in 2054 :param observer: where the numbers go. See :mod:`acemq_amqp.telemetry` 2055 :param transport_options: passed to the transport, which for RabbitMQ is 2056 :func:`aio_pika.connect_robust` 2057 :returns: the connection 2058 :raises SecurityError: when the security settings cannot be honoured, before 2059 anything is dialled 2060 """ 2061 scheme = urlsplit(url).scheme 2062 if scheme not in ("amqp", "amqps"): 2063 raise AceMQError( 2064 f"acemq: no transport knows how to reach {scheme or url!r}; " 2065 "this library speaks amqp:// and amqps://" 2066 ) 2067 2068 # Resolved here rather than in the transport, so that a missing certificate 2069 # or an unset password variable is an exception from connect() with the 2070 # setting named in it, rather than a handshake failure from somewhere in 2071 # aio-pika that says only that the socket closed. 2072 settings = security or Security() 2073 url = settings.applied_to(url) 2074 # The caller's own options win: transport_options is the escape hatch for 2075 # everything this library does not model, and a Security built from defaults 2076 # should not quietly overrule an ssl_context somebody passed by hand. 2077 transport_options = {**settings.transport_options(url), **transport_options} 2078 2079 try: 2080 from .rabbitmq import RabbitMQTransport 2081 except ImportError as missing: # pragma: no cover - depends on how it was installed 2082 raise AceMQError( 2083 "acemq: reaching a broker needs aio-pika, which is an optional extra. " 2084 'Install it with pip install "acemq-amqp[rabbitmq]"' 2085 ) from missing 2086 2087 transport = await RabbitMQTransport.connect(url, **transport_options) 2088 return Connection( 2089 transport, 2090 codec=codec, 2091 origin=origin, 2092 retry=retry, 2093 prefetch=prefetch, 2094 max_outstanding_publishes=max_outstanding_publishes, 2095 confirm_timeout=confirm_timeout, 2096 on_publish=on_publish, 2097 on_consume=on_consume, 2098 observer=observer, 2099 )
Opens a connection to a broker.
The URL's scheme picks the transport, and says whether the connection is
encrypted: amqps:// is, amqp:// is not. Both need aio-pika, which is
why it is imported here rather than by the package: a program that only
reads AceMQ envelopes should not be made to install an AMQP client to import
this library.
An amqps:// URL is verified against the machine's trust store and will
not speak anything older than TLS 1.2, with or without a security. Pass
one to name a different authority, to present a client certificate, or to
supply the login separately from the URL — which is how a password stays out
of a connection string that ends up in a log.
Parameters
- url: where the broker is
- codec: what publishers and consumers use unless they say otherwise
- origin: what to stamp on published messages
- retry: what consumers use unless they say otherwise
- prefetch: how many unacknowledged messages a consumer holds
- max_outstanding_publishes: how many publishes may be waiting for the
broker at once, a thousand by default. See
Connection - confirm_timeout: how long a publish waits for room before raising
- security: how to verify the broker and who to log in as. See
~acemq_amqp.Security - on_publish: what to wrap every publish in, outermost first. See
acemq_amqp.interceptors - on_consume: what to wrap every handler in
- observer: where the numbers go. See
acemq_amqp.telemetry - transport_options: passed to the transport, which for RabbitMQ is
aio_pika.connect_robust():returns: the connection
Raises
- SecurityError: when the security settings cannot be honoured, before anything is dialled
def
default_origin() -> str:
2102def default_origin() -> str: 2103 """``acemq@{hostname}``, which names the machine but not the service. 2104 2105 Enough to tell two pods apart in a log, which is what an origin is for, and 2106 deliberately not a guess at the service name: a wrong service name in a 2107 message that has already been published is worse than an honest hostname. 2108 """ 2109 return f"acemq@{socket.gethostname()}"
acemq@{hostname}, which names the machine but not the service.
Enough to tell two pods apart in a log, which is what an origin is for, and deliberately not a guess at the service name: a wrong service name in a message that has already been published is worse than an honest hostname.