feat(asyncfast): add request/reply asyncapi support

packages/asyncfast/docs/examples/request_reply.py

@@ -0,0 +1,18 @@
+from collections.abc import AsyncGenerator
+from dataclasses import dataclass
+
+from asyncfast import AsyncFast
+from asyncfast import Message
+from asyncfast import Reply
+
+app = AsyncFast()
+
+
+@dataclass
+class Pong(Message, address="pong"):
+    payload: str
+
+
+@app.channel("ping", reply=Reply())
+async def ping(payload: str) -> AsyncGenerator[Pong, None]:
+    yield Pong(payload="pong")

packages/asyncfast/docs/examples/request_reply_dynamic_address.py

@@ -0,0 +1,24 @@
+from collections.abc import AsyncGenerator
+from dataclasses import dataclass
+from typing import Annotated
+
+from asyncfast import AsyncFast
+from asyncfast import Message
+from asyncfast import Reply
+from asyncfast import ReplyTo
+
+app = AsyncFast()
+
+
+@dataclass
+class Pong(Message, address="{reply_to}"):
+    reply_to: str
+    payload: str
+
+
+@app.channel("ping", reply=Reply())
+async def ping(
+    payload: str,
+    reply_to: Annotated[str, ReplyTo(alias="replyTo")],
+) -> AsyncGenerator[Pong, None]:
+    yield Pong(reply_to=reply_to, payload="pong")

packages/asyncfast/docs/index.rst

@@ -83,6 +83,7 @@ Taking ideas from:
 
    receiving
    sending
+   request_reply
    message_send_router
    dependencies
    lifespan

packages/asyncfast/docs/request_reply.rst

@@ -0,0 +1,86 @@
+###############
+ Request/Reply
+###############
+
+Request/reply is where a requester sends a message and waits for a response, with the response sent back to the
+requester. In AsyncFast the reply of the interaction is the message the channel handler sends, so the pattern is
+declared by passing :py:class:`Reply` to the channel.
+
+*************
+ Basic Usage
+*************
+
+Marking a channel with :py:class:`Reply` adds the AsyncAPI Operation Reply object to the receive operation, where the
+reply channel is the channel of the message the handler sends:
+
+.. async-fast-example:: examples/request_reply.py
+
+The handler must send exactly one message type, otherwise an
+:py:class:`InvalidChannelDefinitionError` is raised when the channel is registered, that is, when the ``@app.channel(...)`` decorator is applied.
+
+:py:class:`Reply` works with either sending style, so the :py:class:`MessageSender` can be used instead of yielding:
+
+.. code-block:: python
+
+   @app.channel("ping", reply=Reply())
+   async def ping(message_sender: MessageSender[Pong]) -> None:
+       await message_sender.send(Pong(payload="pong"))
+
+***********************
+ Dynamic Reply Address
+***********************
+
+When the address of the reply isn't known at design time, it is determined at runtime from the request itself. This
+is typically done by the requester including a ``replyTo`` header. Marking that header with :py:class:`ReplyTo`
+tells AsyncFast it carries the reply address, so the AsyncAPI reply address is derived from it and the header only
+needs to be written once:
+
+.. async-fast-example:: examples/request_reply_dynamic_address.py
+
+:py:class:`ReplyTo` behaves like :py:class:`Header`, so the value is injected into the handler, and it can equally be
+used in a dependency of the channel. A channel can have at most one :py:class:`ReplyTo` header.
+
+The derived location is an AsyncAPI runtime expression, ``$message.header#/replyTo`` in the example above. To describe
+the address, or to point at a location that isn't a header marked with :py:class:`ReplyTo`, pass a
+:py:class:`ReplyAddress` to :py:class:`Reply`. The location must be either ``$message.header`` or
+``$message.payload`` followed by a JSON Pointer. As a shortcut, a location can also be passed as a string:
+
+.. code-block:: python
+
+   @app.channel("ping", reply=Reply(address="$message.payload#/replyTo"))
+   async def ping(payload: PingPayload) -> AsyncGenerator[Pong, None]:
+       yield Pong(payload="pong")
+
+The reply address is documentation only, so the handler is still responsible for sending to the address the requester
+supplied. Binding the :py:class:`ReplyTo` header to a parameter of the reply channel address, as in the example above,
+is the most direct way of doing this.
+
+.. note::
+
+   AsyncAPI 3.0 requires that, when a reply address is specified, the address of the channel referenced by the reply
+   is ``null``. When :py:class:`Reply` is given an address, the reply channel is therefore rendered with a ``null``
+   address, with the reply address runtime expression carrying the routing information. The reply message should not
+   be reused as the send message of a standalone channel, since its channel address is rendered as ``null``.
+
+********************
+ Same Reply Address
+********************
+
+The request and the reply can also share an address, which is a reply channel using the same address as the request
+channel:
+
+.. code-block:: python
+
+   @dataclass
+   class Pong(Message, address="ping"):
+       payload: str
+
+
+   @app.channel("ping", reply=Reply())
+   async def ping(payload: str) -> AsyncGenerator[Pong, None]:
+       yield Pong(payload="pong")
+
+.. note::
+
+   AsyncFast generates a channel per message sent, so the reply is a new channel which happens to reuse the request
+   address, rather than the request channel itself.

