AsyncAPI Code Generation

August 19, 2026 · View on GitHub

Try the AsyncAPI Playground — generate strongly-typed producers, consumers, handlers, and request/response (request/reply) code in your browser.

Overview

Corvus.Text.Json includes a code generator that produces strongly-typed producers and consumers from AsyncAPI specifications (versions 2.6 and 3.0). AsyncAPI 2.6 channel-level publish operations generate receive-side consumers; subscribe operations generate send-side producers.

Both sides are generated from the same spec. The generated code handles all the messaging plumbing — payload serialization, JSON Schema validation, header encoding, channel address construction, error handling, and dead-letter routing — so you focus purely on business logic.

The generator leverages the Corvus.JsonSchema V5 engine for model generation, producing zero-allocation, pooled-memory types with full JSON Schema validation built in.

Installation

# Install the CLI tool globally
dotnet tool install --global Corvus.Json.Cli

# Or as a local tool
dotnet tool install Corvus.Json.Cli

For consumer/producer projects, add the runtime package:

dotnet add package Corvus.Text.Json.AsyncApi

Add a transport implementation:

# Choose one (or more) for your broker:
dotnet add package Corvus.Text.Json.AsyncApi.Nats
dotnet add package Corvus.Text.Json.AsyncApi.Kafka
dotnet add package Corvus.Text.Json.AsyncApi.Amqp
dotnet add package Corvus.Text.Json.AsyncApi.Mqtt
dotnet add package Corvus.Text.Json.AsyncApi.WebSocket
dotnet add package Corvus.Text.Json.AsyncApi.AzureServiceBus

Both also need the core library:

dotnet add package Corvus.Text.Json

Quick Start — Producer

Generate a typed producer from any AsyncAPI spec:

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

Use it with just a few lines:

using System.Text;
using Corvus.Text.Json.AsyncApi;
using Corvus.Text.Json.AsyncApi.Testing;
using Streetlights.Client;
using Streetlights.Client.Models;

// InMemoryMessageTransport for testing; in production, use NatsMessageTransport,
// KafkaMessageTransport, etc. — the API is identical.
await using InMemoryMessageTransport transport = new();

TurnOnProducer producer = new(transport);

// Publish a validated message — schema validation runs before the message leaves your process
await producer.PublishTurnOnOffAsync(
    payload: TurnOnOffPayload.Build(command: "on"u8, sentAt: DateTimeOffset.UtcNow),
    streetlightId: "lamp-001");

// Inspect what was published
PublishedMessage msg = transport.PublishedMessages[0];
Console.WriteLine($"Channel: {msg.Channel}");
Console.WriteLine($"Payload: {Encoding.UTF8.GetString(msg.PayloadBytes)}");

Quick Start — Consumer

Generate a typed consumer:

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

Implement your message handler and start consuming:

using Corvus.Text.Json.AsyncApi;
using Corvus.Text.Json.AsyncApi.Testing;
using Streetlights.Client;
using Streetlights.Client.Models;

await using InMemoryMessageTransport transport = new();

LightMeasuredHandler handler = new();
ReceiveLightMeasurementConsumer consumer = new(transport, handler);

// The channel address declares a {streetlightId} parameter, so the consumer takes it on
// StartAsync and subscribes to the address those parameters compose.
await consumer.StartAsync(streetlightId: "1");

// Simulate an incoming message (in production, the broker delivers these)
await transport.DeliverAsync<LightMeasuredPayload>(
    "smartylighting.streetlights.1.0.action.1.lighting.measured",
    """{"lumens":250,"sentAt":"2024-01-15T10:30:00Z"}"""u8.ToArray());

await consumer.StopAsync();

Implement your business logic:

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

What the Generator Produces

Producer Generation (`--mode producer`)

Generated artifact	What it handles
Producer class (e.g., `TurnOnProducer`)	Orchestrates publish lifecycle: builds payload, validates against JSON Schema, constructs channel address, publishes via transport
Message types (e.g., `TurnOnTurnOnOffMessage`)	Message-level metadata and content type constants
Model types (e.g., `TurnOnOffPayload`)	Strongly-typed JSON Schema models with validation, zero-allocation access, and builder patterns
Lock file (`corvusjson-asyncapi.lock`)	Tracks spec hash for incremental regeneration

Consumer Generation (`--mode consumer`)

Generated artifact	What it handles
Consumer class (e.g., `ReceiveLightMeasurementConsumer`)	Manages subscription lifecycle: subscribes to channel, deserializes payloads, validates, dispatches to handler, handles errors
Handler interface (e.g., `IReceiveLightMeasurementHandler`)	One async method per message — you implement business logic here
Model types	Same models as producer generation
Lock file	Tracks spec hash for incremental regeneration

Both (`--mode both`, the default)

Generates producers for send operations and consumers for receive operations in a single pass.

Namespace Layout

The generator places request/response infrastructure (producers, consumers, handler interfaces, message types) in the root namespace you specify, and JSON Schema model types in a .Models sub-namespace:

Streetlights.Client/              ← root namespace (--rootNamespace)
├── TurnOnProducer                ← producer class
├── ReceiveLightMeasurementConsumer  ← consumer class
├── IReceiveLightMeasurementHandler  ← handler interface
├── TurnOnTurnOnOffMessage        ← message type
└── Models/                       ← model sub-namespace
    ├── TurnOnOffPayload          ← JSON Schema model
    └── LightMeasuredPayload      ← JSON Schema model

Consumer code typically imports both namespaces:

using Streetlights.Client;         // producers, consumers, handlers
using Streetlights.Client.Models;  // payload model types

Generated Code Architecture

Producer Flow

Your Code → Producer.PublishAsync(payload, channelParams...)
  → Generated Workspace: build typed payload from Source
  → Generated Validation: validate payload against JSON Schema
  → Channel Address: construct from template + parameters (UTF-8)
  → Authentication: call IMessageAuthenticationProvider
  → IMessageTransport.PublishAsync: serialize and send
  → Cleanup: dispose the workspace, return the channel rental to the pool

Consumer Flow

Message arrives on transport
  → IMessageTransport: parse bytes into ParsedJsonDocument<TPayload>
  → Generated Consumer.HandleMessageAsync: receive typed payload
  → Generated Validation: validate payload (and optionally headers)
  → Your Handler: receives strongly-typed, validated payload
  → On error: IMessageErrorPolicy decides Skip/DeadLetter/Abort

Message Delivery Context

