Add ConsumerSettleStrategy with DirectReplyTo support

Replace ConsumerBuilder.presettled(bool) with settle_strategy(),
accepting EXPLICIT_SETTLE/PRESETTLED/DIRECT_REPLY_TO per the updated
step_060_consumer_strategy.md spec. DIRECT_REPLY_TO attaches to
RabbitMQ's direct-reply-to pseudo-queue with a dynamic source,
resolves the broker-generated address after attach (re-resolving it
on every reconnect, since it's session-scoped), and build() rejects
combining it with queue() or single_active_consumer_state_changed().

Signed-off-by: Gabriele Santomaggio <G.santomaggio@gmail.com>

Author: Gsantomaggio

README.md

@@ -51,11 +51,11 @@ finally:
 
 ## Examples
 
-[`docs/examples/README.md`](docs/examples/README.md) catalogues eleven runnable
+[`docs/examples/README.md`](docs/examples/README.md) catalogues twelve runnable
 scripts, in reading order — from `basic_connection.py` through publishing and
-consuming to auto-reconnection, presettled consumers, rejection reasons,
-single-active-consumer notifications and stream filtering. Each one talks to a
-local broker:
+consuming to auto-reconnection, presettled consumers, direct reply-to,
+rejection reasons, single-active-consumer notifications and stream filtering.
+Each one talks to a local broker:
 
 ```sh
 PYTHONPATH=. .venv/bin/python docs/examples/basic_connection.py

docs/examples/README.md

@@ -20,7 +20,8 @@ one layer at a time, and the rest build on them.
 | [`auto_reconnection.py`](auto_reconnection.py) | A connection surviving a forced socket drop: publishers and consumers built before the drop keep working afterwards, and `RecoveryConfiguration(topology=True)` re-declares the topology the broker lost. |
 | [`console_application.py`](console_application.py) | The whole client end to end — declare, publish, consume, report — as a scriptable smoke test with meaningful exit codes. `--help` lists its options. |
 | [`performance_test.py`](performance_test.py) | A throughput/latency generator: many publishers and consumers on one or more connections, with periodic rate and latency reporting. |
-| [`presettled_consumer_example.py`](presettled_consumer_example.py) | At-most-once consumption: `presettled()` attaches with `snd-settle-mode = settled`, so the broker settles every delivery itself, the handler never touches its `Context`, and `unsettled_message_count` stays at `0` for the consumer's whole life — including across a forced disconnect that auto-reconnection recovers from. |
+| [`presettled_consumer_example.py`](presettled_consumer_example.py) | At-most-once consumption: `settle_strategy(ConsumerSettleStrategy.PRESETTLED)` attaches with `snd-settle-mode = settled`, so the broker settles every delivery itself, the handler never touches its `Context`, and `unsettled_message_count` stays at `0` for the consumer's whole life — including across a forced disconnect that auto-reconnection recovers from. |
+| [`direct_reply_to_example.py`](direct_reply_to_example.py) | Request/reply without a dedicated reply queue: `settle_strategy(ConsumerSettleStrategy.DIRECT_REPLY_TO)` attaches to no queue at all and reads back a broker-generated pseudo-queue address from `consumer.queue`; a request naming it as `properties.reply_to` gets answered by an ordinary queue-bound consumer publishing to that address. The pseudo-queue is session-scoped, so a forced disconnect and auto-reconnection leave the requester with a brand new address, which the script re-reads before sending its second request. |
 | [`rejection_reason_example.py`](rejection_reason_example.py) | Why a publish was refused: a quorum queue with `max_length(5)` and the `reject-publish` overflow strategy returns a `REJECTED` outcome carrying `RejectionDetails.reason` and `.rejected_by_queue` (needs RabbitMQ 4.3+; the script says so when the broker supplies neither). |
 | [`quorum_single_active_consumer.py`](quorum_single_active_consumer.py) | Single active consumer notifications on a quorum queue (needs RabbitMQ 4.3+): two consumers on a queue declared with `single_active_consumer(True)` each register a `single_active_consumer_state_changed` handler, and the broker tells one of them it is active and the other that it is standby. Only the active one receives messages; closing it promotes the standby one, which then starts receiving. |
 | [`stream_filtering.py`](stream_filtering.py) | Reading a stream queue from a chosen offset and narrowing what arrives (needs RabbitMQ 4.2+ for the SQL filter): the same batch of messages is read back three times from `offset(FIRST)` — once with `filter().subject(...).property("region", ...)`, once with the equivalent `filter().sql(...)`, and once with the cheap, probabilistic `filter_values(...)` bloom filter, whose publisher side is just an `x-stream-filter-value` message annotation. The script reports whether the broker really enforces the SQL filter. |

docs/examples/console_application.py

@@ -670,8 +670,8 @@ def build_consumer(connection: Connection, options: Options, counters: Counters)
     """Attach the consumer that counts every delivery (§3.3).
 
     Built before the publisher, so nothing this run publishes can arrive before
-    something is already listening. ``initial_credits`` and ``presettled`` are
-    left at ``ConsumerBuilder``'s own defaults.
+    something is already listening. ``initial_credits`` and ``settle_strategy``
+    are left at ``ConsumerBuilder``'s own defaults (``EXPLICIT_SETTLE``).
 
     Args:
         connection: The connection to attach on.

docs/examples/direct_reply_to_example.py

@@ -0,0 +1,219 @@
+"""Direct reply-to: RabbitMQ's request/reply pseudo-queue (step_060_consumer_strategy.md §7).
+
+Run against a local broker::
+
+    PYTHONPATH=. .venv/bin/python docs/examples/direct_reply_to_example.py
+
+``consumer_builder().settle_strategy(ConsumerSettleStrategy.DIRECT_REPLY_TO)``
+attaches a receiver link to no caller-supplied queue at all: the broker
+dynamically generates a private, per-attach pseudo-queue address
+(``/queues/amq.rabbitmq.reply-to.<opaque-suffix>``) and returns it in the
+``attach`` reply, read back here as ``consumer.queue``. A caller who wants a
+reply puts that address in an outgoing request's ``properties.reply_to``;
+whoever answers just needs to publish to it — no dedicated, exclusive reply
+queue to declare and clean up per requester. Like a presettled consumer, its
+``Context`` methods all raise :class:`~src.ConsumerError`: the broker considers
+every delivery on this link already settled the instant it sends it.
+
+This script plays both roles, on two separate connections:
+
+* the **requester** (``requester_connection``) builds one ``DIRECT_REPLY_TO``
+  consumer and never declares a queue of its own;
+* the **responder** (``responder_connection``) declares an ordinary request
+  queue, consumes it with a normal ``ExplicitSettle`` consumer, and replies to
+  whatever address arrived in the request's ``reply_to`` — an ordinary
+  anonymous :class:`~src.Publisher` (step_020_publishers.md §3.3) also sends
+  the request itself, targeting the request queue.
+
+Because the pseudo-queue is scoped to the exact connection and session that
+attached it (§3.3 point 5), it does not survive a reconnect: the second half of
+this script forces the requester's socket down, lets auto-reconnection redial
+and re-attach, and shows ``consumer.queue`` reading back a *different*
+broker-generated address afterward — a caller must always re-read it after
+recovery, never cache it. Tearing down the live socket
+(``connection._socket.shutdown``) reaches into the client on purpose;
+application code never does this.
+"""
+
+from __future__ import annotations
+
+import logging
+import queue
+import socket
+import time
+import uuid
+
+from src import (
+    Connection,
+    ConnectionParameters,
+    ConnectionState,
+    Consumer,
+    ConsumerSettleStrategy,
+    Context,
+    Message,
+    Publisher,
+    RecoveryConfiguration,
+    queue_address,
+)
+from src.wire import Properties
+
+#: How long the example waits for a reply it expects.
+TIMEOUT_SECONDS = 15.0
+
+#: How long it waits for auto-reconnection to finish.
+RECOVERY_TIMEOUT_SECONDS = 60.0
+
+POLL_INTERVAL_SECONDS = 0.05
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)-7s %(name)s: %(message)s")
+logger = logging.getLogger("example")
+
+
+class ReplyBox:
+    """Records every reply the requester's direct-reply-to consumer receives."""
+
+    def __init__(self) -> None:
+        """Start with nothing received."""
+        self.replies: queue.Queue[str] = queue.Queue()
+
+    def on_message(self, context: Context, message: Message) -> None:
+        """Log and record one reply, without touching ``context`` — it is presettled."""
+        body = message.body_as_string()
+        logger.info("requester received reply %r (presettled=%s)", body, context.is_presettled)
+        self.replies.put(body)
+
+    def drain_one(self) -> str:
+        """Return the next reply body, or raise once the wait times out."""
+        return self.replies.get(timeout=TIMEOUT_SECONDS)
+
+
+def wait_for(connection: Connection, state: ConnectionState, timeout: float) -> bool:
+    """Whether ``connection`` is seen in ``state`` within ``timeout`` seconds."""
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if connection.state is state:
+            return True
+        time.sleep(POLL_INTERVAL_SECONDS)
+    return False
+
+
+def build_responder(connection: Connection, request_queue: str) -> Consumer:
+    """Attach the queue-bound consumer that answers every request with a reply."""
+
+    def on_request(context: Context, message: Message) -> None:
+        reply_to = message.properties.reply_to if message.properties else None
+        if not reply_to:
+            logger.warning("a request with no reply_to arrived; discarding it")
+            context.discard()
+            return
+        logger.info("responder received request %r, replying to %r", message.body_as_string(), reply_to)
+        reply_publisher = connection.publisher_builder().build()  # anonymous: reply_to is per-request
+        try:
+            reply_publisher.publish(
+                Message(
+                    f"pong-for-{message.body_as_string()}",
+                    properties=Properties(
+                        to=reply_to,
+                        correlation_id=message.properties.message_id if message.properties else None,
+                    ),
+                ),
+                timeout=TIMEOUT_SECONDS,
+            )
+        finally:
+            reply_publisher.close()
+        context.accept()
+
+    return connection.consumer_builder().queue(request_queue).message_handler(on_request).build()
+
+
+def send_request(connection: Connection, request_queue: str, reply_to: str, body: str) -> None:
+    """Publish one request naming ``reply_to``, from a fresh anonymous publisher.
+
+    Anonymous (step_020_publishers.md §3.3) so each request can carry its own
+    ``to`` alongside its ``reply_to`` — a queue-bound publisher could not set
+    the former, since its target is fixed at ``build()`` time.
+    """
+    publisher: Publisher = connection.publisher_builder().build()
+    try:
+        publisher.publish(
+            Message(
+                body,
+                properties=Properties(
+                    message_id=str(uuid.uuid4()),
+                    to=queue_address(request_queue),
+                    reply_to=reply_to,
+                ),
+            ),
+            timeout=TIMEOUT_SECONDS,
+        )
+    finally:
+        publisher.close()
+    logger.info("requester sent %r to %r with reply_to=%r", body, request_queue, reply_to)
+
+
+def direct_reply_to() -> None:
+    """Round-trip a request/reply exchange over direct-reply-to, across a forced disconnect."""
+    requester_connection = Connection(
+        ConnectionParameters(
+            container_id=f"example-direct-reply-to-requester-{uuid.uuid4().hex[:8]}",
+            on_unexpected_close=lambda error: logger.error("the requester connection died for good: %s", error),
+            recovery_configuration=RecoveryConfiguration(),  # activated=True, topology=False
+        )
+    )
+    responder_connection = Connection(
+        ConnectionParameters(container_id=f"example-direct-reply-to-responder-{uuid.uuid4().hex[:8]}")
+    )
+    request_queue = f"example-direct-reply-to-{uuid.uuid4().hex[:8]}"
+    reply_box = ReplyBox()
+    try:
+        responder_connection.management().queue(request_queue).declare()
+        logger.info("declared the request queue %r", request_queue)
+
+        responder = build_responder(responder_connection, request_queue)
+        requester = (
+            requester_connection.consumer_builder()
+            .message_handler(reply_box.on_message)
+            .settle_strategy(ConsumerSettleStrategy.DIRECT_REPLY_TO)
+            .build()
+        )
+        try:
+            first_address = requester.queue
+            assert first_address is not None, "build() only returns once DIRECT_REPLY_TO's address is resolved"
+            logger.info("requester attached to the broker-generated pseudo-queue %r", first_address)
+            send_request(responder_connection, request_queue, first_address, "hello")
+            logger.info("requester received %r", reply_box.drain_one())
+
+            logger.info("simulating a network failure on the requester connection")
+            requester_connection._socket.shutdown(socket.SHUT_RDWR)
+            if not wait_for(requester_connection, ConnectionState.RECONNECTING, timeout=10.0):
+                logger.warning("never observed RECONNECTING — detection and recovery both landed inside one poll")
+            if not wait_for(requester_connection, ConnectionState.OPEN, RECOVERY_TIMEOUT_SECONDS):
+                logger.error("the requester connection never recovered")
+                return
+            logger.info(
+                "state is back to %s, the receiver link re-attached underneath the same Consumer object",
+                requester_connection.state.value,
+            )
+
+            second_address = requester.queue
+            assert second_address is not None, "the re-attach must have resolved a fresh address"
+            logger.info("requester re-attached to a fresh pseudo-queue %r", second_address)
+            assert second_address != first_address, (
+                "a direct-reply-to pseudo-queue is session-scoped and must not survive a reconnect"
+            )
+            send_request(responder_connection, request_queue, second_address, "hello-again")
+            logger.info("requester received %r", reply_box.drain_one())
+        finally:
+            requester.close()
+            responder.close()
+            logger.info("closed the requester and the responder")
+
+        responder_connection.management().queue(request_queue).delete()
+        logger.info("deleted the request queue %r", request_queue)
+    finally:
+        requester_connection.close()
+        responder_connection.close()
+
+
+if __name__ == "__main__":
+    direct_reply_to()

docs/examples/performance_test.py

@@ -984,9 +984,9 @@ def build_consumer(
 
     Built before the publisher, so nothing this run publishes can arrive before
     something is already listening. ``initial_credits`` comes from
-    ``--initial-credits``; ``presettled`` is left at ``ConsumerBuilder``'s own
-    default of ``False``, since a presettled run's latency would exclude the
-    disposition round-trip this program means to measure (§10).
+    ``--initial-credits``; ``settle_strategy`` is left at ``ConsumerBuilder``'s
+    own default of ``EXPLICIT_SETTLE``, since a presettled run's latency would
+    exclude the disposition round-trip this program means to measure (§10).
 
     Args:
         connection: The connection to attach on.

docs/examples/presettled_consumer_example.py

@@ -1,14 +1,15 @@
-"""Presettled consumption: at-most-once delivery with no dispositions (step_060 §6).
+"""Presettled consumption: at-most-once delivery with no dispositions (step_060_consumer_strategy.md §7).
 
 Run against a local broker::
 
     PYTHONPATH=. .venv/bin/python docs/examples/presettled_consumer_example.py
 
-``consumer_builder().presettled()`` attaches the receiver link with
-``snd-settle-mode = settled``. The broker then considers every delivery settled
-the moment it puts it on the wire: it forgets the message immediately, and the
-client never sends a ``disposition`` back. That is the whole trade — one frame
-per message instead of two, and no redelivery if the consumer dies holding one.
+``consumer_builder().settle_strategy(ConsumerSettleStrategy.PRESETTLED)`` attaches
+the receiver link with ``snd-settle-mode = settled``. The broker then considers
+every delivery settled the moment it puts it on the wire: it forgets the message
+immediately, and the client never sends a ``disposition`` back. That is the
+whole trade — one frame per message instead of two, and no redelivery if the
+consumer dies holding one.
 
 Two consequences show up directly in the API and are asserted below:
 
@@ -20,12 +21,13 @@
 * ``consumer.unsettled_message_count`` stays at ``0`` for the consumer's whole
   life, because nothing is ever outstanding.
 
-The second half of the script is the reconnection requirement of step_060 §6:
-the socket is torn down underneath the connection, the default
-``RecoveryConfiguration`` redials and re-attaches both links, and the *same*
-publisher and consumer objects keep working — still presettled, still settling
-nothing. Tearing down the live socket (``connection._socket.shutdown``) reaches
-into the client on purpose; application code never does this.
+The second half of the script is the reconnection requirement of
+step_060_consumer_strategy.md §7: the socket is torn down underneath the
+connection, the default ``RecoveryConfiguration`` redials and re-attaches both
+links, and the *same* publisher and consumer objects keep working — still
+presettled, still settling nothing. Tearing down the live socket
+(``connection._socket.shutdown``) reaches into the client on purpose;
+application code never does this.
 """
 
 from __future__ import annotations
@@ -41,6 +43,7 @@
     ConnectionParameters,
     ConnectionState,
     Consumer,
+    ConsumerSettleStrategy,
     Context,
     Message,
     Publisher,
@@ -117,7 +120,13 @@ def presettled_consumer() -> None:
         connection.management().queue(name).quorum().queue().declare()
         logger.info("declared the quorum queue %r", name)
 
-        consumer = connection.consumer_builder().queue(name).message_handler(tally.on_message).presettled().build()
+        consumer = (
+            connection.consumer_builder()
+            .queue(name)
+            .message_handler(tally.on_message)
+            .settle_strategy(ConsumerSettleStrategy.PRESETTLED)
+            .build()
+        )
         logger.info("consuming from %r presettled=%s", consumer.queue, consumer.is_presettled)
 
         publisher = connection.publisher_builder().queue(name).build()

src/__init__.py

@@ -14,6 +14,7 @@
 from .consumer import (
     Consumer,
     ConsumerBuilder,
+    ConsumerSettleStrategy,
     Context,
     MessageHandler,
     QuorumConsumerOptions,
@@ -157,6 +158,7 @@
     "RejectionDetails",
     "Consumer",
     "ConsumerBuilder",
+    "ConsumerSettleStrategy",
     "Context",
     "MessageHandler",
     "QuorumConsumerOptions",

src/constants.py

@@ -16,6 +16,9 @@
 # --- Single active consumer (quorum queues) ---
 RABBITMQ_ACTIVE_PROPERTY = "rabbitmq:active"
 
+# --- Direct reply-to (step_060 §3.3) ---
+DIRECT_REPLY_TO_CAPABILITY = "rabbitmq:volatile-queue"
+
 # --- Stream offset / filtering symbols (Source.filter map keys) ---
 STREAM_OFFSET_SPEC_FILTER = "rabbitmq:stream-offset-spec"
 STREAM_FILTER_VALUES_FILTER = "rabbitmq:stream-filter"