AsyncAPI Code Generation with Corvus: Typed Producers
At endjin, we maintain Corvus.JsonSchema, and in the OpenAPI series we applied V5's code generation engine to HTTP APIs. Now let's do the same for event-driven messaging.
The messaging serialization problem
If you've built a producer for Kafka or NATS or Service Bus, you've probably written the same boilerplate many times. You serialize a payload to JSON, construct a channel or topic name from a template, set the right headers, publish, and hope the consumer on the other end agrees about the shape of the message.
The problem is that this agreement is informal. The producer serializes a TurnOnOffCommand class. The consumer deserializes into its own copy of what it thinks that class looks like. If someone adds a field on one side but not the other, nothing fails at compile time. You find out at runtime. Worse, the consumer silently ignores the extra field and operates on incomplete data.
AsyncAPI specifications exist to make this contract explicit, in the same way OpenAPI does for HTTP. They describe channels, message payloads with JSON Schema, channel parameters, and security schemes. But until now, there hasn't been a .NET code generator that enforces that contract with schema validation and typed models.
That's what corvusjson asyncapi-generate provides: typed producer classes that validate payloads against their schema before publishing, with the same zero-allocation models and pooled memory from the rest of V5.
Getting started
The full reference documentation is on the Corvus.JsonSchema AsyncAPI guide, and there's an interactive AsyncAPI playground where you can paste a spec and see the generated code. This post focuses on the producer side and the design choices behind it.
dotnet tool install --global Corvus.Json.Cli
dotnet add package Corvus.Text.Json.AsyncApi
dotnet add package Corvus.Text.Json
Then add a transport package for your broker. All transports implement the same IMessageTransport interface, so your producer code doesn't change when you switch brokers:
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
Generating a producer
Given a Streetlights AsyncAPI spec (the canonical AsyncAPI example, equivalent to Petstore for OpenAPI):
corvusjson asyncapi-generate streetlights.json \
--rootNamespace Streetlights.Client \
--outputPath ./Generated \
--mode producer
The generator reads the spec's send operations (or subscribe in AsyncAPI 2.6) and produces a typed producer class, message metadata types, and model types in a .Models sub-namespace. It also writes a lock file that tracks the spec hash for incremental regeneration. If your spec hasn't changed, the next generation run is a no-op.
Publishing with type safety
Here's what using the generated producer looks like. We'll use InMemoryMessageTransport so the example is self-contained, but the API is identical with any real broker:
using System.Text;
using Corvus.Text.Json.AsyncApi;
using Corvus.Text.Json.AsyncApi.Testing;
using Streetlights.Client;
using Streetlights.Client.Models;
await using InMemoryMessageTransport transport = new();
TurnOnProducer producer = new(transport, ValidationMode.Basic);
await producer.PublishTurnOnOffAsync(
payload: new TurnOnOffPayload.Source((ref TurnOnOffPayload.Builder b) =>
{
b.Create(command: "on"u8, sentAt: DateTimeOffset.UtcNow);
}),
streetlightId: "lamp-42");
PublishedMessage msg = transport.PublishedMessages[0];
Console.WriteLine($"Channel: {msg.Channel}");
Console.WriteLine($"Payload: {Encoding.UTF8.GetString(msg.PayloadBytes)}");
A few things to notice here.
The streetlightId parameter comes from the channel address template in the spec (smartylighting.streetlights.1.0.action.{streetlightId}.turn.on). The generated code constructs the full channel address using zero-allocation UTF-8 byte manipulation with pooled buffers, avoiding both string concatenation and intermediate allocations. You pass the parameter as a typed argument, and the generator handles the rest.
The payload uses the same Source and Builder pattern from the rest of V5 (if you're not familiar with this approach to constructing JSON objects without allocations, the Corvus.JsonSchema documentation covers it in detail). Required properties are mandatory; optional ones have defaults. The schema says command must be "on" or "off". With validation enabled, the producer checks that before the message leaves your process.
Validation happens before the wire
This is the same philosophy as the OpenAPI client: catch contract violations immediately, with a clear exception, rather than letting a malformed message propagate through your system.
TurnOnProducer producer = new(transport, ValidationMode.Basic);
try
{
await producer.PublishTurnOnOffAsync(
payload: new TurnOnOffPayload.Source((ref TurnOnOffPayload.Builder b) =>
{
b.Create(command: "invalid-command"u8, sentAt: DateTimeOffset.UtcNow);
}),
streetlightId: "lamp-001");
}
catch (ArgumentException ex)
{
// "Message payload validation failed for 'payload'."
Console.WriteLine(ex.Message);
}
In Basic mode, validation is a fast boolean schema check. In Detailed mode, you get full evaluation diagnostics with JSON Pointer locations - useful during development. In None mode, validation is skipped entirely for maximum throughput on trusted internal services where you control both ends.
Authentication
When your AsyncAPI spec defines security schemes, pass an authentication provider to the producer. The generated code calls AuthenticateAsync before each publish, so credentials are attached consistently without you wiring it up per-message:
IMessageAuthenticationProvider auth = new UserPasswordAuthenticationProvider(
username: "service-account",
password: "kafka-secret");
TurnOnProducer authenticatedProducer = new(transport, ValidationMode.Basic, authProvider: auth);
Swapping transports
Because all transports implement IMessageTransport, switching from the in-memory test transport to a real broker is a one-line change. Your producer code, payload construction, and validation behaviour stay exactly the same:
// NATS
await using NatsMessageTransport transport = await NatsMessageTransport.CreateAsync(new()
{
Url = "nats://broker.example.com:4222",
});
// Kafka
await using KafkaMessageTransport transport = new(new()
{
BootstrapServers = "kafka.example.com:9092",
GroupId = "streetlights-producer",
});
// Azure Service Bus
await using AzureServiceBusMessageTransport transport =
await AzureServiceBusMessageTransport.CreateAsync(new()
{
ConnectionString = "<connection-string>",
QueueName = "streetlights",
});
All transports implement IMessageTransport. Your producer code doesn't change.
What's next
In the [ref slug=asyncapi-code-generation-with-corvus-typed-consumers text=next post], we'll look at the consumer side - implementing handler interfaces, error policies, and dead-letter routing for reliable message processing.