Consumers can opt into transport delivery metadata. For every receive operation that is not a request/reply responder, the generator emits a second handler/consumer pair alongside the legacy one, named by appending WithDeliveryContext to the operation name: I{Operation}WithDeliveryContextHandler and {Operation}WithDeliveryContextConsumer. The handler receives a MessageDeliveryContext carrying the subscribed channel (as UTF-8 bytes), the message headers, and the transport-native message when one exists (for example RabbitMQ's BasicDeliverEventArgs):

public sealed class LightMeasuredHandler : IReceiveLightMeasurementWithDeliveryContextHandler
{
    public ValueTask HandleLightMeasuredAsync(
        LightMeasuredPayload payload,
        MessageDeliveryContext context,
        CancellationToken cancellationToken = default)
    {
        // context.ChannelUtf8, context.Headers, context.NativeMessage
        return ValueTask.CompletedTask;
    }
}

A message that declares a typed headers schema keeps its typed headers parameter: the handler signature becomes (payload, headers, context, cancellationToken).

The rules that come with the capability:

The context is valid only for the duration of the handler invocation. Transports recycle the buffers it references once the handler returns (RabbitMQ, for example, reuses the network-frame buffer backing BasicDeliverEventArgs.Body). Copy anything you need to keep before returning.
The delivery-context consumer requires an IMessageDeliveryContextTransport. All seven transports implement it. When wrapping with instrumentation, use InstrumentedMessageTransport.Create so the wrapper preserves the capability.
A channel has one subscription. Every transport allows exactly one subscription per channel, of any kind: legacy data, delivery-context data, or responder. Subscribing a channel that already has one throws InvalidOperationException; unsubscribe first, which frees the slot for a subscription of any kind. Nothing is displaced silently, and an existing subscription is never touched by the subscribe path: a subscribe that loses a race for the slot tears down only the consumer it was building, though on some brokers that consumer may receive a message or two before the refusal lands. A handler may call UnsubscribeAsync for its own channel (the Abort error-policy path does): the subscription is released immediately, and the transport detects that the call came from inside the subscription's own handler and completes the broker-side teardown as the handler returns, rather than deadlocking by waiting for a handler that is waiting for it.
The two generated consumers for one operation are mutually exclusive. {Operation}Consumer and {Operation}WithDeliveryContextConsumer subscribe the same channel address, so start one or the other, not both. Starting the second throws rather than quietly taking the channel over.
Generated consumer lifecycle is strict. A second StartAsync on a started consumer throws; StopAsync on a consumer that is not started throws; DisposeAsync stops only a started consumer and completes quietly otherwise. A StopAsync (or the Abort error-policy path) that lands while the subscription is still being established is honored: the stop reports success, and the start releases the subscription it created and throws an InvalidOperationException saying the consumer was stopped during start. In that window a restart can be refused with "already has a subscription" because the superseded start has not yet released its claim at the transport — wait for the original start's exception to surface, then restart.
Direct transport API users own the channel buffer. SubscribeWithDeliveryContextAsync retains the channelUtf8 memory for the subscription's lifetime and hands it to every delivery, so do not subscribe with a pooled or reused buffer. (Generated consumers manage this for you.)
Bindings travel. An operation declaring channel or operation bindings subscribes through a MessageContext overload, mirroring the legacy consumer; the default implementation drops the context, and a transport that honors bindings can override it.

Validation

Every generated producer and consumer accepts a ValidationMode parameter that controls whether messages are validated against their JSON Schema, and the degree of diagnostic information provided on failure.

Validation Modes

Mode	Behaviour	Use for
`ValidationMode.None`	No validation	Trusted internal services, maximum throughput
`ValidationMode.Basic`	Fast boolean `EvaluateSchema()` check; on failure throws with message name but no details	Production default — catches malformed messages
`ValidationMode.Detailed`	Full evaluation with `JsonSchemaResultsCollector`; on failure includes JSON diagnostics with evaluation paths and error messages	Development, debugging, API gateways

Producer-Side Validation

Validation runs before the message leaves your process:

// Basic validation (default) — catches schema violations before publish
TurnOnProducer producer = new(transport, validationMode: ValidationMode.Basic);

try
{
    await producer.PublishTurnOnOffAsync(
        payload: TurnOnOffPayload.Build(command: "invalid-command"u8, sentAt: DateTimeOffset.UtcNow),
        streetlightId: "lamp-001");
}
catch (ArgumentException ex)
{
    // "Message payload validation failed for 'payload'."
    Console.WriteLine(ex.Message);
}

Consumer-Side Validation

Validation runs after deserialization, before your handler is called:

// Detailed validation — full diagnostics on failure
ReceiveLightMeasurementConsumer consumer = new(
    transport,
    handler,
    validationMode: ValidationMode.Detailed);

If validation fails, the error policy determines what happens (skip, dead-letter, or abort). Your handler is never called with invalid data.

Error Handling

Error Policies

When message processing fails (validation error, handler exception, transport issue), the IMessageErrorPolicy decides the terminal action:

Action	Behaviour
`MessageErrorAction.Skip`	Discard the message and continue processing the next one
`MessageErrorAction.DeadLetter`	Publish the failed message to a dead-letter channel for later inspection
`MessageErrorAction.Abort`	Stop the consumer subscription entirely

Default Policy

The DefaultMessageErrorPolicy applies sensible defaults:

Error kind	Default action
Deserialization failure	Dead-letter
Handler exception	Dead-letter
Transport connectivity error	Abort

// Customise: skip deserialization errors, dead-letter handler errors, abort on transport issues
IMessageErrorPolicy policy = new DefaultMessageErrorPolicy(
    deserializationAction: MessageErrorAction.Skip,
    handlerAction: MessageErrorAction.DeadLetter,
    transportAction: MessageErrorAction.Abort);

ReceiveLightMeasurementConsumer consumer = new(
    transport, handler,
    errorPolicy: policy);

Custom Error Policies

Implement IMessageErrorPolicy for complex routing logic:

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);
    }
}

Dead-Letter Channels

Dead-lettered messages are published to the subscribed channel's address behind a dead-letter. prefix (e.g., dead-letter.smartylighting.streetlights.1.0.action.lamp-42.lighting.measured). Whether the channel address is static, composed from parameters, or supplied by the caller at start, the dead-letter address is derived from the channel the consumer actually subscribed. The generated consumer calls IMessageTransport.DeadLetterAsync when the error policy returns MessageErrorAction.DeadLetter, so validation failures and handler exceptions are handled consistently across transports.

The dead-letter message includes:

The original payload bytes
The original headers
The exception that caused the failure
The original channel address

Each transport maps this to the most natural broker mechanism: Kafka, NATS, MQTT, AMQP, and WebSocket publish a new message to a configured dead-letter topic/subject/channel; Azure Service Bus also uses the broker's native dead-letter settlement path for messages already being processed by its receiver loop.

Resilience (Polly Integration)

The Corvus.Text.Json.AsyncApi.Polly package provides handler middleware backed by Polly resilience pipelines.

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

Configuring Retry with Circuit Breaker

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