packages/asyncfast/src/asyncfast/__init__.py

@@ -5,6 +5,9 @@
 from asyncfast._channel import MessageSender
 from asyncfast._channel import Parameter
 from asyncfast._channel import Payload
+from asyncfast._channel import Reply
+from asyncfast._channel import ReplyAddress
+from asyncfast._channel import ReplyTo
 from asyncfast._message import Message
 from asyncfast._utils import ChannelNotFoundError
 from asyncfast._utils import InvalidChannelDefinitionError
@@ -18,6 +21,9 @@
     "MessageSender",
     "Parameter",
     "Payload",
+    "Reply",
+    "ReplyAddress",
+    "ReplyTo",
     "Message",
     "ChannelNotFoundError",
 ]

packages/asyncfast/src/asyncfast/_asyncapi.py

@@ -20,8 +20,11 @@
 from asyncfast._channel import MessageSenderResolver
 from asyncfast._channel import Payload
 from asyncfast._channel import PayloadResolver
+from asyncfast._channel import Reply
+from asyncfast._channel import ReplyAddress
 from asyncfast._channel import Resolver
 from asyncfast._message import Message
+from asyncfast._utils import InvalidChannelDefinitionError
 from pydantic import BaseModel
 from pydantic import create_model
 from pydantic import TypeAdapter
@@ -49,14 +52,15 @@ class MessageDefinition:
     headers: Sequence[tuple[str, type[Any], Any]]
     bindings: Sequence[tuple[str, str, type[Any], CoreSchema]]
     payload: tuple[type[Any], TypeAdapter[Any], Payload] | None
+    dynamic_address: bool = False
 
     @property
     def channel_definition(self) -> dict[str, Any]:
         definition = {
-            "address": self.address,
+            "address": None if self.dynamic_address else self.address,
             "messages": {self.name: {"$ref": f"#/components/messages/{self.name}"}},
         }
-        if self.parameters:
+        if self.parameters and not self.dynamic_address:
             definition["parameters"] = {name: {} for name in self.parameters}
         return definition
 
@@ -175,6 +179,64 @@ def payload(self) -> PayloadResolver[Any] | None:
             return payloads[0]
         return None
 
