Write a custom persistent-stream queue adapter

Use a custom queue adapter when Orleans doesn’t include a provider for the queue technology you need. The adapter translates between Orleans stream batches and the external queue. Orleans supplies the persistent stream provider, pulling agents, subscription routing, queue balancing, and cache management.

You don’t create a subclass of PersistentStreamProvider. Register an IQueueAdapterFactory with AddPersistentStreams; Orleans creates and hosts the provider.

Before implementing an adapter, understand the persistent stream pulling architecture. In particular, decide the adapter’s delivery, acknowledgement, ordering, partitioning, and rewind semantics.

Implement the transport boundary

Keep the queue SDK and wire-format logic behind a transport abstraction. The transport must:

serialize the stream ID, event payloads, and request context into an evolvable envelope;
assign a monotonically ordered sequence number within each queue partition;
return only messages from the requested partition;
acknowledge or delete messages only after Orleans calls the completion method; and
surface queue failures instead of returning a successful empty read.

public interface ICustomQueueTransport
{
    Task SendAsync(
        QueueId queueId,
        StreamId streamId,
        object[] events,
        Dictionary<string, object>? requestContext,
        CancellationToken cancellationToken);

    Task<IReadOnlyList<CustomQueueMessage>> ReceiveAsync(
        QueueId queueId,
        int maxCount,
        CancellationToken cancellationToken);

    Task CompleteAsync(
        QueueId queueId,
        IReadOnlyList<CustomQueueMessage> messages,
        CancellationToken cancellationToken);
}

The sequence number must remain stable when the queue redelivers a message. The example assigns sequence numbers in the transport, so it rejects caller-supplied sequence tokens when producing events.

Implement the adapter and receiver

IQueueAdapter handles writes and creates one receiver per queue partition. The stream-to-queue mapper used for writes must be the same mapper returned by the factory.

public sealed class CustomQueueAdapter(
    string name,
    ICustomQueueTransport transport,
    IStreamQueueMapper mapper) : IQueueAdapter
{
    public string Name => name;

    public bool IsRewindable => false;

    public StreamProviderDirection Direction => StreamProviderDirection.ReadWrite;

    public Task QueueMessageBatchAsync<T>(
        StreamId streamId,
        IEnumerable<T> events,
        StreamSequenceToken? token,
        Dictionary<string, object>? requestContext)
    {
        if (token is not null)
        {
            throw new ArgumentException(
                "This adapter doesn't support caller-supplied sequence tokens.",
                nameof(token));
        }

        var queueId = mapper.GetQueueForStream(streamId);
        return transport.SendAsync(
            queueId,
            streamId,
            events.Cast<object>().ToArray(),
            requestContext,
            CancellationToken.None);
    }

    public IQueueAdapterReceiver CreateReceiver(QueueId queueId) =>
        new CustomQueueReceiver(queueId, transport);
}

IQueueAdapterReceiver reads queue messages and acknowledges them after every consumer has processed them. If the queue uses visibility leases, renew them while Orleans retains the message and make shutdown cancel outstanding reads.

public sealed class CustomQueueReceiver(
    QueueId queueId,
    ICustomQueueTransport transport) : IQueueAdapterReceiver
{
    public Task Initialize(TimeSpan timeout) => Task.CompletedTask;

    [Obsolete("Use the overload which accepts a CancellationToken.")]
    public Task<IList<IBatchContainer>> GetQueueMessagesAsync(int maxCount) =>
        GetQueueMessagesAsync(maxCount, CancellationToken.None);

    public async Task<IList<IBatchContainer>> GetQueueMessagesAsync(
        int maxCount,
        CancellationToken cancellationToken)
    {
        var messages = await transport.ReceiveAsync(
            queueId,
            maxCount,
            cancellationToken);

        return messages.Cast<IBatchContainer>().ToList();
    }

    [Obsolete("Use the overload which accepts a CancellationToken.")]
    public Task MessagesDeliveredAsync(IList<IBatchContainer> messages) =>
        MessagesDeliveredAsync(messages, CancellationToken.None);

    public Task MessagesDeliveredAsync(
        IList<IBatchContainer> messages,
        CancellationToken cancellationToken) =>
        transport.CompleteAsync(
            queueId,
            messages.Cast<CustomQueueMessage>().ToArray(),
            cancellationToken);

    public Task Shutdown(TimeSpan timeout) => Task.CompletedTask;
}

The batch container restores the stream identity, per-event sequence tokens, and request context when Orleans delivers the batch.

public sealed class CustomQueueMessage(
    StreamId streamId,
    object[] events,
    long sequenceNumber,
    Dictionary<string, object>? requestContext) : IBatchContainer
{
    private readonly EventSequenceToken _sequenceToken = new(sequenceNumber);

    public StreamId StreamId { get; } = streamId;

    public StreamSequenceToken SequenceToken => _sequenceToken;

    public IEnumerable<Tuple<T, StreamSequenceToken>> GetEvents<T>() =>
        events.OfType<T>().Select(
            (item, index) => Tuple.Create<T, StreamSequenceToken>(
                item,
                _sequenceToken.CreateSequenceTokenForEvent(index)));

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

        RequestContextExtensions.Import(requestContext);
        return true;
    }
}

