Microsoft.Orleans.Streaming.AzureStorage 9.2.0-preview2

Microsoft Orleans Streaming for Azure Storage Queues

Introduction

Microsoft Orleans Streaming for Azure Storage provides a stream provider implementation for Orleans using Azure Storage Queues. This allows for publishing and subscribing to streams of events with Azure Storage Queues as the underlying messaging infrastructure.

Getting Started

To use this package, install it via NuGet:

dotnet add package Microsoft.Orleans.Streaming.AzureStorage

Example - Configuring Azure Storage Queues Streaming

using Microsoft.Extensions.Hosting;
using Orleans.Hosting;
using Orleans.Streams;

var builder = Host.CreateApplicationBuilder(args)
    .UseOrleans(siloBuilder =>
    {
        siloBuilder
            .UseLocalhostClustering()
            // Configure Azure Storage Queues as a stream provider
            .AddAzureQueueStreams(
                name: "AzureQueueStreamProvider", 
                b => b.ConfigureAzureQueue(ob => ob.Configure((options, dep) =>
                {
                    options.ConfigureTestDefaults();
                    options.QueueNames = Enumerable.Range(0, 8).Select(num => $"{dep.Value.ClusterId}-{num}").ToList();
                })));
    });

// Run the host
await builder.RunAsync();

Example - Using Azure Storage Queue Streams in a Grain

// Producer grain
public class ProducerGrain : Grain, IProducerGrain
{
    private IAsyncStream<string> _stream;

    public override Task OnActivateAsync(CancellationToken cancellationToken)
    {
        // Get a reference to a stream
        var streamProvider = GetStreamProvider("AzureQueueStreamProvider");
        _stream = streamProvider.GetStream<string>(Guid.NewGuid(), "MyStreamNamespace");
        
        return base.OnActivateAsync(cancellationToken);
    }

    public async Task SendMessage(string message)
    {
        // Send a message to the stream
        await _stream.OnNextAsync(message);
    }
}

// Consumer grain
public class ConsumerGrain : Grain, IConsumerGrain, IAsyncObserver<string>
{
    private StreamSubscriptionHandle<string> _subscription;

    public override async Task OnActivateAsync(CancellationToken cancellationToken)
    {
        // Get a reference to a stream
        var streamProvider = GetStreamProvider("AzureQueueStreamProvider");
        var stream = streamProvider.GetStream<string>(this.GetPrimaryKey(), "MyStreamNamespace");
        
        // Subscribe to the stream
        _subscription = await stream.SubscribeAsync(this);
        
        await base.OnActivateAsync(cancellationToken);
    }

    public Task OnNextAsync(string item, StreamSequenceToken token = null)
    {
        Console.WriteLine($"Received message: {item}");
        return Task.CompletedTask;
    }

    public Task OnCompletedAsync()
    {
        Console.WriteLine("Stream completed");
        return Task.CompletedTask;
    }

    public Task OnErrorAsync(Exception ex)
    {
        Console.WriteLine($"Stream error: {ex.Message}");
        return Task.CompletedTask;
    }
}

Documentation

For more comprehensive documentation, please refer to:

Feedback & Contributing

Showing the top 20 packages that depend on Microsoft.Orleans.Streaming.AzureStorage.

Packages Downloads
Microsoft.Orleans.OrleansAzureUtils
Support library for hosting Orleans on Microsoft Azure.
2
Microsoft.Orleans.OrleansAzureUtils
Support library for hosting Orleans on Microsoft Azure.
1

.NET 8.0

