AsyncAPI Code Generation with Corvus: Typed Consumers

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

At endjin, we maintain Corvus.JsonSchema, and in the previous post we generated typed producers that validate messages before they leave your process.

Now let's look at the other side: consuming messages reliably.

The consumer's dilemma

Publishing is the easy half of messaging. You control the data, you control when you send, and if something goes wrong you get an immediate exception. Consuming is harder. Messages arrive from the outside world, on someone else's schedule, and some of them will be malformed, unexpected, or arrive when your downstream dependencies are having a bad day.

A robust consumer needs to answer several questions: Is this message well-formed? Does it match the expected schema? What do I do if my handler throws? What happens if the broker connection drops? And critically - how do I avoid silently losing messages?

Most messaging frameworks leave you to wire this up yourself. You write deserialization code, catch exceptions, decide what to log, and hope you remembered to handle every failure mode. The generated consumer in Corvus takes a different approach: it handles the entire subscription lifecycle and makes the failure-handling strategy explicit through an IMessageErrorPolicy.

Generating a consumer

corvusjson asyncapi-generate streetlights.json \
    --rootNamespace Streetlights.Client \
    --outputPath ./Generated \
    --mode consumer

The generator reads the spec's receive operations (or publish in AsyncAPI 2.6) and produces a consumer class, a handler interface, and the same model types used by the producer.

Your handler only sees valid data

The generated handler interface has a single async method. Your implementation receives a strongly-typed payload that has already been deserialized and validated against the JSON Schema from the spec:

internal sealed class LightMeasuredHandler : IReceiveLightMeasurementHandler
{
    public ValueTask HandleLightMeasuredAsync(
        LightMeasuredPayload payload,
        CancellationToken cancellationToken = default)
    {
        // Payload is already validated - lumens >= 0 is guaranteed by the schema
        int lumens = (int)payload.Lumens;
        Console.WriteLine($"Light measured: {lumens} lumens at {payload.SentAt}");
        return default;
    }
}

This is the same philosophy as the OpenAPI server handlers: the generated infrastructure owns deserialization and validation, and your handler only runs when the data is known to be good. If a malformed message arrives, your handler is never called. The error policy handles it.

Wiring up the consumer

await using InMemoryMessageTransport transport = new();

LightMeasuredHandler handler = new();
ReceiveLightMeasurementConsumer consumer = new(
    transport,
    handler,
    validationMode: ValidationMode.Basic);

await consumer.StartAsync();
// Messages arriving on the transport are now dispatched to the handler

await consumer.StopAsync();

StartAsync subscribes to the channel defined in the spec. From that point, every message that arrives on the transport is deserialized, validated, and dispatched to your handler. If something goes wrong, it is routed through the error policy.

What happens when things go wrong

This is where the design gets interesting. In most messaging code, error handling is ad-hoc: a try/catch around the handler, maybe a log statement, maybe a retry. The failure strategy is buried in the implementation and varies from consumer to consumer.

Corvus makes the strategy explicit with IMessageErrorPolicy. When processing fails - whether from a deserialization error, a handler exception, or a transport issue - the policy decides the terminal action:

The DefaultMessageErrorPolicy provides a sensible starting point: dead-letter deserialization failures and handler exceptions (so you can inspect them later), and abort on transport connectivity errors (because retrying won't help if the broker is down):

IMessageErrorPolicy policy = new DefaultMessageErrorPolicy(
    deserializationAction: MessageErrorAction.Skip,
    handlerAction: MessageErrorAction.DeadLetter,
    transportAction: MessageErrorAction.Abort);

ReceiveLightMeasurementConsumer consumer = new(
    transport, handler,
    validationMode: ValidationMode.Basic,
    errorPolicy: policy);

Custom error policies

For more complex scenarios - say you want to retry a few times before dead-lettering - implement IMessageErrorPolicy directly:

internal sealed class RetryThenDeadLetterPolicy : IMessageErrorPolicy
{
    private readonly int maxAttempts;
    private int attempts;

    public RetryThenDeadLetterPolicy(int maxAttempts = 3)
    {
        this.maxAttempts = maxAttempts;
    }

    public ValueTask<MessageErrorAction> HandleErrorAsync(
        Exception exception,
        MessageErrorContext context,
        CancellationToken cancellationToken = default)
    {
        if (context.ErrorKind == MessageErrorKind.Transport)
        {
            return new(MessageErrorAction.Abort);
        }

        this.attempts++;
        MessageErrorAction action = this.attempts >= this.maxAttempts
            ? MessageErrorAction.DeadLetter
            : MessageErrorAction.Skip;

        return new(action);
    }
}

The MessageErrorContext tells you what kind of error occurred, so you can make different decisions for different failure modes. Transport errors are typically fatal (abort), while handler exceptions might be transient (retry then dead-letter).

Dead-letter routing

When the error policy returns DeadLetter, the generated consumer publishes the failed message to a derived channel address - for example, dead-letter.smartylighting.streetlights.1.0.action.{id}.lighting.measured. The dead-letter message carries the original payload bytes, headers, the exception that caused the failure, and the original channel address. You have everything you need to investigate and replay.

Each transport maps this to whatever mechanism the broker provides natively. Kafka publishes to a dead-letter topic. Azure Service Bus uses its built-in dead-letter settlement path. NATS publishes to a dead-letter subject. The abstraction is consistent, but the underlying behaviour is idiomatic for each broker.

Resilience with Polly

For transient failures where you want automatic retry with backpressure, the Corvus.Text.Json.AsyncApi.Polly package wraps handler invocations with Polly resilience pipelines:

dotnet add package Corvus.Text.Json.AsyncApi.Polly

using Corvus.Text.Json.AsyncApi.Polly;
using Polly;

ResiliencePipeline pipeline = new ResiliencePipelineBuilder()
    .AddRetry(new()
    {
        MaxRetryAttempts = 3,
        BackoffType = DelayBackoffType.Exponential,
        Delay = TimeSpan.FromMilliseconds(200),
    })
    .AddCircuitBreaker(new()
    {
        FailureRatio = 0.5,
        SamplingDuration = TimeSpan.FromSeconds(10),
        MinimumThroughput = 5,
        BreakDuration = TimeSpan.FromSeconds(30),
    })
    .Build();

NatsTransportOptions options = new()
{
    Url = "nats://localhost:4222",
    HandlerMiddleware = PollyResilienceMiddleware.Create(pipeline),
};

await using NatsMessageTransport transport = await NatsMessageTransport.CreateAsync(options);

The middleware wraps every handler invocation. If all retries are exhausted and the circuit breaker opens, the exception propagates to the IMessageErrorPolicy for a terminal decision. This gives you two layers of resilience: Polly handles transient retries, and the error policy handles the final disposition when retries are spent.

The relationship between producer and consumer

One thing worth making explicit: the producer and consumer are generated from the same spec. The producer validates outgoing payloads against the schema, and the consumer validates incoming payloads against the same schema. If both sides are regenerated from the same spec version, they agree on the contract by construction.

This means you can evolve your schema with confidence. Add a new optional field to the payload, regenerate both sides, and the producer can start sending it while existing consumers ignore it (because it's optional). Make a field required, and any consumer that hasn't regenerated will start dead-lettering messages that contain it. That is exactly what you want, because it surfaces the incompatibility immediately rather than silently dropping data.

In the [ref slug=asyncapi-code-generation-with-corvus-request-reply text=next post], we'll look at a different messaging pattern entirely - request/reply, where you send a message and await a correlated response.

FAQs