Implement the factory

The factory composes the adapter with queue mapping, caching, and failure handling. Use named options because one process can register multiple providers with different names.

public sealed class CustomQueueAdapterFactory : IQueueAdapterFactory
{
    private readonly string _name;
    private readonly ICustomQueueTransport _transport;
    private readonly IStreamQueueMapper _mapper;
    private readonly IQueueAdapterCache _cache;

    public CustomQueueAdapterFactory(
        string name,
        ICustomQueueTransport transport,
        HashRingStreamQueueMapperOptions mapperOptions,
        SimpleQueueCacheOptions cacheOptions,
        ILoggerFactory loggerFactory)
    {
        _name = name;
        _transport = transport;
        _mapper = new HashRingBasedStreamQueueMapper(mapperOptions, name);
        _cache = new SimpleQueueAdapterCache(cacheOptions, name, loggerFactory);
    }

    [Obsolete("Use the overload which accepts a CancellationToken.")]
    public Task<IQueueAdapter> CreateAdapter() =>
        CreateAdapter(CancellationToken.None);

    public Task<IQueueAdapter> CreateAdapter(CancellationToken cancellationToken) =>
        Task.FromResult<IQueueAdapter>(
            new CustomQueueAdapter(_name, _transport, _mapper));

    public IQueueAdapterCache GetQueueAdapterCache() => _cache;

    public IStreamQueueMapper GetStreamQueueMapper() => _mapper;

    public Task<IStreamFailureHandler> GetDeliveryFailureHandler(QueueId queueId) =>
        Task.FromResult<IStreamFailureHandler>(
            new NoOpStreamDeliveryFailureHandler());

    public static IQueueAdapterFactory Create(IServiceProvider services, string name) =>
        new CustomQueueAdapterFactory(
            name,
            services.GetRequiredService<ICustomQueueTransport>(),
            services.GetOptionsByName<HashRingStreamQueueMapperOptions>(name),
            services.GetOptionsByName<SimpleQueueCacheOptions>(name),
            services.GetRequiredService<ILoggerFactory>());
}

SimpleQueueAdapterCache is suitable for a non-rewindable adapter whose queue remains the durability boundary. A rewindable adapter usually needs a cache and sequence-token implementation which can position cursors at retained historical messages.

Checkpointing is adapter-specific, not a requirement imposed by AddPersistentStreams. The non-rewindable example acknowledges completed messages through its receiver and therefore has no independent checkpoint. For a retained-log transport, implement an IStreamQueueCheckpointerFactory, have the receiver or cache load and update the per-partition position, and register it as a named component with ConfigureComponent. Persist a checkpoint only after all consumers have advanced beyond the corresponding cached messages. A no-op checkpointer is suitable only when replay position is deliberately disposable.

Register the provider

Register the transport client in dependency injection, then pass the factory’s Create method to AddPersistentStreams. Configure queue count and cache capacity through the provider configurator.

public static void ConfigureSilo(
    ISiloBuilder builder,
    ICustomQueueTransport transport)
{
    builder.Services.AddSingleton(transport);
    builder.AddPersistentStreams(
        ProviderName,
        CustomQueueAdapterFactory.Create,
        streams =>
        {
            streams.Configure<HashRingStreamQueueMapperOptions>(
                options => options.Configure(
                    value => value.TotalQueueCount = 8));
            streams.Configure<SimpleQueueCacheOptions>(
                options => options.Configure(
                    value => value.CacheSize = 4_096));
        });
}

public static void ConfigureClient(
    IClientBuilder builder,
    ICustomQueueTransport transport)
{
    builder.Services.AddSingleton(transport);
    builder.AddPersistentStreams(
        ProviderName,
        CustomQueueAdapterFactory.Create,
        streams => streams.Configure<HashRingStreamQueueMapperOptions>(
            options => options.Configure(
                value => value.TotalQueueCount = 8)));
}

Keep the provider name and partition count stable. Changing either can map an existing stream to a different queue and strand previously enqueued messages. Configure durable PubSubStore grain storage for explicit subscriptions in production; queue durability doesn’t preserve Orleans subscription records.

Validate failure behavior

Test the adapter against the real queue service, including:

batches containing multiple event types and request-context values;
empty reads, cancellation, transient errors, throttling, and shutdown;
producer, receiver, and silo failure before and after acknowledgement;
queue ownership moving between silos during membership changes;
duplicate delivery and consumer idempotency;
stable stream-to-partition mapping across restarts and upgrades; and
sustained load beyond cache capacity to verify backpressure and queue retention.

Monitor queue depth and oldest-message age by partition, receive and acknowledgement latency, redelivery count, throttling, pulling-agent errors, and consumer delivery failures. Alert before retention or visibility limits can cause data loss or a redelivery storm.