AsyncAPI Code Generation with Corvus: Request/Reply

By Matthew Adams Co-Founder 27th July 2026 · 4 min read

At endjin, we maintain Corvus.JsonSchema, and in the previous post we generated consumers with error policies and resilience middleware.

So far in this series, we've looked at fire-and-forget messaging: a producer publishes, a consumer eventually processes. But there's another pattern that messaging systems support, and it's one that often trips people up when they first encounter it: request/reply.

When fire-and-forget isn't enough

The produce/consume pattern works beautifully for event-driven architectures where the publisher doesn't care (or need to know) what happens downstream. An order is placed, an event is published, and various consumers react independently. The publisher carries on with its life.

But some operations are inherently conversational. You send a query and need a result. You submit a calculation and need the answer. You request a validation and need a yes/no. These are request/reply interactions. While you could model them as two independent channels with manual correlation, that's tedious to implement correctly and easy to get subtly wrong.

The tricky parts are: generating a unique correlation ID for each request, subscribing to the reply channel filtered by that ID before sending the request (to avoid race conditions), handling timeouts when the responder is slow or dead, and cleaning up the reply subscription once you have the answer. Every one of these is a bug waiting to happen if you write it by hand.

What the generator produces

When your AsyncAPI spec defines a request/reply operation, the generator produces a method that handles the entire correlation lifecycle:

(QueryResponse reply, JsonElement replyHeaders) = await queryProducer.RequestQueryAsync(
    request: new QueryPayload.Source((ref QueryPayload.Builder b) =>
    {
        b.Create(filter: "status=active"u8);
    }),
    cancellationToken: ct);

// reply is already deserialized and validated
foreach (var item in reply.Results.EnumerateArray())
{
    Console.WriteLine($"Found: {item.Name}");
}

What's happening under the hood is worth understanding. The generated method:

1. Creates a correlation ID - a GUID formatted directly into a pooled byte[36] buffer, avoiding any string allocation
2. Subscribes to the reply channel, filtered by that correlation ID
3. Publishes the request with the correlation ID attached as a header
4. Awaits the reply (with cancellation token support for timeouts)
5. Deserializes and validates the response payload against its schema
6. Returns the typed response and any reply headers
7. Cleans up the reply subscription

The caller sees a single await that returns a typed result. The generated code owns the complexity of the correlation dance.

Defining request/reply in your spec

AsyncAPI 3.0

In AsyncAPI 3.0, request/reply is a first-class concept. You define an operation with a reply object that specifies the reply channel, the reply message, and where in the message the correlation ID lives:

operations:
  requestQuery:
    action: send
    channel:
      $ref: '#/channels/queries'
    messages:
      - $ref: '#/channels/queries/messages/queryRequest'
    reply:
      channel:
        $ref: '#/channels/queryReplies'
      messages:
        - $ref: '#/channels/queryReplies/messages/queryResponse'
      address:
        location: $message.header#/correlationId

The address.location tells the generator (and runtime) where in the reply message to find the correlation ID. The generator uses this to filter incoming replies.

AsyncAPI 2.6

AsyncAPI 2.6 has correlationId on messages but no standard way to declare the reply channel or message. The generator could try to infer request/reply pairs from matching correlation IDs, but in real-world 2.6 documents that's ambiguous. Multiple operations might share a correlation ID schema without being request/reply pairs.

Instead, Corvus supports an explicit x-corvus-reply extension that mirrors the 3.0 shape:

{
  "subscribe": {
    "operationId": "calculate",
    "message": {
      "$ref": "#/components/messages/CalculateRequest"
    },
    "x-corvus-reply": {
      "channel": {
        "$ref": "#/channels/rpc~1calculate~1replies"
      },
      "address": {
        "location": "$message.header#/replyTo"
      },
      "message": {
        "$ref": "#/components/messages/CalculateResponse"
      }
    }
  }
}

The generated request/reply method is identical regardless of which spec version you use. The extension simply gives the generator the information it needs to produce that method.

The responder side

A request/reply pattern has two participants: the requester (who sends and awaits) and the responder (who receives, processes, and replies). The generator produces code for both.

On the responder side, the generated consumer calls your handler with both the request payload and a reply callback:

internal sealed class QueryHandler : IQueryHandler
{
    public async ValueTask HandleQueryAsync(
        QueryPayload request,
        Func<QueryResponse.Source, ValueTask> reply,
        CancellationToken cancellationToken)
    {
        // Process the query
        var results = await SearchAsync(request.Filter, cancellationToken);

        // Send the reply - the generated code handles correlation
        await reply(new QueryResponse.Source((ref QueryResponse.Builder b) =>
        {
            b.Create(results: results, count: results.Length);
        }));
    }
}

Your handler receives a typed request, does whatever processing is appropriate, and calls the reply callback with a typed response. The generated consumer handles extracting the correlation ID from the inbound request, attaching it to the outbound reply, and routing it to the correct reply channel. You don't touch correlation at all.

Performance characteristics

The request/reply path follows the same zero-allocation philosophy as the rest of V5. The correlation ID is a GUID formatted directly into a rented byte[36] buffer. There's no Guid.ToString() and no string allocation. The reply channel address is computed once and hoisted to a static field. The reply payload is parsed into pooled memory.

In benchmarks against Wolverine (a .NET messaging framework with its own request/reply support), the numbers look like this:

Without validation, Corvus is 28% faster and allocates 65% less. With basic validation enabled (checking schemas on both the request and reply payloads), the time cost is modest while the allocation advantage remains. You get full schema conformance checking in both directions for roughly 1.25× the baseline time, but with the same 65% allocation reduction.

When to use request/reply vs. produce/consume

The two patterns serve different architectural needs, and choosing between them is a design decision worth being intentional about.

Produce/consume (fire-and-forget) is right when:

• The publisher genuinely doesn't need a response
• Multiple independent consumers may react to the same event
• You want temporal decoupling (the consumer can process hours later)
• You're building event-sourced or CQRS architectures

Request/reply is right when:

• The caller needs a result before it can proceed
• The interaction is inherently conversational (query/response, validate/result)
• You want the decoupling benefits of messaging (independent deployment, transport abstraction) but the calling semantics of an RPC

The important thing is that both patterns are generated from the same AsyncAPI spec, use the same typed models, and run over the same transports. You don't need a separate framework for each. A service can produce events on some channels and make request/reply calls on others, all from the same generated code.

For the full request/reply reference, including transport-specific considerations and timeout configuration, see the AsyncAPI documentation on corvus-oss.org.

In the [ref slug=asyncapi-code-generation-with-corvus-health-and-telemetry text=next post], we'll look at observability - distributed tracing, metrics, and health checks for your messaging infrastructure.

FAQs