Version Downloads Last updated
9.2.0-preview3 1 6/14/2025
9.2.0-preview2 2 6/6/2025
9.2.0-preview1 1 6/7/2025
9.1.2 1 6/7/2025
9.0.1 2 6/6/2025
9.0.0 1 6/7/2025
8.2.0 1 6/7/2025
8.2.0-preview1 1 6/7/2025
8.1.0 1 6/7/2025
8.1.0-preview3 1 6/8/2025
8.1.0-preview2 1 6/7/2025
8.1.0-preview1 1 6/8/2025
8.0.0 1 6/7/2025
8.0.0-rc2 1 6/7/2025
8.0.0-rc1 2 6/7/2025
7.2.7 1 6/7/2025
7.2.6 1 6/7/2025
7.2.5 1 6/7/2025
7.2.4 1 6/7/2025
7.2.3 1 6/7/2025
7.2.2 1 6/7/2025
7.2.1 1 6/7/2025
7.2.0 1 6/7/2025
7.1.2 1 6/7/2025
7.1.1 1 6/7/2025
7.1.0 1 6/7/2025
7.0.0 1 6/7/2025
7.0.0-rc2 1 6/7/2025
4.0.0-preview2 1 6/7/2025
4.0.0-preview1 1 6/7/2025
3.8.0 1 6/10/2025
3.8.0-preview5 0 5/12/2025
3.8.0-preview3 1 6/7/2025
3.8.0-preview2 1 6/7/2025
3.8.0-preview1 1 6/7/2025
3.7.2 1 6/7/2025
3.7.1 1 6/7/2025
3.7.0 1 6/7/2025
3.6.5 1 6/8/2025
3.6.4 1 6/7/2025
3.6.3 1 6/7/2025
3.6.2 1 6/7/2025
3.6.1 1 6/7/2025
3.6.0 1 6/7/2025
3.5.1 1 6/7/2025
3.5.0 1 6/7/2025
3.4.4 1 6/7/2025
3.4.3 1 6/7/2025
3.4.2 2 6/7/2025
3.4.1 1 6/7/2025
3.4.0 1 6/7/2025
3.4.0-rc1 1 6/7/2025
3.3.0 1 6/7/2025
3.3.0-rc2 1 6/7/2025
3.3.0-rc1 1 6/7/2025
3.2.2 1 6/7/2025
3.2.1 1 6/7/2025
3.2.0 1 6/7/2025
3.2.0-rc2 1 6/7/2025
3.2.0-rc1 2 6/7/2025
3.1.7 1 6/7/2025
3.1.6 1 6/7/2025
3.1.5 1 6/7/2025
3.1.4 1 6/7/2025
3.1.3 1 6/7/2025
3.1.2 1 6/7/2025
3.1.0 1 6/7/2025
3.1.0-rc3 1 6/7/2025
3.1.0-rc2 1 6/7/2025
3.1.0-rc1 1 6/7/2025
3.0.2 1 6/7/2025
3.0.1 1 6/8/2025
3.0.0 1 6/7/2025
3.0.0-rc2 1 6/7/2025
3.0.0-rc1 1 6/7/2025
3.0.0-beta1 1 6/7/2025
2.4.5 1 6/7/2025
2.4.4 1 6/7/2025
2.4.3 1 6/7/2025
2.4.2 2 6/7/2025
2.4.1 1 6/7/2025
2.4.0 1 6/7/2025
2.3.6 1 6/7/2025
2.3.5 1 6/7/2025
2.3.4 1 6/7/2025
2.3.3 1 6/7/2025
2.3.2 1 6/7/2025
2.3.1 1 6/7/2025
2.3.0 1 6/7/2025
2.3.0-rc2 1 6/7/2025
2.3.0-rc1 1 6/7/2025
2.2.4 1 6/7/2025
2.2.3 1 6/7/2025
2.2.0 1 6/7/2025
2.2.0-rc1 1 6/7/2025
2.2.0-beta1 1 6/7/2025
2.1.2 1 6/7/2025
2.1.0 1 6/7/2025
2.1.0-rc2 1 6/7/2025
2.1.0-rc1 1 6/7/2025
2.1.0-beta1 1 6/7/2025
2.0.0 1 6/7/2025
2.0.0-rc2 1 6/7/2025
2.0.0-rc1 1 6/7/2025
2.0.0-beta3 1 6/7/2025
2.0.0-beta2 1 6/7/2025