Orleans streaming APIs

Stream identity

An Orleans stream is selected by three application-level choices:

• Provider name selects a configured IStreamProvider.
• Stream namespace groups related streams and participates in implicit-subscription matching.
• Stream key identifies one stream within the namespace and can be a string, GUID, or integer.

Together, namespace and key form StreamId. Keep identity construction in shared code:

public static class TemperatureStreams
{
    public const string ProviderName = "Telemetry";
    public const string Namespace = "device-telemetry";

    public static IAsyncStream<TemperatureReading> Get(
        Grain grain,
        string deviceId)
    {
        var provider = grain.GetStreamProvider(ProviderName);
        var streamId = StreamId.Create(Namespace, deviceId);
        return provider.GetStream<TemperatureReading>(streamId);
    }
}

GetStream returns a typed IAsyncStream<T>. Getting the provider and stream handles is local and doesn’t create a broker entity. Producers and consumers must agree on T; Orleans serialization rules apply to payloads.

Producers

Any grain or configured Orleans client can publish. A stream can have multiple producers:

public sealed class TemperatureProducerGrain : Grain, ITemperatureProducerGrain
{
    private IAsyncStream<TemperatureReading> _stream = null!;
    private IGrainTimer? _timer;

    public override Task OnActivateAsync(CancellationToken cancellationToken)
    {
        _stream = TemperatureStreams.Get(this, this.GetPrimaryKeyString());
        return Task.CompletedTask;
    }

    public Task StartAsync()
    {
        _timer ??= this.RegisterGrainTimer(
            PublishAsync,
            dueTime: TimeSpan.Zero,
            period: TimeSpan.FromSeconds(5));

        return Task.CompletedTask;
    }

    private Task PublishAsync(CancellationToken cancellationToken) =>
        _stream.OnNextAsync(new TemperatureReading
        {
            Celsius = Random.Shared.Next(-20, 45),
            ObservedAt = DateTimeOffset.UtcNow,
        });
}

Await each OnNextAsync call when publication order matters. The returned task reports provider acceptance, not end-to-end consumer completion. Provider failures and ambiguous timeouts require application-specific retry and deduplication decisions.

Consumers

Consumers attach an IAsyncObserver<T> implementation or callback delegates. OnNextAsync receives the item and, when the provider supplies one, a StreamSequenceToken.

Complete the consumer task only after the application has accepted responsibility for the item. Persistent providers use that completion to advance delivery or retry after failure. Avoid blocking threads; asynchronous consumer work naturally applies backpressure to that subscription.

Streams are multicast. Each subscription receives each item, and one grain can create multiple explicit subscriptions to the same stream. Each subscription has its own StreamSubscriptionHandle<T>.

Explicit and implicit subscriptions

The provider’s StreamPubSubType controls which subscription models are available:

Value	Choose this value when	Tradeoff
ExplicitGrainBasedAndImplicit (default)	The application uses both models or expects its subscription requirements to evolve.	Provides the most flexibility. Producer registration checks both the grain-based rendezvous and implicit grain metadata, and the explicit portion requires a PubSubStore.
ExplicitGrainBasedOnly	Every consumer uses runtime-created subscriptions, including client consumers.	Focuses discovery on explicit subscriptions. Subscription and producer changes use rendezvous grains and PubSubStore, and the application manages subscription handles and recovery.
ImplicitOnly	Every consumer is a grain declared with ImplicitStreamSubscriptionAttribute.	Uses cluster grain metadata as its subscription directory, with zero rendezvous-grain calls and zero PubSubStore operations. This gives it the lowest pub/sub control-plane overhead and makes it attractive wherever metadata-defined grain subscriptions fit. Runtime-created, individually removable, and client subscriptions use an explicit-capable mode.

The mode determines the subscription discovery, coordination, and storage work. Event transport and delivery follow the selected stream provider. Specialized modes can reduce control-plane overhead; choose a mode based on the required subscription semantics and measure the effect in the application’s workload.

Change the configured mode

Apply a pub/sub type change by updating ConfigureStreamPubSub and restarting every silo and client which uses that named provider. Restart them as a coordinated deployment so every host uses the same value and computes the same subscription set.

Each subscription model retains its own lifecycle across a change:

• Implicit subscriptions come from grain metadata. ExplicitGrainBasedOnly selects explicit records, while either implicit-capable mode applies matching metadata.
• Explicit subscription records follow the configured PubSubStore durability. ImplicitOnly selects metadata-derived subscriptions and leaves retained explicit records in the store. Account for those records before changing modes. Re-enabling explicit support with the same service ID, provider name, and durable PubSubStore makes retained records available again; each activated consumer then resumes its handle.

Use the default combined mode when subscription requirements are expected to evolve and continuous support for both models outweighs the additional pub/sub control-plane work.

Explicit subscriptions

Use an explicit subscription when application behavior decides whether and when a grain or client subscribes. SubscribeAsync creates a new subscription every time, so activation code must resume existing handles rather than subscribe again:

