Add opt-in receipt-confirmed sends

README.md

@@ -24,14 +24,10 @@ async with stompman.Client(
         stompman.ConnectionParameters(host="171.0.0.1", port=61616, login="user1", passcode="passcode1"),
         stompman.ConnectionParameters(host="172.0.0.1", port=61616, login="user2", passcode="passcode2"),
     ],
-
-
     # SSL — can be either `None` (default), `True`, or `ssl.SSLContext'
     ssl=None,
-
     # Error frame handler:
     on_error_frame=lambda error_frame: print(error_frame.body),
-
     # Optional parameters with sensible defaults:
     heartbeat=stompman.Heartbeat(will_send_interval_ms=1000, want_to_receive_interval_ms=1000),
     connect_retry_attempts=3,
@@ -81,6 +77,41 @@ async with client.begin() as transaction:
         await asyncio.sleep(0.1)
 ```
 
+#### Confirming sends
+
+By default `client.send()` returns once the frame has been written to the socket. That says nothing about whether the
+broker accepted it. Pass `receipt_timeout` to wait for the broker's
+[STOMP receipt](https://stomp.github.io/stomp-specification-1.2.html#RECEIPT) instead:
+
+```python
+await client.send(b"hi there!", destination="DLQ", headers={"persistent": "true"}, receipt_timeout=3.0)
+```
+
+The default `receipt_timeout=None` preserves the existing write-only behavior. A timeout must be finite and positive.
+It covers writing `SEND` and waiting for its receipt, after a connection is available. The client generates the
+`receipt` header itself, so passing your own `receipt` header with `receipt_timeout` is an error.
+
+A failure raises `stompman.SendReceiptError`, with `reason` equal to `rejected`, `timeout`, or `connection_lost`.
+Raw broker error frames are available through `error.frame`, but are excluded from the exception's representation.
+
+Only `rejected` is a definitive answer. On `timeout` and `connection_lost` the outcome is **ambiguous**: the frame is
+fully buffered before the socket is drained, so the broker may well have accepted the message even though the
+confirmation was lost. A confirmed send still waits for a connection through `connect_retry_attempts`, but the write
+itself is attempted exactly once: `write_retry_attempts` does not apply, because a connection can fail after the frame
+reached the broker and no connection type can tell the two apart. Nothing is silently replayed on a new connection —
+retrying is your decision. Give each application event a stable ID and deduplicate in the broker or downstream
+consumer; `JMSCorrelationID` alone does not deduplicate anything.
+
+A receipt means the broker took ownership of the message. It does not mean any consumer has received or processed it.
+
+Concurrent confirmed sends never consume each other's receipts, and a receipt is bound to the connection it was issued
+on, so a late receipt from a previous connection cannot confirm a newer send. An `ERROR` without `receipt-id` fails all
+pending confirmations on that connection, the same policy as subscriptions. Broker errors still reach the
+`on_error_frame` callback.
+
+Confirmation is not available for `transaction.send()`: a receipt for a transactional `SEND` says nothing about whether
+the `COMMIT` succeeded.
+
 ### Listening for Messages
 
 Now, let's subscribe to a destination and listen for messages:
@@ -125,7 +156,13 @@ await client.subscribe("DLQ", handle_message_from_dlq, ack="auto", on_suppressed
 You can pass custom headers to `client.subscribe()`:
 
 ```python
-await client.subscribe("DLQ", handle_message_from_dlq, ack="client", headers={"selector": "location = 'Europe'"}, on_suppressed_exception=print)
+await client.subscribe(
+    "DLQ",
+    handle_message_from_dlq,
+    ack="client",
+    headers={"selector": "location = 'Europe'"},
+    on_suppressed_exception=print,
+)
 ```
 
 #### Handling ACK/NACKs yourself
@@ -137,6 +174,7 @@ async def handle_message_from_dlq(message_frame: stompman.AckableMessageFrame) -
     print(message_frame.body)
     await message_frame.ack()
 
+
 await client.subscribe_with_manual_ack("DLQ", handle_message_from_dlq, ack="client")
 ```
 