ResiliencePipeline pipeline = new ResiliencePipelineBuilder()
    .AddRetry(new RetryStrategyOptions
    {
        MaxRetryAttempts = 3,
        BackoffType = DelayBackoffType.Exponential,
        Delay = TimeSpan.FromMilliseconds(200),
    })
    .AddCircuitBreaker(new CircuitBreakerStrategyOptions
    {
        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 operation still fails, the exception propagates to the IMessageErrorPolicy for terminal action (dead-letter, skip, or abort).

Transport Implementations

All transports implement IMessageTransport and can be swapped without changing your producer/consumer code.

Package	Broker	Protocol
`Corvus.Text.Json.AsyncApi.Nats`	NATS	NATS (JetStream compatible)
`Corvus.Text.Json.AsyncApi.Kafka`	Apache Kafka	Kafka
`Corvus.Text.Json.AsyncApi.Amqp`	RabbitMQ	AMQP 0-9-1
`Corvus.Text.Json.AsyncApi.Mqtt`	Mosquitto, HiveMQ, etc.	MQTT 3.1.1 / 5.0
`Corvus.Text.Json.AsyncApi.WebSocket`	Any WebSocket server	WebSocket
`Corvus.Text.Json.AsyncApi.AzureServiceBus`	Azure Service Bus	Queues and topics
`Corvus.Text.Json.AsyncApi.Testing`	In-memory (tests only)	None

For durable consumption, acknowledgement, redelivery, and restart behavior, see AsyncAPI message resumption.

NATS

using Corvus.Text.Json.AsyncApi.Nats;

NatsTransportOptions options = new()
{
    Url = "nats://broker.example.com:4222",
    Name = "streetlights-service",
    RequestTimeout = TimeSpan.FromSeconds(10),
};

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

Kafka

using Corvus.Text.Json.AsyncApi.Kafka;

KafkaTransportOptions options = new()
{
    BootstrapServers = "kafka.example.com:9092",
    GroupId = "streetlights-consumer-group",
    AutoOffsetReset = Confluent.Kafka.AutoOffsetReset.Earliest,
};

await using KafkaMessageTransport transport = new(options);

AMQP (RabbitMQ)

using Corvus.Text.Json.AsyncApi.Amqp;

AmqpTransportOptions options = new()
{
    ConnectionUri = "amqp://guest:guest@rabbitmq.example.com:5672/",
    ExchangeName = "streetlights",
    ExchangeType = "topic",
};

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

MQTT

using Corvus.Text.Json.AsyncApi.Mqtt;

MqttTransportOptions options = new()
{
    Host = "mqtt.example.com",
    Port = 1883,
    ClientId = "streetlights-publisher",
    QualityOfServiceLevel = MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce,
};

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

WebSocket

using Corvus.Text.Json.AsyncApi.WebSocket;

WebSocketTransportOptions options = new()
{
    ServerUri = "wss://ws.example.com/events",
};

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

Azure Service Bus

using Corvus.Text.Json.AsyncApi.AzureServiceBus;

AzureServiceBusTransportOptions options = new()
{
    ConnectionString = "<service-bus-connection-string>",
    QueueName = "streetlights",
    RequestTimeout = TimeSpan.FromSeconds(30),
};

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

Use UseTopic = true with TopicName and SubscriptionName for pub/sub:

AzureServiceBusTransportOptions options = new()
{
    ConnectionString = "<service-bus-connection-string>",
    UseTopic = true,
    TopicName = "streetlights",
    SubscriptionName = "processors",
};

For Microsoft Entra ID authentication, set FullyQualifiedNamespace and Credential instead of ConnectionString.

In-Memory (Testing)

The Corvus.Text.Json.AsyncApi.Testing package provides an in-memory transport for unit and integration tests:

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

using Corvus.Text.Json.AsyncApi.Testing;

InMemoryMessageTransport transport = new();

// Use with producers — messages are captured for assertions
TurnOnProducer producer = new(transport, ValidationMode.None);

await producer.PublishTurnOnOffAsync(
    payload: TurnOnOffPayload.Build(command: "on"u8, sentAt: DateTimeOffset.UtcNow),
    streetlightId: "lamp-001");

// Assert published messages
Assert.AreEqual(1, transport.PublishedMessages.Count);

// Deliver messages to consumers for testing — target the address the consumer's
// parameters composed, not the literal template
await transport.DeliverAsync<LightMeasuredPayload>(
    "smartylighting.streetlights.1.0.action.lamp-001.lighting.measured",
    """{"lumens":250,"sentAt":"2024-01-15T10:30:00Z"}"""u8.ToArray());

Local Integration Test Troubleshooting

The transport integration tests use Testcontainers and need either Docker or Podman running locally. When running targeted transport tests, include the integration category intentionally; for normal test runs, continue to exclude it.

dotnet test --project tests\Corvus.Text.Json.AsyncApi.Transport.IntegrationTests\Corvus.Text.Json.AsyncApi.Transport.IntegrationTests.csproj -c Debug -f net10.0 --filter "FullyQualifiedName~KafkaTransportTests&TestCategory=integration&TestCategory!=failing&TestCategory!=outerloop"

Common issues:

Symptom	Check
Testcontainers cannot connect	Start Docker Desktop or Podman and ensure the `DOCKER_HOST` environment variable points at the active socket when Podman is not the default.
Kafka consumers time out	Use the Testcontainers-provided bootstrap address and create topics before subscribing; do not hard-code `localhost:9092`.
Azure Service Bus emulator exits during startup	The bind-mounted emulator config file must be readable inside the Linux container. On Docker, this requires owner read/write and group/world read permissions.
RabbitMQ/MQTT/NATS tests fail after a previous interrupted run	Remove stale containers and volumes with the container engine's cleanup command, then rerun the targeted test class.

Authentication

Generated producers and consumers accept an optional IMessageAuthenticationProvider. The provider supplies credentials to the transport before messages are published or subscriptions are established.

The generated code determines which security scheme to use from the AsyncAPI spec's securitySchemes. It constructs a MessageAuthenticationContext with the scheme type and name, passes it to your provider, then the transport reads the populated credentials. This is called once per publish (producers) or once at subscription time (consumers).

How It Works

Your code constructs a producer/consumer with an IMessageAuthenticationProvider
  → Generated code creates MessageAuthenticationContext(schemeType, schemeName)
  → Calls provider.AuthenticateAsync(context, ct)
  → Provider populates context.Credentials dictionary
  → Transport reads credentials to configure connection/message

The MessageAuthenticationContext.Credentials dictionary is transport-agnostic — each transport reads the keys it understands:

Credential key	Used by
`token`	Bearer token transports (NATS auth token, Kafka OAUTHBEARER)
`username` / `password`	SASL PLAIN, AMQP, MQTT username/password
`key`	API key (transport-specific header or connection param)
`access_token` / `token_type` / `scopes`	OAuth2 flows
`certificate`	TLS client certificate (base64 PFX)

Microsoft Entra ID (Azure AD / MSAL)

For Azure-hosted messaging (Event Hubs via Kafka protocol, Azure Service Bus via AMQP), use Azure.Identity with the OAuth2AuthenticationProvider's token factory.

Requires: dotnet add package Azure.Identity

using Azure.Core;
using Azure.Identity;
using Corvus.Text.Json.AsyncApi;
using Corvus.Text.Json.AsyncApi.Testing;
using Streetlights.Client;
using Streetlights.Client.Models;

await using InMemoryMessageTransport transport = new();

// Client credentials flow (service-to-service)
string tenantId = "your-tenant-id";
string clientId = "your-client-id";
string clientSecret = "your-client-secret";

TokenCredential credential = new ClientSecretCredential(tenantId, clientId, clientSecret);

IMessageAuthenticationProvider auth = new OAuth2AuthenticationProvider(
    accessTokenFactory: async ct =>
    {
        AccessToken token = await credential.GetTokenAsync(
            new TokenRequestContext(["https://eventhubs.azure.net/.default"]),
            ct);
        return token.Token;
    },
    tokenType: "Bearer");

TurnOnProducer producer = new(transport, authProvider: auth);

await producer.PublishTurnOnOffAsync(
    payload: TurnOnOffPayload.Build(command: "on"u8, sentAt: DateTimeOffset.UtcNow),
    streetlightId: "lamp-001");

For different Azure Identity flows, substitute the credential type:

using Azure.Identity;

// Managed identity (Azure-hosted services — no secrets needed)
TokenCredential managedIdentity = new DefaultAzureCredential();

// Interactive browser login (desktop/native apps)
TokenCredential browser = new InteractiveBrowserCredential(
    new InteractiveBrowserCredentialOptions
    {
        ClientId = "your-client-id",
        TenantId = "your-tenant-id",
    });

// Device code flow (CLI tools, headless terminals)
TokenCredential deviceCode = new DeviceCodeCredential(
    new DeviceCodeCredentialOptions
    {
        ClientId = "your-client-id",
        TenantId = "your-tenant-id",
        DeviceCodeCallback = (info, cancel) =>
        {
            Console.WriteLine(info.Message);
            return Task.CompletedTask;
        },
    });

All credential types work with the same OAuth2AuthenticationProvider factory pattern — the factory acquires tokens using whichever TokenCredential you supply.

OAuth 2.0 (Static Token)

For short-lived scripts, testing, or pre-acquired tokens:

using Corvus.Text.Json.AsyncApi;
using Corvus.Text.Json.AsyncApi.Testing;
using Streetlights.Client;
using Streetlights.Client.Models;

await using InMemoryMessageTransport transport = new();

IMessageAuthenticationProvider auth = new OAuth2AuthenticationProvider(
    accessToken: "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...",
    tokenType: "Bearer",
    scopes: "read:messages write:messages");

TurnOnProducer producer = new(transport, authProvider: auth);

await producer.PublishTurnOnOffAsync(
    payload: TurnOnOffPayload.Build(command: "off"u8, sentAt: DateTimeOffset.UtcNow),
    streetlightId: "lamp-002");

The accessTokenFactory is called on every publish (producers) and once at subscribe time (consumers). For caching and refresh logic, wrap your token client with Microsoft.Extensions.Caching.Memory or use your SDK's built-in token cache.

Bearer Token (Simple JWT)

For brokers that accept a static JWT or pre-acquired bearer token:

using Corvus.Text.Json.AsyncApi;
using Corvus.Text.Json.AsyncApi.Testing;
using Streetlights.Client;
using Streetlights.Client.Models;

await using InMemoryMessageTransport transport = new();

IMessageAuthenticationProvider auth = new BearerTokenAuthenticationProvider("my-jwt-token");

TurnOnProducer producer = new(transport, authProvider: auth);

await producer.PublishTurnOnOffAsync(
    payload: TurnOnOffPayload.Build(command: "on"u8, sentAt: DateTimeOffset.UtcNow),
    streetlightId: "lamp-003");

With dynamic token refresh:

using Corvus.Text.Json.AsyncApi;
using Corvus.Text.Json.AsyncApi.Testing;
using Streetlights.Client;
using Streetlights.Client.Models;

await using InMemoryMessageTransport transport = new();
LightMeasuredHandler handler = new();

// In production, this factory would call your token endpoint
IMessageAuthenticationProvider auth = new BearerTokenAuthenticationProvider(
    tokenFactory: ct => new ValueTask<string>("refreshed-token-value"));

ReceiveLightMeasurementConsumer consumer = new(transport, handler, authProvider: auth);
await consumer.StartAsync(streetlightId: "1");
await consumer.StopAsync();

internal sealed class LightMeasuredHandler : IReceiveLightMeasurementHandler
{
    public ValueTask HandleLightMeasuredAsync(
        LightMeasuredPayload payload, CancellationToken cancellationToken = default) => default;
}

API Key

For brokers or gateways that authenticate via API keys:

using Corvus.Text.Json.AsyncApi;
using Corvus.Text.Json.AsyncApi.Testing;
using Streetlights.Client;
using Streetlights.Client.Models;

await using InMemoryMessageTransport transport = new();
LightMeasuredHandler handler = new();

// Simple key — transport uses it as-is
IMessageAuthenticationProvider simpleAuth = new ApiKeyAuthenticationProvider(
    apiKey: "sk-live-abc123");

// HTTP API key style — with name and location metadata
IMessageAuthenticationProvider namedAuth = new ApiKeyAuthenticationProvider(
    apiKey: "sk-live-abc123",
    name: "X-API-Key",
    location: "header");

ReceiveLightMeasurementConsumer consumer = new(transport, handler, authProvider: namedAuth);
await consumer.StartAsync(streetlightId: "1");
await consumer.StopAsync();

internal sealed class LightMeasuredHandler : IReceiveLightMeasurementHandler
{
    public ValueTask HandleLightMeasuredAsync(
        LightMeasuredPayload payload, CancellationToken cancellationToken = default) => default;
}

Username/Password (SASL)

For Kafka SASL PLAIN, AMQP PLAIN, or MQTT username/password authentication:

using Corvus.Text.Json.AsyncApi;
using Corvus.Text.Json.AsyncApi.Testing;
using Streetlights.Client;
using Streetlights.Client.Models;

await using InMemoryMessageTransport transport = new();

IMessageAuthenticationProvider auth = new UserPasswordAuthenticationProvider(
    username: "service-account",
    password: "secret-from-keyvault");

TurnOnProducer producer = new(transport, authProvider: auth);

await producer.PublishTurnOnOffAsync(
    payload: TurnOnOffPayload.Build(command: "off"u8, sentAt: DateTimeOffset.UtcNow),
    streetlightId: "lamp-005");

Client Certificate (mTLS)

For brokers requiring mutual TLS (Kafka SSL, RabbitMQ peer verification):

using System.Security.Cryptography;
using System.Security.Cryptography.X509Certificates;
using Corvus.Text.Json.AsyncApi;
using Corvus.Text.Json.AsyncApi.Testing;
using Streetlights.Client;
using Streetlights.Client.Models;

await using InMemoryMessageTransport transport = new();

// In production, load from PEM, PFX, or certificate store:
//   X509Certificate2 cert = X509Certificate2.CreateFromPemFile("client-cert.pem", "client-key.pem");
//   X509Certificate2 cert = new("client.pfx", "pfx-password");

// Self-signed cert for demonstration:
using RSA rsa = RSA.Create(2048);
CertificateRequest req = new("CN=example-client", rsa, HashAlgorithmName.SHA256, RSASignaturePadding.Pkcs1);
using X509Certificate2 cert = req.CreateSelfSigned(DateTimeOffset.UtcNow, DateTimeOffset.UtcNow.AddHours(1));

IMessageAuthenticationProvider auth = new CertificateAuthenticationProvider(cert);

TurnOnProducer producer = new(transport, authProvider: auth);

Composite (Multiple Security Schemes)

When your AsyncAPI spec declares multiple security schemes on a server (e.g., both SASL and API key), the generated code calls AuthenticateAsync with different SchemeType values. Use CompositeAuthenticationProvider to route to the correct provider:

using Corvus.Text.Json.AsyncApi;
using Corvus.Text.Json.AsyncApi.Testing;
using Streetlights.Client;
using Streetlights.Client.Models;

await using InMemoryMessageTransport transport = new();

IMessageAuthenticationProvider auth = new CompositeAuthenticationProvider(
    new KeyValuePair<SecuritySchemeType, IMessageAuthenticationProvider>[]
    {
        new(SecuritySchemeType.ScramSha256, new UserPasswordAuthenticationProvider("user", "pass")),
        new(SecuritySchemeType.Http, new BearerTokenAuthenticationProvider("token-value")),
        new(SecuritySchemeType.HttpApiKey, new ApiKeyAuthenticationProvider("api-key-123")),
    });

// The generated code emits the correct SchemeType per operation/server —
// the composite automatically selects the matching inner provider.
TurnOnProducer producer = new(transport, authProvider: auth);

await producer.PublishTurnOnOffAsync(
    payload: TurnOnOffPayload.Build(command: "on"u8, sentAt: DateTimeOffset.UtcNow),
    streetlightId: "lamp-006");

Custom Authentication Provider

For protocols with non-standard auth requirements, implement IMessageAuthenticationProvider directly:

using Corvus.Text.Json.AsyncApi;
using Corvus.Text.Json.AsyncApi.Testing;
using Streetlights.Client;
using Streetlights.Client.Models;

await using InMemoryMessageTransport transport = new();

// Custom provider that acquires credentials from a secret rotation service
IMessageAuthenticationProvider auth = new RotatingSecretProvider(
    secretFactory: ct => new ValueTask<(string User, string Pass)>(("svc-acct", "rotated-secret")));

TurnOnProducer producer = new(transport, authProvider: auth);

await producer.PublishTurnOnOffAsync(
    payload: TurnOnOffPayload.Build(command: "off"u8, sentAt: DateTimeOffset.UtcNow),
    streetlightId: "lamp-007");

The implementation:

using Corvus.Text.Json.AsyncApi;

internal sealed class RotatingSecretProvider : IMessageAuthenticationProvider
{
    private readonly Func<CancellationToken, ValueTask<(string User, string Pass)>> secretFactory;

    public RotatingSecretProvider(
        Func<CancellationToken, ValueTask<(string User, string Pass)>> secretFactory)
    {
        this.secretFactory = secretFactory;
    }

    public async ValueTask AuthenticateAsync(
        MessageAuthenticationContext context,
        CancellationToken cancellationToken = default)
    {
        (string user, string pass) = await this.secretFactory(cancellationToken).ConfigureAwait(false);
        context.Credentials["username"] = user;
        context.Credentials["password"] = pass;
    }
}

See also: Example Recipe 039 — AsyncAPI Authentication for a fully compilable project demonstrating all authentication patterns.

Channel Parameters

AsyncAPI channels can contain parameters (e.g., smartylighting.streetlights.1.0.action.{streetlightId}.turn.on). The generated producer exposes these as method parameters:

// The streetlightId parameter is part of the publish method signature
await producer.PublishTurnOnOffAsync(
    payload: TurnOnOffPayload.Build(command: "on"u8, sentAt: DateTimeOffset.UtcNow),
    streetlightId: "lamp-42");
// Wire: publishes to "smartylighting.streetlights.1.0.action.lamp-42.turn.on"

The generated code constructs the channel address from the template using zero-allocation UTF-8 byte manipulation with pooled buffers — no string concatenation or allocation on the hot path.

Consumers

A consumer for a parameterised channel takes the same parameters on StartAsync, and subscribes to the address they compose:

await using ReceiveLightMeasurementConsumer consumer = new(transport, handler);

// Subscribes to "smartylighting.streetlights.1.0.action.lamp-42.lighting.measured"
await consumer.StartAsync(streetlightId: "lamp-42");

A ReadOnlySpan<char> overload sits beneath the string one, so a caller that already holds a span does not create a string just to have it measured and copied:

await consumer.StartAsync(streetlightId: idSpan);

The address is composed the same way the producer composes its own: the template is split when the code is generated, so its literal parts are u8 literals and only the parameter values are transcoded, filled once into the array the subscription retains. The dead-letter address is built from those same bytes rather than by concatenating a second string.

Before 5.3.0 a consumer for a parameterised channel had no parameters on StartAsync and subscribed to the template literally, placeholder and all — so it listened on a channel no publisher ever wrote to. Supplying the parameters is therefore a breaking change with a fix inside it: the call that compiled before was subscribing to nothing real.

Message Headers

When an AsyncAPI message defines a headers schema (directly or via message traits), the generated code produces typed header structures. Headers provide metadata about the message — correlation IDs, trace context, content versioning — separate from the payload.

Defining Headers in the Spec

Headers are typically defined via message traits (shared across multiple messages):

{
  "components": {
    "schemas": {
      "CommonHeaders": {
        "type": "object",
        "properties": {
          "correlationId": { "type": "string", "format": "uuid" },
          "timestamp": { "type": "string", "format": "date-time" }
        },
        "required": ["correlationId"]
      }
    },
    "messageTraits": {
      "commonHeaders": {
        "headers": { "$ref": "#/components/schemas/CommonHeaders" }
      }
    },
    "messages": {
      "userSignedUp": {
        "payload": { "$ref": "#/components/schemas/UserSignedUpPayload" },
        "traits": [{ "$ref": "#/components/messageTraits/commonHeaders" }]
      }
    }
  }
}

Producing Messages with Headers

The generated producer accepts a typed CommonHeaders.Source parameter alongside the payload:

using Corvus.Text.Json.AsyncApi;

await producer.PublishUserSignedUpAsync(
    payload: UserSignedUpPayload.Build(email: "alice@example.com"u8, userId: "user-123"u8),
    headers: CommonHeaders.Build(
        correlationId: Guid.NewGuid().ToString("D"),
        timestamp: DateTimeOffset.UtcNow.ToString("O")));

The transport serializes headers to a JSON object using a thread-static pooled Utf8JsonWriter, keeping allocation constant (152 bytes) regardless of header count. For transports whose header slots are string-typed (MQTT, NATS), the compact JSON text itself is the value: the writer's default encoder escapes non-ASCII characters, so the value is ASCII-safe for those protocols without a base64 expansion.

Consuming Messages with Headers

The generated consumer's handler interface includes the typed headers parameter:

using Corvus.Text.Json.AsyncApi;

// Generated handler interface includes typed headers
public interface IConsumeUserSignedUpHandler
{
    ValueTask HandleUserSignedUpAsync(
        UserSignedUpPayload payload,
        CommonHeaders headers,
        CancellationToken cancellationToken);
}

// Your handler implementation with typed header access
public class UserSignedUpHandler : IConsumeUserSignedUpHandler
{
    public ValueTask HandleUserSignedUpAsync(
        UserSignedUpPayload payload,
        CommonHeaders headers,
        CancellationToken cancellationToken)
    {
        // Typed access — no string parsing, no dictionary lookup
        string correlationId = (string)headers.CorrelationId;
        DateTimeOffset timestamp = DateTimeOffset.Parse((string)headers.Timestamp);

        Console.WriteLine($"[{correlationId}] User {payload.UserId} signed up at {timestamp}");
        return ValueTask.CompletedTask;
    }
}

Messages without Headers

When the spec does not define headers for a message, the generated handler interface omits the headers parameter entirely:

// Generated handler — no headers parameter since the spec defines none
public interface IReceiveLightMeasurementHandler
{
    ValueTask HandleLightMeasuredAsync(
        LightMeasuredPayload payload,
        CancellationToken cancellationToken = default);
}

The transport still delivers headers internally (for error policy context and dead-lettering), but your handler code never sees them. If you later add a headers schema to the message in your spec and regenerate, the handler interface gains the typed headers parameter automatically.

Header Validation

Headers are validated alongside the payload when ValidationMode is Basic or Detailed. Validation uses the same compiled-schema approach as payload validation:

using Corvus.Text.Json.AsyncApi;

// Validation mode applies to both payload AND headers
ReceiveUserSignedUpConsumer consumer = new(
    transport,
    handler,
    validationMode: ValidationMode.Detailed);

// If a message arrives with an invalid correlationId (e.g., not a UUID),
// the consumer catches the validation failure and routes through the error policy.
// With Detailed mode, the ArgumentException includes full schema evaluation results:
//   "Message headers validation failed: /correlationId: format 'uuid' validation failed"

The generated ValidateHeaders<THeaders> method mirrors ValidatePayload<TPayload>:

// Generated validation (internal to the consumer)
private static void ValidateHeaders<THeaders>(THeaders headers, ValidationMode mode)
    where THeaders : struct, IJsonElement<THeaders>
{
    if (mode == ValidationMode.Basic)
    {
        if (!headers.EvaluateSchema())
        {
            ThrowHelper.ThrowMessageHeadersValidationFailed("headers");
        }
    }
    else if (mode == ValidationMode.Detailed)
    {
        using var collector = JsonSchemaResultsCollector.Create(JsonSchemaResultsLevel.Detailed);
        if (!headers.EvaluateSchema(collector))
        {
            ThrowHelper.ThrowMessageHeadersValidationFailed(
                "headers", SchemaValidationDetail.FormatResults(collector));
        }
    }
}

Transport-Level Header Encoding

Different transports handle header serialization differently:

Transport	Header mechanism	Notes
Kafka	Native message headers (`Message.Headers`)	Key-value byte pairs
AMQP	Application properties	Native key-value map
Azure Service Bus	Application properties (`ServiceBusMessage.ApplicationProperties`)	User metadata map; values are encoded as strings and reconstructed as typed JSON headers for generated handlers
NATS	Compact JSON text in message headers	Values are ASCII-escaped JSON, legal per the NATS header rules
MQTT	User properties (MQTT 5) carrying compact JSON text	MQTT 5 user-property values are UTF-8 strings
WebSocket	JSON envelope field	Framed alongside payload

The transport layer handles encoding/decoding transparently — your handler always receives the typed struct regardless of the wire format.

Azure Service Bus deserves one extra note because it has both broker system properties and user application properties. Corvus writes AsyncAPI message headers to ApplicationProperties; it does not use them for broker control fields such as CorrelationId, ReplyTo, or SessionId. Request/reply support uses those native Service Bus fields separately, while your AsyncAPI header schema remains in the application-property map and is passed to generated handlers as the same typed header struct used by other transports.

Request/Reply

AsyncAPI operations can model request/reply patterns. The generator produces methods that send a request and await a correlated response:

// Generated request/reply method — SendAndReceive{Message}Async, returning the typed reply.
// The workspace you pass owns the reply: it stays valid until the workspace is disposed.
using JsonWorkspace workspace = JsonWorkspace.CreateUnrented();
QueryResponse reply = await queryProducer.SendAndReceiveQueryAsync(
    payload: QueryPayload.Build(filter: "status=active"u8),
    workspace: workspace,
    cancellationToken: ct);

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

The generated code handles correlation ID generation (GUID formatted directly to a byte[36] — no string allocation) and request/reply channel pairing; cancellation flows through the supplied token.

AsyncAPI 3.0 uses the standard operation reply object. AsyncAPI 2.6 has correlationId but no standard reply object, so Corvus supports an explicit x-corvus-reply extension on a 2.6 operation. The extension mirrors the 3.0 shape closely enough for the generated request/reply method to use the same runtime path:

{
  "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 extension is intentionally explicit. The generator does not infer request/reply pairs from matching correlationId values because that is ambiguous in real-world 2.6 documents.

Responder (request/reply receive)

The methods above are the requester half of request/reply (send a request, await the reply). The responder half — receive a request and send back a correlated reply — is modelled by a receive operation that declares a reply. For these operations the generator produces a reply-returning handler and a consumer that publishes the reply for you.

The handler returns the reply payload instead of void:

public sealed class CalculateHandler : ICalculateHandler
{
    // A receive operation with a reply: return the reply payload; the consumer publishes it.
    public ValueTask<CalculateResponse> HandleCalculateRequestAsync(
        CalculateRequest payload,
        CancellationToken cancellationToken = default)
    {
        int sum = payload.A + payload.B;
        return ValueTask.FromResult(new CalculateResponse.Source((ref CalculateResponse.Builder b) =>
        {
            b.Create(result: sum);
        }));
    }
}

The generated consumer subscribes through the transport's responder primitive:

ValueTask SubscribeReplyAsync<TRequest, TReply>(
    ReadOnlyMemory<byte> channelUtf8,
    Func<TRequest, JsonElement, CancellationToken, ValueTask<TReply>> handler,
    CancellationToken cancellationToken = default)
    where TRequest : struct, IJsonElement<TRequest>
    where TReply : struct, IJsonElement<TReply>;

The transport owns correlation: for each delivered request it reads the request's reply-to address and correlation id (native broker fields — the same CorrelationId/ReplyTo the requester sets), invokes the handler, and publishes the returned reply to the reply-to address correlated to the request. The handler never sees the correlation plumbing.

Reply ownership. RequestAsync takes a JsonWorkspace and threads it through to the parse of the reply: the returned payload and headers are views over documents that workspace owns, so they stay valid until the workspace is disposed. The generated SendAndReceive* requester surfaces the same contract. You pass the workspace that will own the reply, and the reply stays valid until you dispose it. Dispose the workspace once the reply is no longer needed. This mirrors IApiResponse on the OpenAPI side, which owns its parsed response body and is itself disposable. (Internally the requester builds the outgoing request in a separate short-lived workspace, disposed when the exchange completes; it never affects the reply's lifetime.)

Implementation status. Every transport implements SubscribeReplyAsync: the broker transports (NATS, Kafka, AMQP, MQTT, WebSocket, Azure Service Bus) each run a real responder over their native correlation fields, and the in-memory testing transport implements a full in-process round-trip — a RequestAsync call delivers the request to a registered responder, whose reply completes the requester's pending call. With no responder registered, the request is delivered to any data subscription on the channel (plain or delivery-context, exactly as a publish would be) and parked for the test helper CompleteRequest. A responder subscription occupies its channel's single subscription slot like any other, and the internal reply-channel consumers that serve RequestAsync live apart from that registry with the transport's own lifetime, so a cancelled request neither kills request-reply on the channel nor blocks an application subscription on the reply channel.

Bindings

AsyncAPI bindings provide protocol-specific configuration. The generator captures bindings at three levels and makes them available to the transport via MessageContext:

Level	Example	What it configures
Channel	Kafka topic configuration	Partition count, retention
Operation	Kafka consumer group	Group ID, offset reset
Message	Kafka message key	Partition key, headers

{
  "channels": {
    "events": {
      "bindings": {
        "kafka": {
          "topic": "streetlights.events",
          "partitions": 12,
          "replicas": 3
        }
      }
    }
  }
}

The generated code emits bindings as static ReadOnlyMemory<byte> constants in the producer/consumer, passing them to the transport in the MessageContext:

// Transport receives bindings in the context — parse as needed
MessageContext context = new()
{
    ContentType = "application/json",
    ChannelBindingsJson = /* static UTF-8 JSON bytes from the spec */,
    OperationBindingsJson = /* ... */,
    MessageBindingsJson = /* ... */,
};

Current transports primarily use their strongly-typed transport options for delivery semantics. MessageContext is the extension point for transports that need to inspect binding JSON directly; if you rely on a specific binding, verify that the selected transport implements it rather than assuming every binding affects runtime behavior automatically.

Telemetry and Observability

OpenTelemetry Distributed Tracing

Wrap any transport with InstrumentedMessageTransport.Create to gain automatic OpenTelemetry instrumentation:

using Corvus.Text.Json.AsyncApi;
using Corvus.Text.Json.AsyncApi.Nats;

NatsMessageTransport raw = await NatsMessageTransport.CreateAsync(
    new NatsTransportOptions { Url = "nats://localhost:4222" });

IMessageTransport transport = InstrumentedMessageTransport.Create(raw, "nats");

Create returns a wrapper that preserves the wrapped transport's optional capabilities. It implements IMessageDeliveryContextTransport and IHealthCheckableTransport exactly when the wrapped transport does, so a capability probe such as transport is IHealthCheckableTransport answers for the wrapped transport and health checks keep working. The new InstrumentedMessageTransport(raw, "nats") constructor form always produces a plain IMessageTransport wrapper that surfaces neither capability; prefer Create.

When the transport is delivery-context capable and you want to hand the wrapper straight to a generated *WithDeliveryContextConsumer, take the result as IMessageDeliveryContextTransport. The overload selected for a capability-typed argument returns that type, so no cast is needed:

using Corvus.Text.Json.AsyncApi;
using Corvus.Text.Json.AsyncApi.Nats;

NatsMessageTransport raw = await NatsMessageTransport.CreateAsync(
    new NatsTransportOptions { Url = "nats://localhost:4222" });

IMessageDeliveryContextTransport transport = InstrumentedMessageTransport.Create(raw, "nats");
ReceiveLightMeasurementWithDeliveryContextConsumer consumer = new(transport, handler);

This provides:

Distributed trace spans (Activities) for publish, subscribe, request, and dead-letter operations
Metrics — counters and histograms for message throughput, processing duration, and errors
W3C trace context propagation — traceparent/tracestate headers are injected on publish and extracted on consume, linking producer and consumer traces automatically

All instrumentation is zero-cost when no listener (e.g., OTLP exporter) is attached.

Processing Loop Heartbeat

The ProcessingLoopHeartbeat monitors consumer liveness by tracking heartbeat ticks from each subscription's processing loop:

ProcessingLoopHeartbeat heartbeat = new();

NatsTransportOptions options = new()
{
    Url = "nats://localhost:4222",
    Heartbeat = heartbeat,
};

// After starting consumers, check liveness — IsAlive is an exact channel-key lookup
// (no wildcard matching), so pass the composed address the consumer subscribed:
bool alive = heartbeat.IsAlive("smartylighting.streetlights.1.0.action.1.lighting.measured");

// Get all subscription statuses:
foreach (var status in heartbeat.GetSubscriptionStatuses())
{
    Console.WriteLine($"{status.Channel}: {(status.IsAlive ? "alive" : "STALE")}");
}

If a loop has not ticked for longer than the staleness threshold (default: 60 seconds), it is considered dead. This catches loops that exit silently due to unhandled exceptions or unexpected cancellation.

Health Checks

The Corvus.Text.Json.AsyncApi.HealthChecks package integrates with ASP.NET Core health checks:

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

using Corvus.Text.Json.AsyncApi.HealthChecks;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddHealthChecks()
    .AddAsyncApiTransport("nats-transport", transport);

var app = builder.Build();
app.MapHealthChecks("/health");
app.Run();

The health check extension works with transports that implement IHealthCheckableTransport (KafkaMessageTransport, NatsMessageTransport, AmqpMessageTransport, and InMemoryMessageTransport). It monitors transport connectivity by checking IsConnected and calling PingAsync.

Use ProcessingLoopHeartbeat separately when you need per-subscription liveness/staleness detection.

CLI Reference

Generate Code (`asyncapi-generate`)

corvusjson asyncapi-generate <specFile> [options]

Option	Description	Default
`<specFile>`	Path to the AsyncAPI specification (JSON or YAML)	Required
`--rootNamespace`	Root namespace for generated types	`GeneratedAsyncApi`
`--outputPath`	Directory to write generated code	`./Generated`
`--mode`	Generation mode: `producer`, `consumer`, or `both`	`both`
`--force`	Regenerate even if lock file indicates no changes	`false`
`--spec-url`	URL to fetch the spec from (recorded in lock file)	—
`--yaml`	Enable YAML support explicitly; otherwise `.yaml`/`.yml` is auto-detected	Auto
`--include-channel`	Glob patterns for channels to include	All
`--exclude-channel`	Glob patterns for channels to exclude	None
`--tag`	Filter by operation tags	All
`--specVersion`	Force spec version (`2.6` or `3.0`); auto-detected if omitted	Auto
`--strict`	Fail the run when generation produces warnings (for example a `$ref` that does not resolve); without it warnings are printed and generation completes	`false`

Examples:

# Generate everything from a local spec
corvusjson asyncapi-generate streetlights.json \
    --rootNamespace Streetlights.Client \
    --outputPath ./Generated

# Generate only producers for specific channels
corvusjson asyncapi-generate streetlights.json \
    --rootNamespace Streetlights.Publisher \
    --outputPath ./Generated \
    --mode producer \
    --include-channel "smartylighting.streetlights.*.turn.*"

# Fetch a remote spec and generate
corvusjson asyncapi-generate streetlights.json \
    --rootNamespace Streetlights.Client \
    --outputPath ./Generated \
    --spec-url https://api.example.com/asyncapi.json

Inspect Operations (`asyncapi-show`)

corvusjson asyncapi-show <specFile> [options]

Displays the channel/operation tree of an AsyncAPI specification:

Streetlights Kafka API v1.0.0 (AsyncAPI 3.0)

Operations (2)
├── smartylighting.streetlights.1.0.action.{streetlightId}.lighting.measured
│   └── RECV receiveLightMeasurement — Inform about environmental lighting conditions (1 msg, kafka-secure)
└── smartylighting.streetlights.1.0.action.{streetlightId}.turn.on
    └── SEND turnOn (1 msg, kafka-secure)

2 operations across 2 channels (1 send, 1 receive)

Supports the same filter options: --include-channel, --exclude-channel, --tag.

Lock File and Regeneration

The generator produces a corvusjson-asyncapi.lock file alongside the generated code. This file records:

The SHA-256 hash of the spec file
The generation options used
The timestamp of generation

On subsequent runs, the generator compares the current spec hash against the lock file. If the spec hasn't changed and --force is not specified, generation is skipped — enabling fast incremental builds.

# Normal run — skips if spec unchanged
corvusjson asyncapi-generate streetlights.json --rootNamespace Streetlights --outputPath ./Generated

# Force regeneration (e.g., after upgrading the tool)
corvusjson asyncapi-generate streetlights.json --rootNamespace Streetlights --outputPath ./Generated --force

When using --spec-url, the lock file also stores the URL. This enables update-style workflows where the spec is re-fetched from the original source on regeneration.

Version Support

AsyncAPI Version	Support
3.0.x	Full — recommended
2.6.x	Full — auto-detected and handled
2.5 and earlier	Not supported

The generator auto-detects the spec version from the asyncapi field. Use --specVersion to override if auto-detection fails.

Key Differences Between 2.6 and 3.0

Feature	2.6	3.0
Operations	Defined on channels (`publish`/`subscribe`)	Top-level with `action: send/receive`
Channel parameters	On channel object	On channel object
Security	On server	On server
Traits	Supported	Supported
Request/reply	`x-corvus-reply` extension	Standard operation `reply` object

The generator normalizes both formats into the same generated code structure — your producers and consumers look identical regardless of spec version.

Best Practices

Use `ValidationMode.Basic` in Production

The basic mode adds ~300ns per message — negligible overhead for catching malformed data that would otherwise cause runtime failures deep in your business logic. Reserve None for benchmarks or trusted internal pipelines.

Configure Dead-Letter Channels

Always configure an error policy that dead-letters unprocessable messages rather than silently dropping them. This gives you an audit trail and enables replay:

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

Use the In-Memory Transport for Tests

Don't mock IMessageTransport — use InMemoryMessageTransport. It provides message capture, delivery simulation, and dead-letter inspection without any broker infrastructure.

Wrap Transports with Instrumentation

Always wrap the transport with InstrumentedMessageTransport.Create in production. The zero-cost-when-idle design means no overhead until you attach an OpenTelemetry exporter, and Create preserves the wrapped transport's delivery-context and health-check capabilities.

Combine Polly with Error Policies

The resilience middleware (Polly) handles transient failures (retries, circuit-breaker). The error policy handles permanent failures (dead-letter, abort). They compose naturally:

Message arrives
  → Handler middleware (Polly): retry transient failures
  → If still failing: exception propagates
  → Error policy: dead-letter the message

Generate Both Sides from the Same Spec

When you control both publisher and subscriber, generate both from the same AsyncAPI spec. This guarantees:

Payload schemas match exactly
Channel addresses are consistent
Header contracts are synchronized
Validation catches drift at development time, not in production

Performance

All benchmarks use a zero-overhead stub transport (no real I/O) to isolate the pipeline cost — what you pay for serialization, validation, and dispatch logic above the transport layer. Since all approaches use the same underlying transport (NATS, Kafka, etc.), the transport cost is identical and cancels out; what differs is the per-message framework overhead.

Comparison:

Wolverine (baseline) — a popular .NET message bus framework (Mediator mode). Measures its full pipeline: STJ serialization/deserialization + handler chain resolution + compiled invoker dispatch + context pooling + handler property access. This represents what a real framework costs today.
Corvus (generated) — the generated AsyncAPI consumer/producer. Measures: pooled-memory parse → optional schema validation → typed dispatch through generated handler interface → error policy.

Both baselines include the serialization cost that a real transport would incur — Wolverine uses System.Text.Json POCO serialization, Corvus uses its generated workspace-based serialization.

Method	Mean	Allocated	Notes
Wolverine (baseline)	535 ns	104 B	STJ deserialize + handler chain + invoker
Corvus consumer (no validation)	160 ns	152 B	Pooled parse → typed dispatch → handler
Corvus consumer (basic validation)	320 ns	152 B	+ compiled schema check (boolean pass/fail)
Corvus consumer (detailed validation)	485 ns	152 B	+ full diagnostics with error locations
Corvus consumer with headers	464 ns	304 B	+ header parse & validate

Corvus without validation is 3.3× faster than Wolverine. Even with basic schema validation enabled, Corvus is 40% faster than Wolverine — which provides no validation at all.

Publish (Producer) Pipeline

Method	Mean	Allocated	Notes
Wolverine (baseline)	557 ns	80 B	STJ serialize + handler chain + compiled invoker + context pooling
Corvus producer (no validation)	293 ns	200 B	Workspace + channel build + serialize
Corvus producer (basic validation)	610 ns	200 B	+ compiled schema check
Corvus producer (detailed validation)	834 ns	200 B	+ full diagnostics collector

Corvus without validation is 1.9× faster than Wolverine. With basic validation enabled, Corvus is at parity with Wolverine's no-validation baseline. The 200B allocation is the short-lived JsonWorkspace the producer builds the message in, disposed when the publish completes.

Request/Reply

Method	Mean	Allocated	Notes
Wolverine + correlation header (baseline)	521 ns	968 B	STJ serialize + dispatch request with correlation, handler returns typed reply
Corvus req/reply (no validation)	377 ns	336 B	Generated producer: workspace + correlate + reply parse
Corvus req/reply (basic validation)	651 ns	336 B	+ schema validation both directions

Corvus without validation is 28% faster and allocates 65% less than the Wolverine baseline (336B vs 968B). The Corvus pipeline includes typed channel construction, schema-aware serialization, correlation ID matching (via pooled buffers), and reply parsing — all with aggressive allocation avoidance: the reply channel address is hoisted to a static field, the correlation ID is rented from ArrayPool<byte>, and the reply is parsed into documents owned by the caller-supplied workspace. With basic validation enabled, Corvus is 1.25× the baseline in time but still allocates 65% less — you get full schema conformance checking on both request and reply payloads for that cost.

Validation Cost Summary

Mode	Overhead	Additional allocation
None	—	0 B
Basic (boolean pass/fail)	~300 ns	0 B
Detailed (error locations)	~540 ns	0 B

All validation modes produce zero additional allocation — the schema evaluator operates entirely on the already-parsed document. This means you can enable validation in production without increasing GC pressure.

BenchmarkDotNet v0.15.8, .NET 10.0.8, 13th Gen Intel Core i7-13800H, Windows 11. OutlierMode=RemoveAll, RunStrategy=Throughput. Measured against the V5.3.0 pipeline; the V5.4.0 lifecycle and request/reply hardening changed the measured paths (notably the requester now threads the caller-supplied workspace), so treat exact figures as indicative of that baseline.

Example Recipes

AsyncAPI Producer — Basic producer generation and publishing
AsyncAPI Consumer — Consumer with validation and error handling
AsyncAPI End-to-End — Producer + consumer with in-memory transport testing
AsyncAPI Authentication — All auth patterns (Azure Identity, OAuth2, API Key, mTLS, composite)