public sealed class ExplicitTelemetryGrain :
    Grain,
    IExplicitTelemetryGrain,
    IAsyncObserver<TemperatureReading>
{
    private readonly ILogger<ExplicitTelemetryGrain> _logger;
    private IAsyncStream<TemperatureReading> _stream = null!;

    public ExplicitTelemetryGrain(ILogger<ExplicitTelemetryGrain> logger) =>
        _logger = logger;

    public override async Task OnActivateAsync(CancellationToken cancellationToken)
    {
        _stream = TemperatureStreams.Get(this, this.GetPrimaryKeyString());

        var handles = await _stream.GetAllSubscriptionHandles();
        foreach (var handle in handles)
        {
            await handle.ResumeAsync(this);
        }
    }

    public async Task SubscribeAsync()
    {
        var handles = await _stream.GetAllSubscriptionHandles();
        if (handles.Count == 0)
        {
            await _stream.SubscribeAsync(this);
        }
    }

    public async Task UnsubscribeAsync()
    {
        var handles = await _stream.GetAllSubscriptionHandles();
        foreach (var handle in handles)
        {
            await handle.UnsubscribeAsync();
        }
    }

    public Task OnNextAsync(
        TemperatureReading item,
        StreamSequenceToken? token = null)
    {
        _logger.LogInformation(
            "Received {Temperature} C at {ObservedAt}",
            item.Celsius,
            item.ObservedAt);
        return Task.CompletedTask;
    }

    public Task OnErrorAsync(Exception ex)
    {
        _logger.LogError(ex, "The telemetry subscription failed");
        return Task.CompletedTask;
    }
}

The subscription belongs to the grain identity, not one activation. After deactivation, a later activation calls GetAllSubscriptionHandles and ResumeAsync to attach its new observer instance. Call UnsubscribeAsync to remove a subscription.

This lifecycle is durable across cluster restarts only when the configured PubSubStore is durable. A memory PubSubStore preserves records only while that cluster state remains available.

End an explicit subscription

End a subscription by awaiting UnsubscribeAsync for every handle. The streaming runtime removes each subscription from pub/sub storage and notifies active producers before the operation completes. The example’s UnsubscribeAsync method follows this sequence.

Implicit subscriptions

Use an implicit subscription when a stream item should activate a grain determined by the stream identity:

[ImplicitStreamSubscription(TemperatureStreams.Namespace)]
public sealed class DeviceTelemetryGrain :
    Grain,
    IDeviceTelemetryGrain,
    IAsyncObserver<TemperatureReading>,
    IStreamSubscriptionObserver
{
    private readonly ILogger<DeviceTelemetryGrain> _logger;
    private double? _latest;

    public DeviceTelemetryGrain(ILogger<DeviceTelemetryGrain> logger) =>
        _logger = logger;

    public Task OnSubscribed(IStreamSubscriptionHandleFactory handleFactory)
    {
        var handle = handleFactory.Create<TemperatureReading>();
        return handle.ResumeAsync(this);
    }

    public Task OnNextAsync(
        TemperatureReading item,
        StreamSequenceToken? token = null)
    {
        _latest = item.Celsius;
        _logger.LogInformation(
            "Device {DeviceId} reported {Temperature} C",
            this.GetPrimaryKeyString(),
            item.Celsius);
        return Task.CompletedTask;
    }

    public Task OnErrorAsync(Exception ex)
    {
        _logger.LogError(ex, "The telemetry subscription failed");
        return Task.CompletedTask;
    }

    public Task<double?> GetLatestAsync() => Task.FromResult(_latest);
}

ImplicitStreamSubscriptionAttribute selects stream namespaces. For each matching grain type, Orleans maps the stream key to the grain key. Implementing IStreamSubscriptionObserver lets Orleans supply the implicit handle; call ResumeAsync once to attach processing logic.

Implicit subscriptions are declared in grain metadata. They aren’t created through SubscribeAsync, can’t be individually removed at runtime, and don’t support multiple subscriptions for the same grain binding.

Clients

Clients can produce and explicitly consume streams after the provider is configured on IClientBuilder. Client subscriptions are tied to the connected client process and must be re-established after reconnecting or restarting. Implicit subscriptions target grains, not clients.

Stateless worker grains

Grains marked with StatelessWorkerAttribute can publish stream events. Orleans rejects stateless worker grain subscription attempts with an InvalidOperationException because a stream consumer uses a grain extension which must bind to one activation, while a stateless worker grain identity can have multiple, replaceable activations.

Use a regular grain as the stream consumer so that the stream-to-grain binding has a stable virtual identity which can own state and the subscription lifecycle. If processing after delivery is stateless and parallelizable, have that grain call stateless worker grains and await the required work before its consumer task completes. This keeps stream acknowledgment and recovery at the regular grain boundary instead of treating a multicast subscription as a competing-consumer work queue.

Support for subscribing directly from stateless worker grains is tracked by dotnet/orleans#433.

For failure behavior and sequence tokens, continue to Delivery, ordering, replay, and recovery.

For a larger compiled example, see SampleStreamingGrain.cs in the Orleans test suite.