Customize persistent-stream data formats

A persistent-stream data adapter translates between Orleans event batches and a provider’s native queue messages. It owns the wire contract: stream identity, event type and payload encoding, schema version, and any request-context values which cross the transport boundary. The provider’s queue adapter continues to own transport access, partition assignment, receive acknowledgement, caching, and recovery.

Use a data adapter when the transport remains the same and the application needs a versioned payload format, needs to consume messages produced outside Orleans, or needs to publish messages which an external consumer can decode. Use a custom queue adapter when Orleans also needs a new transport or different queue semantics.

Choose the provider extension point

The built-in providers expose these data-adapter extension points:

Provider	Data-adapter contract	Registration	Runtime outcome
Azure Queue Storage	IQueueDataAdapter<T, U> with string queue messages and IBatchContainer batches	ConfigureQueueDataAdapter	Replaces encoding and decoding while retaining Azure Queue mapping, visibility, deletion, and non-rewindable delivery
Azure Event Hubs	IEventHubDataAdapter	UseDataAdapter	Replaces wire-format, stream-mapping, and cache conversion behavior while retaining Event Hubs partition reading, checkpointing, and rewindable delivery

IQueueDataAdapter<T> defines conversion from an Orleans batch to a native queue message. IQueueDataAdapter<T, U> adds conversion from a native message to the batch container delivered by Orleans. Provider-specific contracts can add the position and cache operations required by their transport.

Define a versioned contract

Define the transport contract independently of CLR assembly names and Orleans serialization internals. A durable envelope normally contains:

• a schema version;
• the stream namespace and key;
• a stable event-type identifier;
• one or more event payloads; and
• an explicit set of cross-process metadata.

Assign each message to exactly one StreamId. Keep the mapping from stream ID to physical partition stable so events for a stream retain the provider’s ordering behavior. The adapter receives batches, so the envelope must preserve event order within each batch.

Treat request context as an explicit contract. The example carries a string correlation ID and leaves process-local values inside the producing application.

Implement an Azure Queue data adapter

The following adapter writes a versioned JSON envelope and reconstructs an IBatchContainer when Azure Queue Storage returns the message. The Azure Queue receiver supplies sequenceId at read time, and the batch uses it to create a receiver-local sequence token for each event.

public sealed class JsonAzureQueueDataAdapter :
    IQueueDataAdapter<string, IBatchContainer>
{
    public string ToQueueMessage<T>(
        StreamId streamId,
        IEnumerable<T> events,
        StreamSequenceToken? token,
        Dictionary<string, object>? requestContext)
    {
        if (token is not null)
        {
            throw new ArgumentException(
                "The Azure Queue stream provider assigns sequence positions when messages are read.",
                nameof(token));
        }

        return ExternalEventContract.Serialize(
            streamId,
            events,
            requestContext);
    }

    public IBatchContainer FromQueueMessage(
        string queueMessage,
        long sequenceId)
    {
        var envelope = ExternalEventContract.Deserialize(queueMessage);
        return new JsonAzureQueueBatch(
            StreamId.Create(envelope.StreamNamespace, envelope.StreamKey),
            envelope.Events,
            new EventSequenceTokenV2(sequenceId),
            envelope.CorrelationId);
    }
}

The batch container exposes only events compatible with the requested stream type, derives per-event tokens from the queue-message sequence, and imports the metadata defined by the wire contract:

[GenerateSerializer, Immutable]
public sealed class JsonAzureQueueBatch : IBatchContainer
{
    [Id(0)]
    private readonly DeviceReading[] _events;

    [Id(1)]
    private readonly EventSequenceTokenV2 _sequenceToken;

    [Id(2)]
    private readonly string? _correlationId;

    public JsonAzureQueueBatch(
        StreamId streamId,
        DeviceReading[] events,
        EventSequenceTokenV2 sequenceToken,
        string? correlationId)
    {
        StreamId = streamId;
        _events = events;
        _sequenceToken = sequenceToken;
        _correlationId = correlationId;
    }

    [Id(3)]
    public StreamId StreamId { get; }

    public StreamSequenceToken SequenceToken => _sequenceToken;

    public IEnumerable<Tuple<T, StreamSequenceToken>> GetEvents<T>()
    {
        if (typeof(T) != typeof(DeviceReading))
        {
            return [];
        }

        return _events.Select(
            (item, index) => Tuple.Create(
                (T)(object)item,
                (StreamSequenceToken)_sequenceToken.CreateSequenceTokenForEvent(index)));
    }

    public bool ImportRequestContext()
    {
        if (_correlationId is null)
        {
            return false;
        }

        RequestContext.Set(
            ExternalEventContract.CorrelationIdKey,
            _correlationId);
        return true;
    }
}

