feat: Implement request-response pattern for mqtt 5 (#19)

Author: borisalekseev

docs/advanced/mqtt5.md

@@ -88,6 +88,20 @@ async with client.subscribe(
 !!! warning
     `no_local` and `retain_as_published` raise `RuntimeError` when used on a `version="3.1.1"` connection.
 
+## Request / response (`client.request()`)
+
+Send a request and await exactly one reply in a single call:
+
+```python
+reply = await client.request("services/echo", b"hello", timeout=5.0)
+print(reply.payload)  # b"hello"
+```
+
+zmqtt manages the reply topic subscription, the `response_topic` /
+`correlation_data` PUBLISH properties, and cleanup on timeout or
+cancellation automatically. See [Request / Response](request-response.md)
+for the full API and responder example.
+
 ## Enhanced authentication (`client.auth()`)
 
 Send an AUTH packet for extended authentication flows (e.g. SCRAM, Kerberos):

docs/advanced/request-response.md

@@ -0,0 +1,101 @@
+# Request / Response
+
+MQTT 5.0 defines a first-class request/response pattern via two
+`PUBLISH` properties: `response_topic` and `correlation_data`. zmqtt
+implements this as a single `await client.request(…)` call.
+
+## Basic usage
+
+```python
+from zmqtt import create_client
+
+async with create_client("broker", version="5.0") as client:
+    reply = await client.request("services/calculator", b"2+2")
+    print(reply.payload)
+```
+
+`request()` handles the full flow automatically:
+
+1. Subscribes to a unique reply topic before publishing.
+2. Publishes the request with `response_topic` and `correlation_data` set.
+3. Waits for the first message on the reply topic and returns it.
+4. Unsubscribes on return, timeout, or cancellation.
+
+## Customising via `PublishProperties`
+
+Pass a `PublishProperties` instance to control any field of the outgoing
+PUBLISH. Two fields receive special treatment:
+
+| Field              | Behaviour                                                                              |
+| ------------------ | -------------------------------------------------------------------------------------- |
+| `response_topic`   | Used as the reply topic instead of the auto-generated one. Must not contain wildcards. |
+| `correlation_data` | Forwarded to the responder as-is. Auto-generated (16 random bytes) when absent.        |
+
+All other fields (`content_type`, `message_expiry_interval`,
+`user_properties`, …) are forwarded unchanged.
+
+```python
+from zmqtt import PublishProperties
+
+reply = await client.request(
+    "services/translate",
+    b"hello",
+    properties=PublishProperties(
+        content_type="text/plain",
+        response_topic="my-app/replies/translate",
+        correlation_data=b"req-001",
+    ),
+    timeout=10.0,
+)
+```
+
+## Implementing a responder
+
+The responder reads `response_topic` and `correlation_data` from the
+incoming message and publishes the reply there:
+
+```python
+async with client.subscribe("services/translate") as sub:
+    async for msg in sub:
+        assert msg.properties is not None
+        result = translate(msg.payload)
+        await client.publish(
+            msg.properties.response_topic,
+            result,
+            properties=PublishProperties(
+                correlation_data=msg.properties.correlation_data,
+            ),
+        )
+```
+
+## Timeout
+
+`request()` raises `asyncio.TimeoutError` when no reply arrives within
+`timeout` seconds (default `30.0`). The reply subscription is always
+cleaned up, even on timeout or cancellation.
+
+```python
+import asyncio
+
+try:
+    reply = await client.request("slow/service", b"ping", timeout=5.0)
+except asyncio.TimeoutError:
+    print("Service did not respond in time")
+```
+
+## Errors
+
+| Exception               | Raised when                                       |
+| ----------------------- | ------------------------------------------------- |
+| `RuntimeError`          | `request()` is called on an MQTT 3.1.1 connection |
+| `MQTTInvalidTopicError` | `properties.response_topic` contains wildcards    |
+| `MQTTDisconnectedError` | Connection is lost while waiting for the reply    |
+| `asyncio.TimeoutError`  | No reply arrives within `timeout` seconds         |
+
+!!! note
+`request()` is only available on MQTT 5.0 connections. Use
+`create_client(…, version="5.0")` or `MQTTClient(…, version="5.0")`.
+
+---
+
+**See also:** [MQTT 5.0](mqtt5.md) · [Publishing](../publishing.md) · [Error Handling](../error-handling.md)

docs/api-reference.md

@@ -96,3 +96,7 @@
 ::: zmqtt.MQTTTimeoutError
     options:
       show_source: false
+
+::: zmqtt.MQTTInvalidTopicError
+    options:
+      show_source: false

docs/error-handling.md

@@ -7,7 +7,8 @@ MQTTError
   ├── MQTTConnectError      # CONNACK refused (return_code attribute)
   ├── MQTTProtocolError     # malformed or unexpected packet
   ├── MQTTDisconnectedError # connection lost unexpectedly
-  └── MQTTTimeoutError      # ping or operation timed out
+  ├── MQTTTimeoutError      # ping or operation timed out
+  └── MQTTInvalidTopicError # topic string failed MQTT validation
 ```
 
 All exceptions are importable from `zmqtt`:
@@ -19,6 +20,7 @@ from zmqtt import (
     MQTTProtocolError,
     MQTTDisconnectedError,
     MQTTTimeoutError,
+    MQTTInvalidTopicError,
 )
 ```
 
@@ -71,6 +73,62 @@ except MQTTTimeoutError:
 
 See [Manual Ping](advanced/ping.md) for the full `ping()` API.
 
+### `MQTTInvalidTopicError`
+
+Raised when a topic string fails MQTT validation. The check happens eagerly —
+before any I/O — in `publish()`, `subscribe()`, and `request()`.
+
+**`publish()` — topic name rules:**
+
+- Must not be empty.
+- Must not contain `+` or `#` (wildcards are for filters only).
+- `$` is only valid as the very first character.
+
+```python
+from zmqtt import MQTTInvalidTopicError
+
+try:
+    await client.publish("sensors/+/temp", b"22.5")
+except MQTTInvalidTopicError as e:
+    print(e)  # Wildcards not allowed in publish topic: 'sensors/+/temp'
+```
+
+**`subscribe()` — topic filter rules:**
+
+- Must not be empty.
+- `#` must be the last character and, if not the only character, must be
+  preceded by `/`.
+- `+` must occupy an entire level (e.g. `a/+/b` is valid; `a/temp+/b` is not).
+- `$` is only valid as the very first character.
+
+```python
+try:
+    client.subscribe("sensors#")          # missing preceding '/'
+    client.subscribe("sensors/temp+/data") # '+' not a full level
+except MQTTInvalidTopicError as e:
+    print(e)
+```
+
+**`request()` — response topic rules:**
+
+The `response_topic` property follows the same rules as a publish topic (no
+wildcards):
+
+```python
+from zmqtt import MQTTInvalidTopicError, PublishProperties
+
+try:
+    await client.request(
+        "cmd",
+        b"x",
+        properties=PublishProperties(response_topic="reply/+/bad"),
+    )
+except MQTTInvalidTopicError as e:
+    print(e)
+```
+
+See [Request / Response](advanced/request-response.md) for details.
+
 ## Reconnection interaction
 
 When `ReconnectConfig(enabled=True)` (the default), `MQTTDisconnectedError` and `MQTTTimeoutError` inside the protocol loop are suppressed and the client reconnects with exponential backoff. Your `async for msg in sub` loop keeps waiting — it will resume delivering messages once the connection is restored.

mkdocs.yml

@@ -33,6 +33,7 @@ nav:
     - Reconnection: advanced/reconnection.md
     - Ping: advanced/ping.md
     - MQTT 5.0: advanced/mqtt5.md
+    - Request / Response: advanced/request-response.md
     - Backpressure: advanced/backpressure.md
     - Scaling: advanced/scaling.md
   - Error Handling: error-handling.md

src/zmqtt/__init__.py

@@ -14,6 +14,7 @@
     MQTTConnectError,
     MQTTDisconnectedError,
     MQTTError,
+    MQTTInvalidTopicError,
     MQTTProtocolError,
     MQTTTimeoutError,
 )
@@ -27,6 +28,7 @@
     "MQTTConnectError",
     "MQTTDisconnectedError",
     "MQTTError",
+    "MQTTInvalidTopicError",
     "MQTTProtocolError",
     "MQTTTimeoutError",
     "Message",

src/zmqtt/_internal/types/topic.py

@@ -1,64 +1,38 @@
-from typing_extensions import Self
+from zmqtt.errors import MQTTInvalidTopicError
 
 
-def _validate_topic_name(topic: str) -> None:
+def validate_publish(topic: str) -> None:
     if not topic:
         msg = "Topic must not be empty"
-        raise ValueError(msg)
+        raise MQTTInvalidTopicError(msg)
     if "#" in topic or "+" in topic:
         msg = f"Wildcards not allowed in publish topic: {topic!r}"
-        raise ValueError(msg)
+        raise MQTTInvalidTopicError(msg)
     if "$" in topic[1:]:
         msg = f"'$' is only valid as the first character of a topic: {topic!r}"
-        raise ValueError(
-            msg,
-        )
+        raise MQTTInvalidTopicError(msg)
 
 
-def _validate_topic_filter(topic: str) -> None:
+def validate_subscribe_topic(topic: str) -> None:
     if not topic:
         msg = "Topic filter must not be empty"
-        raise ValueError(msg)
+        raise MQTTInvalidTopicError(msg)
     if "$" in topic[1:]:
         msg = f"'$' is only valid as the first character of a topic filter: {topic!r}"
-        raise ValueError(
-            msg,
-        )
+        raise MQTTInvalidTopicError(msg)
     if "#" in topic:
         idx = topic.index("#")
         if idx != len(topic) - 1:
             msg = f"'#' must be the last character in a topic filter: {topic!r}"
-            raise ValueError(
-                msg,
-            )
+            raise MQTTInvalidTopicError(msg)
         if idx > 0 and topic[idx - 1] != "/":
             msg = f"'#' must be preceded by '/' in a topic filter: {topic!r}"
-            raise ValueError(
-                msg,
-            )
+            raise MQTTInvalidTopicError(msg)
     for level in topic.split("/"):
         if "+" in level and level != "+":
             msg = f"'+' must occupy an entire topic level in filter: {topic!r}"
-            raise ValueError(
-                msg,
-            )
+            raise MQTTInvalidTopicError(msg)
 
 
-class Topic(str):
-    """Validated publish topic — no wildcards, '$' only as first character."""
-
-    __slots__ = ()
-
-    def __new__(cls, value: str) -> Self:
-        _validate_topic_name(value)
-        return super().__new__(cls, value)
-
-
-class TopicFilter(str):
-    """Validated subscription filter. Wildcards allowed, '$' only as first character."""
-
-    __slots__ = ()
-
-    def __new__(cls, value: str) -> Self:
-        _validate_topic_filter(value)
-        return super().__new__(cls, value)
+def validate_response_topic(topic: str) -> None:
+    validate_publish(topic)