At endjin, we maintain Corvus.JsonSchema, and in the previous post we looked at testing patterns with the in-memory transport.
Throughout this series, we've used transports from the built-in packages - NATS, Kafka, AMQP, MQTT, Azure Service Bus, and the in-memory transport for testing. But what if your organisation uses a broker that isn't on that list? Perhaps you're running Redis Streams, Amazon SQS, Google Cloud Pub/Sub, or a proprietary internal messaging system. The transport layer is designed to be extended.
The interface contract
The entire transport abstraction is a single interface with four methods:
public interface IMessageTransport : IAsyncDisposable
{
ValueTask PublishAsync<TPayload>(
ReadOnlyMemory<byte> channelUtf8,
in TPayload payload,
in JsonElement headers = default,
CancellationToken cancellationToken = default)
where TPayload : struct, IJsonElement<TPayload>;
ValueTask SubscribeAsync<TPayload>(
ReadOnlyMemory<byte> channelUtf8,
Func<TPayload, JsonElement, CancellationToken, ValueTask> handler,
CancellationToken cancellationToken = default)
where TPayload : struct, IJsonElement<TPayload>;
ValueTask UnsubscribeAsync(
ReadOnlyMemory<byte> channelUtf8,
CancellationToken cancellationToken = default);
ValueTask<(TReply Payload, JsonElement Headers)> RequestAsync<TRequest, TReply>(
ReadOnlyMemory<byte> requestChannelUtf8,
ReadOnlyMemory<byte> replyChannelUtf8,
TRequest request,
ReadOnlyMemory<byte> correlationIdUtf8,
JsonElement headers = default,
CancellationToken cancellationToken = default)
where TRequest : struct, IJsonElement<TRequest>
where TReply : struct, IJsonElement<TReply>;
ValueTask DeadLetterAsync(
ReadOnlyMemory<byte> deadLetterChannelUtf8,
ReadOnlyMemory<byte> originalChannelUtf8,
in JsonElement payload,
in JsonElement headers,
Exception exception,
CancellationToken cancellationToken = default);
}
That's it. The generated producers call PublishAsync and RequestAsync. The generated consumers call SubscribeAsync, UnsubscribeAsync, and DeadLetterAsync. Everything else - schema validation, channel address construction, error policies, handler middleware - lives in the generated code and the runtime library, not in the transport.
What the transport is responsible for
The transport is a low-level messaging pipe. Its responsibilities are deliberately narrow:
For publishing: take the typed payload, serialize it to bytes (via WriteTo(Utf8JsonWriter)), and deliver those bytes to the broker on the specified channel. The channel address arrives as UTF-8 bytes in a ReadOnlyMemory<byte>. If your broker's client library needs a string, convert at the outermost boundary with Encoding.UTF8.GetString().
For subscribing: register with the broker to receive messages on the specified channel pattern, parse incoming bytes into a typed payload (via ParsedJsonDocument<T>.Parse()), and call the handler delegate with the parsed payload and any message headers as a JsonElement.
For request/reply: combine publish and subscribe. Send the request, listen for a correlated reply on the appropriate channel, and return the typed response. If your broker doesn't natively support request/reply, you can implement it with a temporary subscription filtered by correlation ID (the same approach the NATS core transport uses).
For unsubscribe: stop receiving messages on the specified channel and clean up any broker-side subscription state.
A skeleton implementation
Here's the structure of a custom transport. I'll use Redis Streams as an illustrative example, but the pattern applies to any broker:
using System.Text;
using System.Text.Json;
using Corvus.Text.Json;
using Corvus.Text.Json.AsyncApi;
public sealed class RedisStreamTransport : IMessageTransport, IAsyncDisposable
{
private readonly ConnectionMultiplexer _redis;
private readonly ConcurrentDictionary<string, CancellationTokenSource> _subscriptions = new();
public RedisStreamTransport(ConnectionMultiplexer redis)
{
_redis = redis;
}
public async ValueTask PublishAsync<TPayload>(
ReadOnlyMemory<byte> channelUtf8,
TPayload payload,
JsonElement headers = default,
CancellationToken cancellationToken = default)
where TPayload : struct, IJsonElement<TPayload>
{
// 1. Serialize the payload to bytes
using var buffer = new ArrayBufferWriter<byte>();
using var writer = new Utf8JsonWriter(buffer);
payload.WriteTo(writer);
await writer.FlushAsync(cancellationToken);
// 2. Convert channel to string (at the boundary)
string channel = Encoding.UTF8.GetString(channelUtf8.Span);
// 3. Deliver to your broker
var db = _redis.GetDatabase();
await db.StreamAddAsync(
channel,
[new NameValueEntry("payload", buffer.WrittenMemory.ToArray())],
flags: CommandFlags.FireAndForget);
}
public async ValueTask SubscribeAsync<TPayload>(
ReadOnlyMemory<byte> channelUtf8,
Func<TPayload, JsonElement, CancellationToken, ValueTask> handler,
CancellationToken cancellationToken = default)
where TPayload : struct, IJsonElement<TPayload>
{
string channel = Encoding.UTF8.GetString(channelUtf8.Span);
var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
_subscriptions[channel] = cts;
// Start a background loop reading from the stream
_ = Task.Run(async () =>
{
var db = _redis.GetDatabase();
string lastId = "0-0";
while (!cts.Token.IsCancellationRequested)
{
var entries = await db.StreamReadAsync(channel, lastId, count: 10);
foreach (var entry in entries)
{
lastId = entry.Id!;
byte[] payloadBytes = (byte[])entry["payload"]!;
// Parse into the typed payload
using var doc = ParsedJsonDocument<TPayload>.Parse(payloadBytes);
TPayload typed = doc.RootElement;
// Call the handler (headers as empty JsonElement if none)
await handler(typed, default, cts.Token);
}
if (entries.Length == 0)
{
await Task.Delay(100, cts.Token);
}
}
}, cts.Token);
}
public ValueTask UnsubscribeAsync(
ReadOnlyMemory<byte> channelUtf8,
CancellationToken cancellationToken = default)
{
string channel = Encoding.UTF8.GetString(channelUtf8.Span);
if (_subscriptions.TryRemove(channel, out var cts))
{
cts.Cancel();
cts.Dispose();
}
return ValueTask.CompletedTask;
}
public ValueTask<(TReply Payload, JsonElement Headers)> RequestAsync<TRequest, TReply>(
ReadOnlyMemory<byte> requestChannelUtf8,
ReadOnlyMemory<byte> replyChannelUtf8,
TRequest request,
ReadOnlyMemory<byte> correlationIdUtf8,
JsonElement headers = default,
CancellationToken cancellationToken = default)
where TRequest : struct, IJsonElement<TRequest>
where TReply : struct, IJsonElement<TReply>
{
// Implement correlation-based request/reply
// or throw NotSupportedException if your broker doesn't support it
throw new NotSupportedException(
"Redis Streams does not natively support request/reply. " +
"Use produce/consume patterns instead.");
}
public async ValueTask DeadLetterAsync(
ReadOnlyMemory<byte> deadLetterChannelUtf8,
ReadOnlyMemory<byte> originalChannelUtf8,
JsonElement payload,
JsonElement headers,
Exception exception,
CancellationToken cancellationToken = default)
{
string dlqChannel = Encoding.UTF8.GetString(deadLetterChannelUtf8.Span);
string originalChannel = Encoding.UTF8.GetString(originalChannelUtf8.Span);
var db = _redis.GetDatabase();
await db.StreamAddAsync(
dlqChannel,
[
new NameValueEntry("payload", JsonSerializer.SerializeToUtf8Bytes(payload)),
new NameValueEntry("originalChannel", originalChannel),
new NameValueEntry("error", exception.Message),
new NameValueEntry("timestamp", DateTimeOffset.UtcNow.ToString("O")),
]);
}
public async ValueTask DisposeAsync()
{
foreach (var cts in _subscriptions.Values)
{
cts.Cancel();
cts.Dispose();
}
_subscriptions.Clear();
}
}
This is intentionally simplified. A production implementation would handle consumer groups, acknowledgement, error recovery, and connection management. But it illustrates the core contract: serialize on publish, parse on subscribe, dead-letter on failure, and let the generated code handle everything else.
Key implementation considerations
Channel addresses are UTF-8 bytes
The ReadOnlyMemory<byte> channel parameter contains a UTF-8 encoded channel address that the generated code has already resolved (substituting any channel parameters like {streetlightId} with actual values). Your transport receives the final address. If your broker API accepts strings, convert once at the boundary. If it accepts byte spans natively (as NATS does), you can avoid the string allocation entirely.
Payload serialization uses WriteTo
The TPayload type constraint (IJsonElement<TPayload>) guarantees that the payload has a WriteTo(Utf8JsonWriter) method. This is the canonical way to serialize. Write to a Utf8JsonWriter backed by whatever buffer strategy suits your transport. The generated types write directly from their internal pooled representation, so there's no intermediate string or byte array.
Payload parsing uses ParsedJsonDocument
On the subscribe side, incoming bytes from the broker are parsed via ParsedJsonDocument<T>.Parse(). This gives you a typed view over the JSON with the same pooled-memory semantics as the rest of V5. The handler receives the parsed payload and any message headers (as a JsonElement - pass default if your broker doesn't have a header concept).
MessageContext carries bindings
The MessageContext parameter on PublishAsync contains channel, operation, and message bindings as raw JSON bytes. These come from the AsyncAPI spec's bindings section. If your broker uses specific metadata (partition keys, routing keys, priority levels), inspect the relevant binding JSON in your transport. If your broker doesn't use bindings, you can safely ignore the context.
Dead-letter routing
DeadLetterAsync is a first-class method on IMessageTransport. The generated consumer calls it directly when the error policy returns MessageErrorAction.DeadLetter. Your implementation receives the dead-letter channel address (already computed by the generated consumer), the original channel, the raw payload and headers, and the exception that caused the failure:
public async ValueTask DeadLetterAsync(
ReadOnlyMemory<byte> deadLetterChannelUtf8,
ReadOnlyMemory<byte> originalChannelUtf8,
JsonElement payload,
JsonElement headers,
Exception exception,
CancellationToken cancellationToken = default)
{
string dlqChannel = Encoding.UTF8.GetString(deadLetterChannelUtf8.Span);
string originalChannel = Encoding.UTF8.GetString(originalChannelUtf8.Span);
var db = _redis.GetDatabase();
// Store with error metadata for debugging
await db.StreamAddAsync(
dlqChannel,
[
new NameValueEntry("payload", JsonSerializer.SerializeToUtf8Bytes(payload)),
new NameValueEntry("originalChannel", originalChannel),
new NameValueEntry("error", exception.Message),
new NameValueEntry("errorType", exception.GetType().Name),
new NameValueEntry("timestamp", DateTimeOffset.UtcNow.ToString("O")),
]);
}
The advantage of having DeadLetterAsync as a dedicated method (rather than just re-publishing) is that your transport can attach error metadata alongside the failed message. This makes dead-letter queues much easier to inspect and debug. If your broker has a native dead-letter mechanism (as RabbitMQ does with dead-letter exchanges, or Azure Service Bus does with dead-letter subqueues), your implementation can delegate to that native facility and preserve any broker-specific metadata it provides.
Implementing request/reply
The RequestAsync method on IMessageTransport is optional in the sense that not all brokers support it naturally. If your broker has a native request/reply mechanism (as NATS does with its inbox pattern), implement it directly. If it doesn't, you have two choices.
The first is to throw NotSupportedException, which means your transport can't be used with AsyncAPI operations that define a reply object. This is honest and clear. Your transport supports produce/consume but not request/reply.
The second is to implement correlation-based request/reply yourself, using a temporary subscription. Notice that the interface design helps here. The generated producer code handles correlation ID generation (formatting a GUID directly into a pooled byte[36]) and passes the correlation ID, request channel, and reply channel as separate parameters. Your transport just needs to wire them together:
- Subscribe to the reply channel, filtered by the correlation ID
- Publish the request to the request channel
- Wait for the reply to arrive (with timeout via the cancellation token)
- Unsubscribe from the reply channel
- Return the typed reply
Here's a sketch:
public async ValueTask<(TReply Payload, JsonElement Headers)> RequestAsync<TRequest, TReply>(
ReadOnlyMemory<byte> requestChannelUtf8,
ReadOnlyMemory<byte> replyChannelUtf8,
TRequest request,
ReadOnlyMemory<byte> correlationIdUtf8,
JsonElement headers = default,
CancellationToken cancellationToken = default)
where TRequest : struct, IJsonElement<TRequest>
where TReply : struct, IJsonElement<TReply>
{
var tcs = new TaskCompletionSource<(TReply, JsonElement)>();
// Subscribe to the reply channel, matching on correlation ID
await SubscribeAsync<TReply>(
replyChannelUtf8,
(reply, replyHeaders, ct) =>
{
// Check correlation ID in headers matches ours
if (CorrelationMatches(replyHeaders, correlationIdUtf8.Span))
{
tcs.TrySetResult((reply, replyHeaders));
}
return ValueTask.CompletedTask;
},
cancellationToken);
// Publish the request
await PublishAsync(requestChannelUtf8, in request, in headers, cancellationToken);
// Await the reply (cancellation token handles timeout)
var result = await tcs.Task.WaitAsync(cancellationToken);
await UnsubscribeAsync(replyChannelUtf8, cancellationToken);
return result;
}
This is more complex than the other methods, and it's why the built-in transports handle it for you. But if you need request/reply over a broker that doesn't support it natively, the pattern is well-established and the correlation matching is straightforward.
Acknowledgement and at-least-once delivery
If your broker supports acknowledgement (most durable brokers do), your SubscribeAsync implementation should acknowledge messages after the handler delegate completes successfully. If the handler throws, don't acknowledge. Let the broker redeliver.
The simplest approach is to wrap the handler call:
try
{
await handler(typed, headers, cancellationToken);
await AcknowledgeAsync(message); // broker-specific ack
}
catch
{
// Don't ack - broker will redeliver after visibility timeout
throw; // Let the generated consumer's error policy handle it
}
The generated consumer catches this exception and routes it through the IMessageErrorPolicy. If the policy returns Skip, the consumer swallows the error and moves on (you should then ack the message). If it returns DeadLetter, the consumer dead-letters and then acks. If it returns Abort, the consumer stops without acknowledging, so the message will be redelivered when the consumer restarts.
To support this fully, your subscribe loop needs to distinguish between "handler succeeded" and "error policy resolved the failure." The cleanest approach is to not ack inside SubscribeAsync at all, but instead accept an Action<MessageAcknowledgement> callback from the generated consumer. The built-in transports use this pattern internally. Look at the AMQP or Kafka transport source for reference.
Adding health check support
If you want your custom transport to work with the ASP.NET Core health check integration from Corvus.Text.Json.AsyncApi.HealthChecks, implement IHealthCheckableTransport:
public sealed class RedisStreamTransport
: IMessageTransport, IHealthCheckableTransport, IAsyncDisposable
{
public bool IsConnected => _redis.IsConnected;
public string MessagingSystem => "redis-streams";
public async ValueTask<bool> PingAsync(CancellationToken cancellationToken = default)
{
try
{
var db = _redis.GetDatabase();
await db.PingAsync();
return true;
}
catch
{
return false;
}
}
}
The health check extension calls IsConnected for a quick check and PingAsync for an active probe. MessagingSystem is a label that appears in health check results.
You don't need to implement telemetry in your transport. The InstrumentedMessageTransport decorator works with any IMessageTransport:
IMessageTransport raw = new RedisStreamTransport(redis);
IMessageTransport transport = new InstrumentedMessageTransport(raw, "redis-streams");
This gives you distributed tracing, metrics, and W3C trace context propagation. All of this comes from the decorator, with no changes to your transport implementation.
Publishing transport-specific telemetry
The decorator handles the standard messaging telemetry (messages sent, messages consumed, processing duration, dead-letters). But your transport often has its own operational metrics that are worth surfacing - connection state transitions, reconnection attempts, broker-specific error codes, queue depth, or partition rebalancing events.
The AsyncApiTelemetry static class exposes the shared ActivitySource and Meter along with convenience recording methods for common transport events:
using Corvus.Text.Json.AsyncApi;
// Record a transport state transition (connected ↔ disconnected)
AsyncApiTelemetry.TransportStateTransitions.Add(
1,
new KeyValuePair<string, object?>("messaging.system", "redis-streams"),
new KeyValuePair<string, object?>("state", "connected"));
// Record a dead-letter that happened inside your transport's error handling
AsyncApiTelemetry.RecordDeadLetter(
channel: "orders.created",
messagingSystem: "redis-streams",
reason: "deserialization_failure");
// Record a dead-letter that itself failed (message was dropped)
AsyncApiTelemetry.RecordDeadLetterFailure(
channel: "orders.created",
messagingSystem: "redis-streams",
reason: "dlq_full",
exception: ex);
For transport-specific metrics that don't fit the standard model, create your own Meter in your transport package and document it for users to subscribe to:
public sealed class RedisStreamTransport : IMessageTransport, IAsyncDisposable
{
private static readonly Meter TransportMeter = new("Corvus.AsyncApi.RedisStreams");
private static readonly Counter<long> ReconnectionAttempts =
TransportMeter.CreateCounter<long>(
"corvus.asyncapi.redis.reconnection_attempts",
description: "Number of reconnection attempts to Redis");
private static readonly Histogram<double> StreamLag =
TransportMeter.CreateHistogram<double>(
"corvus.asyncapi.redis.stream_lag_ms",
unit: "ms",
description: "Lag between last delivered and last pending message");
// Use in your implementation:
private async Task ReconnectAsync()
{
ReconnectionAttempts.Add(1,
new KeyValuePair<string, object?>("server", _serverAddress));
// ... reconnection logic
}
}
Users then subscribe to both the standard meter and your transport meter:
services.AddOpenTelemetry()
.WithMetrics(b => b
.AddMeter(AsyncApiTelemetry.MeterName) // standard
.AddMeter("Corvus.AsyncApi.RedisStreams")); // transport-specific
This approach keeps transport-specific telemetry separate from the standard messaging metrics, while following the same zero-cost-when-idle pattern. If nobody subscribes to your transport meter, the counters and histograms are no-ops.
Implementing ITransportOptions
The ITransportOptions interface provides the shared configuration contract that the generated consumer code expects. By implementing it on your options class, you allow users to configure error policies, handler middleware, and heartbeat monitoring consistently with the built-in transports:
public sealed class RedisStreamTransportOptions : ITransportOptions
{
// Transport-specific configuration
public string ConnectionString { get; set; } = "localhost:6379";
public string ConsumerGroup { get; set; } = "default";
public string ConsumerName { get; set; } = Environment.MachineName;
public int BatchSize { get; set; } = 10;
public TimeSpan PollInterval { get; set; } = TimeSpan.FromMilliseconds(100);
public TimeSpan ClaimTimeout { get; set; } = TimeSpan.FromMinutes(5);
public string DeadLetterSuffix { get; set; } = "dlq:";
// ITransportOptions - shared resilience configuration
public IMessageErrorPolicy? ErrorPolicy { get; set; }
public MessageHandlerMiddleware? HandlerMiddleware { get; set; }
public ProcessingLoopHeartbeat? Heartbeat { get; set; }
}
The three ITransportOptions properties serve specific roles:
ErrorPolicy is the IMessageErrorPolicy instance that determines what happens when a message permanently fails processing. Your transport passes this to the generated consumer at construction time. If not set, the consumer uses a sensible default (dead-letter on deserialization and handler errors, abort on transport errors).
HandlerMiddleware is a MessageHandlerMiddleware delegate that wraps every handler invocation. This is where Polly resilience pipelines plug in. Your transport passes it to the generated consumer, which calls through it for every message.
Heartbeat is the ProcessingLoopHeartbeat tracker. If provided, your subscribe loop should call heartbeat.Tick(channel) on every iteration so the liveness monitor can detect stalled consumers.
Here's how your transport uses these in the subscribe loop:
public async ValueTask SubscribeAsync<TPayload>(
ReadOnlyMemory<byte> channelUtf8,
Func<TPayload, JsonElement, CancellationToken, ValueTask> handler,
CancellationToken cancellationToken = default)
where TPayload : struct, IJsonElement<TPayload>
{
string channel = Encoding.UTF8.GetString(channelUtf8.Span);
_ = Task.Run(async () =>
{
while (!cancellationToken.IsCancellationRequested)
{
// Tick the heartbeat so liveness monitoring knows we're alive
_options.Heartbeat?.Tick(channel);
var entries = await ReadFromStreamAsync(channel, _options.BatchSize);
foreach (var entry in entries)
{
using var doc = ParsedJsonDocument<TPayload>.Parse(entry.PayloadBytes);
// If middleware is configured, invoke through it
if (_options.HandlerMiddleware is { } middleware)
{
await middleware(
async ct => await handler(doc.RootElement, entry.Headers, ct),
cancellationToken);
}
else
{
await handler(doc.RootElement, entry.Headers, cancellationToken);
}
}
if (entries.Length == 0)
{
await Task.Delay(_options.PollInterval, cancellationToken);
}
}
}, cancellationToken);
}
The key points: tick the heartbeat on every loop iteration, wrap handler calls through the middleware delegate if one is provided, and let exceptions propagate to the generated consumer's error policy.
Packaging conventions
If you're building a transport for others to consume, follow the naming convention of the existing packages:
- Package:
Corvus.Text.Json.AsyncApi.YourBroker
- Transport class:
YourBrokerMessageTransport
- Options class:
YourBrokerTransportOptions : ITransportOptions
The ITransportOptions interface carries the shared ErrorPolicy and HandlerMiddleware properties, so consumers can configure error handling and resilience consistently regardless of transport.
When to build a custom transport
The built-in transports cover the most common brokers. You'd implement a custom transport when:
- Your organisation uses a broker not covered by the existing packages (Redis Streams, Amazon SQS/SNS, Google Cloud Pub/Sub, Apache Pulsar, ZeroMQ)
- You need to integrate with a proprietary internal messaging system
- You want a specialised transport for a specific deployment constraint (perhaps an embedded broker for edge devices, or a file-based transport for offline-first scenarios)
The interface is deliberately minimal - five methods plus disposal - so a basic implementation is straightforward. The complexity lives in making it production-ready: connection management, reconnection, acknowledgement semantics, and graceful shutdown. But those are broker-specific concerns that you'd need to solve regardless of whether you're using Corvus or writing raw messaging code.
For the complete IMessageTransport API reference, see the API documentation on corvus-oss.org.
Log in
:::
