Escendit.Orleans.Streaming.RabbitMQ 0.1.0

NuGet Package: Escendit.Orleans.Streaming.RabbitMQ

Escendit.Orleans.Streaming.RabbitMQ is a NuGet Package that integrates RabbitMQ with Orleans Streaming Provider.

The Library contains 2 different ways of streaming, first via Stream Protocol, and other via AMQP protocol.

The Library provides named connections to Rabbit MQ, which can be used for low-level integration events.

Installation

To install Escendit.Orleans.Streaming.RabbitMQ, run the following command in the Package Manager Console:

Install-Package Escendit.Orleans.Streaming.RabbitMQ

Usage

To use Escendit.Orleans.Streaming.RabbitMQ first register the stream provider using the AddRabbitMqStreaming method in the Orleans configuration:

SiloBuilder

var builder = new SiloBuilder()
    .AddRabbitMqStreaming("ProviderName");

ClientBuilder

var builder = new ClientBuilder()
    .AddRabbitMqStreaming("ProviderName");

Stream Protocol

To use Escendit.Orleans.Streaming.RabbitMQ with Stream Protocol, first register the stream provider using the AddRabbitMqStreaming method in the Orleans configuration, and then call the WithStream method to specify that the provider should use the Stream Protocol:

SiloBuilder

var hostBuilder = Host
    .CreateDefaultBuilder()
    .UseOrleans(builder =>
    {
        builder
            .Configure<ClusterOptions>(options =>
            {
                options.ClusterId = "cluster-id";
                options.ServiceId = "service-id";
            })
            .AddStreaming()
            .AddRabbitMqStreaming("ProviderName")
            .WithStream(options =>
            {
                options.Endpoints.Add(new RabbitEndpoint { HostName = "localhost", Port = 5552 });
                options.UserName = "guest";
                options.Password = "guest";
                options.VirtualHost = "/";
            });
    });

ClientBuilder

var hostBuilder = Host
    .CreateDefaultBuilder()
    .UseOrleansClient(builder =>
    {
        builder
            .Configure<ClusterOptions>(options =>
            {
                options.ClusterId = "cluster-id";
                options.ServiceId = "service-id";
            })
            .AddStreaming()
            .AddRabbitMqStreaming("ProviderName")
            .WithStream(options =>
            {
                options.Endpoints.Add(new RabbitEndpoint { HostName = "localhost", Port = 5552 });
                options.UserName = "guest";
                options.Password = "guest";
                options.VirtualHost = "/";
            });            
    })

AMQP Protocol

To use Escendit.Orleans.Streaming.RabbitMQ with Stream Protocol, first register the stream provider using the AddRabbitMqStreaming method in the Orleans configuration, and then call the WithQueue method to specify that the provider should use the AMQP Protocol:

SiloBuilder

var hostBuilder = Host
    .CreateDefaultBuilder()
    .UseOrleans(builder =>
    {
        builder
            .Configure<ClusterOptions>(options =>
            {
                options.ClusterId = "cluster-id";
                options.ServiceId = "service-id";
            })
            .AddStreaming()
            .AddRabbitMqStreaming("ProviderName")
            .WithQueue(options =>
            {
                options.Endpoints.Add(new RabbitEndpoint { HostName = "localhost", Port = 5672 });
                options.UserName = "guest";
                options.Password = "guest";
                options.VirtualHost = "/";
            });
    });

ClientBuilder

var hostBuilder = Host
    .CreateDefaultBuilder()
    .UseOrleansClient(builder =>
    {
        builder
            .Configure<ClusterOptions>(options =>
            {
                options.ClusterId = "cluster-id";
                options.ServiceId = "service-id";
            })
            .AddStreaming()
            .AddRabbitMqStreaming("ProviderName")
            .WithQueue(options =>
            {
                options.Endpoints.Add(new RabbitEndpoint { HostName = "localhost", Port = 5672 });
                options.UserName = "guest";
                options.Password = "guest";
                options.VirtualHost = "/";
            });            
    })

Low Level Registration

You can register the Rabbit MQ Classic Client, but the name must not conflict with the Orleans Streaming Provider.

SiloBuilder

var hostBuilder = Host
    .CreateDefaultBuilder()
    .UseOrleans(builder =>
    {
        builder
            .Configure<ClusterOptions>(options =>
            {
                options.ClusterId = "cluster-id";
                options.ServiceId = "service-id";
            })
            .ConfigureServices(services => services
                .AddRabbitMq("ProviderName", options =>
                {
                    options.Endpoints.Add(new RabbitEndpoint { HostName = "localhost", Port = 5672 });
                    options.UserName = "guest";
                    options.Password = "guest";
                    options.VirtualHost = "/";
                })
                .WithConnection());
    });

ClientBuilder

var hostBuilder = Host
    .CreateDefaultBuilder()
    .UseOrleansClient(builder =>
    {
        builder
            .Configure<ClusterOptions>(options =>
            {
                options.ClusterId = "cluster-id";
                options.ServiceId = "service-id";
            })
            .ConfigureServices(services => services
                .AddRabbitMq("ProviderName", options =>
                {
                    options.Endpoints.Add(new RabbitEndpoint { HostName = "localhost", Port = 5672 });
                    options.UserName = "guest";
                    options.Password = "guest";
                    options.VirtualHost = "/";
                })
                .WithConnection());          
    })

Consume

Create a stream:

public class MyGrain : Grain, IMyGrain
{
    private IAsyncStream<T> _stream;
    public override async Task OnActivateAsync()
    {
        var streamProvider = this.GetStreamProvider("ProviderName");
        _stream = _streamProvider.GetStream<int>(this.GetPrimaryKey(), "stream-namespace");
        _stream.SubscribeAsync(ReceiveMessage);
        await base.OnActivateAsync();
    }

    public async Task SendMessage()
    {
        // Send a message
        await _stream.OnNextAsync(42);
    }
    
    public Task ReceiveMessage(Event @event)
    {
        // Receive a message.
    }
}

Create a low-level connection:

public class MyService : GrainService
{
    private readonly IConnection _connection;
    public MyService(string name, IServiceProvider serviceProvider)
    {
        _connection = serviceProvider.GetRequiredServiceByName<IConnection>(name);
    }
}

Contributing

If you'd like to contribute to Escendit.Orleans.Streaming.RabbitMQ, please fork the repository and make changes as you'd like. Pull requests are warmly welcome.