packages/faststream-stomp/README.md

@@ -64,6 +64,32 @@ Use `reply_add_content_length` on `subscriber()` when automatic replies need a d
 are available for batch publishing and delayed `StompRoute`/`StompRoutePublisher` registrations. Per-call settings
 override publisher or subscriber reply defaults, which in turn override the broker default.
 
+### Confirming publication
+
+By default `publish()` returns once the frame has been written to the socket, which does not mean the broker
+accepted the message. Pass `receipt_timeout` to wait for the broker's STOMP `RECEIPT` instead. It follows the same precedence as
+`add_content_length`: per-call, then publisher, then broker default.
+
+```python
+broker = faststream_stomp.StompBroker(stompman.Client([server]), receipt_timeout=3.0)
+publisher = broker.publisher("events", receipt_timeout=5.0)
+
+await broker.publish("text message", "events", receipt_timeout=1.0)
+await publisher.publish("another message")
+```
+
+The default is `None` everywhere, which preserves the existing behavior. A failed confirmation raises
+`stompman.SendReceiptError` from `publish()`, with `reason` equal to `rejected`, `timeout`, or `connection_lost`.
+
+`timeout` and `connection_lost` are ambiguous outcomes: the broker may have accepted the message even though its
+receipt was lost. Nothing is replayed automatically: give each application event a stable ID and deduplicate in the
+broker or downstream consumer if you retry. `JMSCorrelationID` alone does not deduplicate anything. And a receipt only
+means the broker took ownership: it does not mean a consumer has processed the message.
+
+`receipt_timeout` is not available for batch publishing, which publishes inside a transaction: a receipt for a
+transactional `SEND` says nothing about whether the `COMMIT` succeeded. A broker or publisher default is ignored there
+rather than silently claiming confirmation.
+
 ### Caveats
 
 - When exception is raised in consumer handler, the message will be nacked (FastStream doesn't do this by default)

packages/faststream-stomp/faststream_stomp/broker.py

@@ -102,6 +102,7 @@ def __init__(
         graceful_timeout: float | None = 15.0,
         routers: Sequence[Registrator[stompman.MessageFrame]] = (),
         add_content_length: bool = True,
+        receipt_timeout: float | None = None,
         # Logging args
         logger: LoggerProto | None = EMPTY,
         log_level: int = logging.INFO,
@@ -129,6 +130,7 @@ def __init__(
                 client=client,
                 serializer=fd_config._serializer,
                 add_content_length=add_content_length,
+                receipt_timeout=receipt_timeout,
             ),
             client=client,
         )
@@ -199,6 +201,7 @@ async def publish(
         correlation_id: str | None = None,
         headers: dict[str, str] | None = None,
         add_content_length: bool | None = None,
+        receipt_timeout: float | None = None,
     ) -> None:
         publish_command = StompPublishCommand(
             message,
@@ -207,6 +210,7 @@ async def publish(
             correlation_id=correlation_id,
             headers=headers,
             add_content_length=add_content_length,
+            receipt_timeout=receipt_timeout,
         )
         return typing.cast("None", await self._basic_publish(publish_command, producer=self.config.producer))
 
@@ -218,6 +222,7 @@ async def request(  # type: ignore[override]
         correlation_id: str | None = None,
         headers: dict[str, str] | None = None,
         add_content_length: bool | None = None,
+        receipt_timeout: float | None = None,
     ) -> Any:  # noqa: ANN401
         publish_command = StompPublishCommand(
             message,
@@ -226,6 +231,7 @@ async def request(  # type: ignore[override]
             correlation_id=correlation_id,
             headers=headers,
             add_content_length=add_content_length,
+            receipt_timeout=receipt_timeout,
         )
         return await self._basic_request(publish_command, producer=self.config.producer)
 

packages/faststream-stomp/faststream_stomp/models.py

@@ -56,6 +56,7 @@ def __init__(
         correlation_id: str | None = None,
         headers: dict[str, Any] | None = None,
         add_content_length: bool | None = None,
+        receipt_timeout: float | None = None,
     ) -> None:
         super().__init__(
             body,
@@ -67,6 +68,7 @@ def __init__(
             headers=headers,
         )
         self.add_content_length = add_content_length
+        self.receipt_timeout = receipt_timeout
 
     @classmethod
     def from_cmd(
@@ -75,10 +77,14 @@ def from_cmd(
         *,
         batch: bool = False,  # noqa: ARG003
         add_content_length: bool | None = None,
+        receipt_timeout: float | None = None,
     ) -> Self:
         messages = cmd.batch_bodies
-        if isinstance(cmd, StompPublishCommand) and cmd.add_content_length is not None:
-            add_content_length = cmd.add_content_length
+        if isinstance(cmd, StompPublishCommand):
+            if cmd.add_content_length is not None:
+                add_content_length = cmd.add_content_length
+            if cmd.receipt_timeout is not None:
+                receipt_timeout = cmd.receipt_timeout
         return cls(
             *messages,
             _publish_type=cmd.publish_type,
@@ -87,6 +93,7 @@ def from_cmd(
             correlation_id=cmd.correlation_id,
             headers=cmd.headers,
             add_content_length=add_content_length,
+            receipt_timeout=receipt_timeout,
         )
 
 
@@ -137,6 +144,7 @@ class StompPublisherSpecificationConfig(_StompBasePublisherConfig, PublisherSpec
 class StompPublisherUsecaseConfig(_StompBasePublisherConfig, PublisherUsecaseConfig):
     _outer_config: BrokerConfigWithStompClient
     add_content_length: bool | None
+    receipt_timeout: float | None
 
     @property
     def full_destination(self) -> str:

packages/faststream-stomp/faststream_stomp/publisher.py

@@ -31,10 +31,12 @@ def __init__(
         client: stompman.Client,
         serializer: SerializerProto | None,
         add_content_length: bool = True,
+        receipt_timeout: float | None = None,
     ) -> None:
         self.client = client
         self.serializer = serializer
         self.add_content_length = add_content_length
+        self.receipt_timeout = receipt_timeout
         self.codec: CodecProto = DefaultCodec()
 
     async def publish(self, cmd: StompPublishCommand) -> None:
@@ -45,6 +47,7 @@ async def publish(self, cmd: StompPublishCommand) -> None:
             content_type=content_type,
             add_content_length=self._resolve_add_content_length(cmd),
             headers=_make_headers_for_publish(cmd),
+            receipt_timeout=self._resolve_receipt_timeout(cmd),
         )
 
     async def request(self, cmd: StompPublishCommand) -> NoReturn:
@@ -66,6 +69,9 @@ async def publish_batch(self, cmd: StompPublishCommand) -> None:
     def _resolve_add_content_length(self, cmd: StompPublishCommand) -> bool:
         return self.add_content_length if cmd.add_content_length is None else cmd.add_content_length
 
+    def _resolve_receipt_timeout(self, cmd: StompPublishCommand) -> float | None:
+        return self.receipt_timeout if cmd.receipt_timeout is None else cmd.receipt_timeout
+
 
 def _make_headers_for_publish(cmd: StompPublishCommand) -> dict[str, str]:
     all_headers = cmd.headers.copy()
@@ -102,7 +108,9 @@ def __init__(self, config: StompPublisherUsecaseConfig, specification: StompPubl
     async def _publish(
         self, cmd: PublishCommand, *, _extra_middlewares: typing.Iterable[PublisherMiddleware[PublishCommand]]
     ) -> None:
-        publish_command = StompPublishCommand.from_cmd(cmd, add_content_length=self.config.add_content_length)
+        publish_command = StompPublishCommand.from_cmd(
+            cmd, add_content_length=self.config.add_content_length, receipt_timeout=self.config.receipt_timeout
+        )
         publish_command.destination = self.config.full_destination
         return typing.cast(
             "None",
@@ -118,6 +126,7 @@ async def publish(
         correlation_id: str | None = None,
         headers: dict[str, str] | None = None,
         add_content_length: bool | None = None,
+        receipt_timeout: float | None = None,
     ) -> None:
         publish_command = StompPublishCommand(
             message,
@@ -126,6 +135,7 @@ async def publish(
             correlation_id=correlation_id,
             headers=headers,
             add_content_length=self.config.add_content_length if add_content_length is None else add_content_length,
+            receipt_timeout=self.config.receipt_timeout if receipt_timeout is None else receipt_timeout,
         )
         return typing.cast(
             "None",
@@ -141,6 +151,7 @@ async def request(
         correlation_id: str | None = None,
         headers: dict[str, str] | None = None,
         add_content_length: bool | None = None,
+        receipt_timeout: float | None = None,
     ) -> Any:  # noqa: ANN401
         publish_command = StompPublishCommand(
             message,
@@ -149,6 +160,7 @@ async def request(
             correlation_id=correlation_id,
             headers=headers,
             add_content_length=self.config.add_content_length if add_content_length is None else add_content_length,
+            receipt_timeout=self.config.receipt_timeout if receipt_timeout is None else receipt_timeout,
         )
         return await self._basic_request(publish_command, producer=self.config._outer_config.producer)
 

packages/faststream-stomp/faststream_stomp/registrator.py

@@ -74,11 +74,13 @@ def publisher(  # type: ignore[override]
         description_: str | None = None,
         include_in_schema: bool = True,
         add_content_length: bool | None = None,
+        receipt_timeout: float | None = None,
     ) -> StompPublisher:
         usecase_config = StompPublisherUsecaseConfig(
             _outer_config=self.config,  # type: ignore[arg-type]
             destination_without_prefix=destination,
             add_content_length=add_content_length,
+            receipt_timeout=receipt_timeout,
         )
         specification = StompPublisherSpecification(
             _outer_config=self.config,  # type: ignore[arg-type]

packages/faststream-stomp/faststream_stomp/router.py

@@ -26,6 +26,7 @@ def __init__(
         description_: str | None = None,
         include_in_schema: bool = True,
         add_content_length: bool | None = None,
+        receipt_timeout: float | None = None,
     ) -> None:
         super().__init__(
             destination=destination,
@@ -34,6 +35,7 @@ def __init__(
             description_=description_,
             include_in_schema=include_in_schema,
             add_content_length=add_content_length,
+            receipt_timeout=receipt_timeout,
         )
 
 

packages/faststream-stomp/faststream_stomp/testing.py

@@ -66,7 +66,9 @@ async def nack(self) -> None: ...
 class FakeStompProducer(StompProducer):
     def __init__(self, broker: StompBroker) -> None:
         self.broker = broker
-        self.add_content_length = typing.cast("StompProducer", broker.config.producer).add_content_length
+        real_producer = typing.cast("StompProducer", broker.config.producer)
+        self.add_content_length = real_producer.add_content_length
+        self.receipt_timeout = real_producer.receipt_timeout
         self.codec = DefaultCodec()
 
     async def publish(self, cmd: StompPublishCommand) -> None:

packages/faststream-stomp/pyproject.toml

@@ -2,7 +2,7 @@
 name = "faststream-stomp"
 description = "FastStream STOMP broker"
 authors = [{ name = "Lev Vereshchagin", email = "mail@vrslev.com" }]
-dependencies = ["faststream~=0.7", "stompman>=3.11.1"]
+dependencies = ["faststream~=0.7", "stompman>=3.16.0"]
 requires-python = ">=3.11"
 readme = "README.md"
 license = { text = "MIT" }