Orleans streaming quickstart

This quickstart uses the Microsoft.Orleans.Streaming package and the memory stream provider. It needs no external broker, but both queued data and the PubSubStore are in memory. Use this configuration for development and tests, not for production durability.

Configure the silo

Register the same provider name used by producers and consumers. PubSubStore stores explicit stream subscription metadata.

public static IHostApplicationBuilder AddStreamingSilo(
    this IHostApplicationBuilder builder)
{
    builder.UseOrleans(siloBuilder =>
    {
        siloBuilder
            .AddMemoryStreams(TemperatureStreams.ProviderName)
            .AddMemoryGrainStorage("PubSubStore");
    });

    return builder;
}

If an external Orleans client publishes or subscribes, configure the same provider on that client:

public static IHostApplicationBuilder AddStreamingClient(
    this IHostApplicationBuilder builder)
{
    builder.UseOrleansClient(clientBuilder =>
    {
        clientBuilder.AddMemoryStreams(TemperatureStreams.ProviderName);
    });

    return builder;
}

Define the event and stream identity

Stream payloads use the normal Orleans serialization model. The example also centralizes the provider name and stream namespace so producers and consumers construct identical stream identities.

[GenerateSerializer]
public sealed class TemperatureReading
{
    [Id(0)]
    public required double Celsius { get; init; }

    [Id(1)]
    public required DateTimeOffset ObservedAt { get; init; }
}

public interface ITemperatureProducerGrain : IGrainWithStringKey
{
    Task StartAsync();
}

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);
    }
}

The stream namespace is device-telemetry, and the stream key is the device ID. The payload type is part of the typed handle, not part of StreamId.

Publish events

The producer obtains its stream handle during activation and starts an Orleans grain timer when requested:

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,
        });
}

Use RegisterGrainTimer for grain timers. Awaiting OnNextAsync waits until the provider accepts responsibility according to its contract; it doesn’t generally wait for every consumer to finish. See Producer acknowledgment.

Consume with an implicit subscription

An implicit subscription maps the stream key to a grain key. Publishing to device-telemetry/device-17 activates the DeviceTelemetryGrain whose string key is device-17. Implement IStreamSubscriptionObserver to attach the observer supplied by Orleans:

[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);
}

Obtain ITemperatureProducerGrain with the same device key and invoke StartAsync to begin publishing. The producer and consumer don’t reference one another; they agree on provider name, stream identity, and event type.

Next, read Streaming APIs, then replace the memory provider and memory PubSubStore with production services selected from the provider matrix.