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 Reply to the channel.

Basic Usage

Marking a channel with Reply adds the AsyncAPI Operation Reply object to the receive operation, where the reply channel is the channel of the message the handler sends:

Example

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")

AsyncAPI

{
  "asyncapi": "3.0.0",
  "info": {
    "title": "AsyncFast",
    "version": "0.1.0"
  },
  "channels": {
    "Ping": {
      "address": "ping",
      "messages": {
        "PingMessage": {
          "$ref": "#/components/messages/PingMessage"
        }
      }
    },
    "Pong": {
      "address": "pong",
      "messages": {
        "Pong": {
          "$ref": "#/components/messages/Pong"
        }
      }
    }
  },
  "operations": {
    "receivePing": {
      "action": "receive",
      "channel": {
        "$ref": "#/channels/Ping"
      },
      "reply": {
        "channel": {
          "$ref": "#/channels/Pong"
        },
        "messages": [
          {
            "$ref": "#/channels/Pong/messages/Pong"
          }
        ]
      }
    },
    "sendPong": {
      "action": "send",
      "channel": {
        "$ref": "#/channels/Pong"
      }
    }
  },
  "components": {
    "messages": {
      "PingMessage": {
        "payload": {
          "type": "string"
        }
      },
      "Pong": {
        "payload": {
          "type": "string"
        }
      }
    }
  }
}

The handler must send exactly one message type, otherwise an InvalidChannelDefinitionError is raised when the channel is registered, that is, when the @app.channel(...) decorator is applied.

Reply works with either sending style, so the MessageSender can be used instead of yielding:

@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 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:

Example

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")

AsyncAPI

{
  "asyncapi": "3.0.0",
  "info": {
    "title": "AsyncFast",
    "version": "0.1.0"
  },
  "channels": {
    "Ping": {
      "address": "ping",
      "messages": {
        "PingMessage": {
          "$ref": "#/components/messages/PingMessage"
        }
      }
    },
    "Pong": {
      "address": null,
      "messages": {
        "Pong": {
          "$ref": "#/components/messages/Pong"
        }
      }
    }
  },
  "operations": {
    "receivePing": {
      "action": "receive",
      "channel": {
        "$ref": "#/channels/Ping"
      },
      "reply": {
        "address": {
          "location": "$message.header#/replyTo"
        },
        "channel": {
          "$ref": "#/channels/Pong"
        },
        "messages": [
          {
            "$ref": "#/channels/Pong/messages/Pong"
          }
        ]
      }
    },
    "sendPong": {
      "action": "send",
      "channel": {
        "$ref": "#/channels/Pong"
      }
    }
  },
  "components": {
    "messages": {
      "PingMessage": {
        "headers": {
          "$ref": "#/components/schemas/PingMessageHeaders"
        },
        "payload": {
          "type": "string"
        }
      },
      "Pong": {
        "payload": {
          "type": "string"
        }
      }
    },
    "schemas": {
      "PingMessageHeaders": {
        "properties": {
          "replyTo": {
            "title": "Replyto",
            "type": "string"
          }
        },
        "required": [
          "replyTo"
        ],
        "title": "PingMessageHeaders",
        "type": "object"
      }
    }
  }
}

ReplyTo behaves like 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 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 ReplyTo, pass a ReplyAddress to 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:

@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 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 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:

@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.