Register the same adapter, queue service client, and physical queue names for the provider on every silo and Orleans client which uses it. Silos use the adapter for reads and writes; clients use it when publishing.

public static void ConfigureAzureQueueSilo(
    ISiloBuilder builder,
    QueueServiceClient queueServiceClient)
{
    builder.AddAzureQueueStreams(
        ProviderName,
        (SiloAzureQueueStreamConfigurator streams) =>
        {
            streams.ConfigureAzureQueue(options => options.Configure(value =>
            {
                value.QueueServiceClient = queueServiceClient;
                value.QueueNames = ["external-events-0"];
            }));
            streams.ConfigureQueueDataAdapter<JsonAzureQueueDataAdapter>();
        });
}

public static void ConfigureAzureQueueClient(
    IClientBuilder builder,
    QueueServiceClient queueServiceClient)
{
    builder.AddAzureQueueStreams(
        ProviderName,
        (ClusterClientAzureQueueStreamConfigurator streams) =>
        {
            streams.ConfigureAzureQueue(options => options.Configure(value =>
            {
                value.QueueServiceClient = queueServiceClient;
                value.QueueNames = ["external-events-0"];
            }));
            streams.ConfigureQueueDataAdapter<JsonAzureQueueDataAdapter>();
        });
}

Configure durable PubSubStore grain storage alongside this registration as described in Orleans stream providers.

Implement an Event Hubs data adapter

For Event Hubs, derive from EventHubDataAdapter when its cached-message representation and checkpoint behavior fit the application. Override:

• GetStreamIdentity to map each EventData instance to a stream;
• GetBatchContainer to decode cached payloads for Orleans consumers;
• ToQueueMessage to encode events published through Orleans; and
• GetPartitionKey to select the physical Event Hubs partition key.

The adapter also participates in cache conversion and sequence positioning through IEventHubDataAdapter. Preserve the Event Hubs offset and sequence number when constructing batch tokens so checkpoint and rewind behavior remains aligned with the partition log.

Register the adapter and Event Hubs connection under the same provider name on silos and publishing clients. The silo registration also configures durable Azure Table checkpoints:

public static void ConfigureEventHubSilo(
    ISiloBuilder builder,
    string connectionString,
    string eventHubName,
    string consumerGroup,
    TableServiceClient checkpointStore)
{
    builder.AddEventHubStreams(
        ProviderName,
        (ISiloEventHubStreamConfigurator streams) =>
        {
            streams.ConfigureEventHub(options => options.Configure(value =>
                value.ConfigureEventHubConnection(
                    connectionString,
                    eventHubName,
                    consumerGroup)));
            streams.UseAzureTableCheckpointer(
                options => options.Configure(
                    value => value.TableServiceClient = checkpointStore));
            streams.UseDataAdapter(
                (services, _) =>
                    ActivatorUtilities.CreateInstance<CustomEventHubDataAdapter>(
                        services));
        });
}

public static void ConfigureEventHubClient(
    IClientBuilder builder,
    string connectionString,
    string eventHubName,
    string consumerGroup)
{
    builder.AddEventHubStreams(
        ProviderName,
        (IClusterClientEventHubStreamConfigurator streams) =>
        {
            streams.ConfigureEventHub(options => options.Configure(value =>
                value.ConfigureEventHubConnection(
                    connectionString,
                    eventHubName,
                    consumerGroup)));
            streams.UseDataAdapter(
                (services, _) =>
                    ActivatorUtilities.CreateInstance<CustomEventHubDataAdapter>(
                        services));
        });
}

The custom data adapter sample demonstrates a read-side Event Hubs adapter for JSON messages from an external producer. See Integrate external stream producers and consumers for the end-to-end interoperability workflow.

Evolve the wire contract

Use an expand-and-contract rollout:

1. Deploy readers which accept the current and next schema versions.
2. Change writers to emit the next version.
3. Keep the old reader until the queue or event-log retention window no longer contains the old version.
4. Remove the old version after verifying queue depth, oldest-message age, and checkpoint position.

Keep provider name, stream identity encoding, partition mapping, and sequence interpretation stable during a payload-only migration. A change to any of those values is a stream-topology or recovery migration and needs a separate cutover plan.

Conversion failures surface as stream-delivery failures. Throw a descriptive exception for malformed payloads, unsupported versions, and missing routing metadata so the provider retains or replays the source message according to its delivery semantics. Monitor conversion failures and quarantine poison messages through the transport’s operational process before they exhaust retention or block partition progress.

Test both conversion directions with retained messages from every deployed schema version. Include heterogeneous batches, request context, duplicate delivery, malformed envelopes, unknown versions, rolling upgrades, and replay from an older Event Hubs checkpoint.