+    @cached_property
+    def reply(self) -> Reply | None:
+        return self.channel.reply
+
+    @cached_property
+    def reply_address(self) -> ReplyAddress | None:
+        if self.reply is None:
+            return None
+        if isinstance(self.reply.address, ReplyAddress):
+            return self.reply.address
+
+        reply_to_names = tuple(
+            dict.fromkeys(
+                resolver.name
+                for resolver in self.resolvers
+                if isinstance(resolver, HeaderResolver) and resolver.reply_to
+            )
+        )
+        if len(reply_to_names) > 1:
+            raise InvalidChannelDefinitionError(
+                "Channel must have no more than 1 reply to header"
+            )
+        if not reply_to_names:
+            return None
+        pointer = reply_to_names[0].replace("~", "~0").replace("/", "~1")
+        return ReplyAddress(f"$message.header#/{pointer}")
+
+    @cached_property
+    def reply_definition(self) -> dict[str, Any] | None:
+        if self.reply is None:
+            return None
+
+        messages = tuple(dict.fromkeys(self.send_messages))
+        if len(messages) != 1:
+            raise InvalidChannelDefinitionError(
+                "Channel with reply must send exactly one message type"
+            )
+        (message,) = messages
+
+        definition: dict[str, Any] = {}
+
+        reply_address = self.reply_address
+        if isinstance(reply_address, ReplyAddress):
+            definition["address"] = {
+                "location": reply_address.location,
+                **(
+                    {"description": reply_address.description}
+                    if reply_address.description
+                    else {}
+                ),
+            }
+
+        definition["channel"] = {"$ref": f"#/channels/{message.__name__}"}
+        definition["messages"] = [
+            {"$ref": f"#/channels/{message.__name__}/messages/{message.__name__}"}
+        ]
+        return definition
+
     def generate_send_messages(self) -> Generator[type[Message], None, None]:
         signature = inspect.signature(self.channel.func)
 
@@ -207,6 +269,7 @@ def send_messages(self) -> Sequence[type[Message]]:
 
     @cached_property
     def send_message_definitions(self) -> Sequence[MessageDefinition]:
+        dynamic_address = self.reply_address is not None
         return tuple(
             MessageDefinition(
                 message.__name__,
@@ -231,6 +294,7 @@ def send_message_definitions(self) -> Sequence[MessageDefinition]:
                     if message.__payload__
                     else None
                 ),
+                dynamic_address,
             )
             for message in self.send_messages
         )
@@ -323,11 +387,17 @@ def generate_operations(
     channel_definitions: Iterable[ChannelDefinition],
 ) -> Generator[tuple[str, dict[str, Any]], None, None]:
     for channel_definition in channel_definitions:
-        yield f"receive{channel_definition.title}", {
+        receive_operation: dict[str, Any] = {
             "action": "receive",
             "channel": {"$ref": f"#/channels/{channel_definition.title}"},
         }
 
+        reply_definition = channel_definition.reply_definition
+        if reply_definition is not None:
+            receive_operation["reply"] = reply_definition
+
+        yield f"receive{channel_definition.title}", receive_operation
+
         for message in channel_definition.send_messages:
             yield f"send{message.__name__}", {
                 "action": "send",

packages/asyncfast/src/asyncfast/_asyncfast.py

@@ -14,8 +14,12 @@
 from amgi_types import LifespanShutdownCompleteEvent
 from amgi_types import LifespanStartupCompleteEvent
 from amgi_types import Scope
+from asyncfast._asyncapi import ChannelDefinition
 from asyncfast._asyncapi import get_asyncapi
 from asyncfast._channel import ChannelRouter
+from asyncfast._channel import get_channel
+from asyncfast._channel import Reply
+from asyncfast._utils import InvalidChannelDefinitionError
 from asyncfast.middleware.errors import ServerErrorMiddleware
 
 P = ParamSpec("P")
@@ -67,13 +71,26 @@ def title(self) -> str:
     def version(self) -> str:
         return self._version
 
-    def channel(self, address: str) -> Callable[[DecoratedCallable], DecoratedCallable]:
-        return partial(self._add_channel, address)
+    def channel(
+        self,
+        address: str,
+        *,
+        reply: Reply | None = None,
+    ) -> Callable[[DecoratedCallable], DecoratedCallable]:
+        return partial(self._add_channel, address, reply)
 
     def _add_channel(
-        self, address: str, function: DecoratedCallable
+        self, address: str, reply: Reply | None, function: DecoratedCallable
     ) -> DecoratedCallable:
-        self._router.add_channel(address, function)
+        channel = get_channel(function, address, reply)
+        if (
+            reply is not None
+            and len(set(ChannelDefinition(channel).send_messages)) != 1
+        ):
+            raise InvalidChannelDefinitionError(
+                "Channel with reply must send exactly one message type"
+            )
+        self._router.add_route(address, channel)
         return function
 
     async def __call__(