Skip to content
Lightweight messaging wrapper of MassTransit
Branch: master
Clone or download
Latest commit 0a23d34 Apr 5, 2019
Permalink
Type Name Latest commit message Commit time
Failed to load latest commit information.
.vscode
misc First commit Feb 13, 2017
src Autofac pck fixed in order to add multiple consumer Apr 5, 2019
tests
.gitattributes First commit Feb 13, 2017
.gitignore First commit Feb 13, 2017
LICENSE First commit Feb 13, 2017
MetroBus.sln microsoft ex di add to the sln Oct 16, 2018
README.md some badge added Apr 5, 2019

README.md

MetroBus


alt tag

Lightweight messaging wrapper of MassTransit

NuGet version NuGet version NuGet version NuGet version

NuGet Packages

> dotnet add package MetroBus
> dotnet add package MetroBus.Autofac
> dotnet add package MetroBus.Microsoft.Extensions.DependencyInjection

Supports:

  • .NET Standard 2.0

Features:

  • Currently only supports RabbitMQ transport
  • Provides easy way to create Producer and Consumer for Pub/Sub
  • Provides easy way to handle Request/Response conversations
  • Provides message scheduling
  • Includes optional incremental auto retry policy
  • Includes optional circuit breaker
  • Includes optional rate limiter
  • Autofac support
  • Microsoft.Extensions.DependencyInjection support

Usage:

Initializing bus instance for Producer:

// For events
IBusControl bus = MetroBusInitializer.Instance.UseRabbitMq(rabbitMqUri, rabbitMqUserName, rabbitMqPassword)
					.InitializeEventProducer();

// For commands
ISendEndpoint bus = MetroBusInitializer.Instance.UseRabbitMq(rabbitMqUri, rabbitMqUserName, rabbitMqPassword)
                    .InitializeCommandProducer(queueName);

after bus instance initializing then you can use Send or Publish methods.

// For events
await bus.Publish<TEvent>(new
{
    SomeProperty = SomeValue
}));

// For commands
await bus.Send<TCommand>(new
{
    SomeProperty = SomeValue
}));

using Consumer:

static void Main(string[] args)
{
	IBusControl bus = MetroBusInitializer.Instance
                        .UseRabbitMq(string rabbitMqUri, string rabbitMqUserName, string rabbitMqPassword)
                        .RegisterConsumer<TCommandConsumer>(string queueName)
                        .RegisterConsumer<TEventConsumer>(string queueName)
                        .Build();

	bus.Start();

	//if you want to stop
	bus.Stop();

	Console.ReadLine();
}

TCommandConsumer could like below:

public class TCommandConsumer : IConsumer<TCommand>
{
    public async Task Consume(ConsumeContext<TCommand> context)
    {
        var command = context.Message;

		//do something...
        await Console.Out.WriteAsync($"{command.SomeProperty}");
    }
}

Initializing bus instance for Request/Response conversation:

IRequestClient<TRequest, TResponse> client = MetroBusInitializer.Instance.UseRabbitMq(string rabbitMqUri, string rabbitMqUserName, string rabbitMqPassword)
                                                                    .InitializeRequestClient<TRequest, TResponse>(string queueName);

TResponse result = await client.Request(new TRequest
{
    Command = "Say hello!"
});

and consumer for Request/Response conversation could like below:

public class TCommandConsumer : IConsumer<TRequest>
{
    public async Task Consume(ConsumeContext<TRequest> context)
    {
        var command = context.Message;

		//do something...
        await Console.Out.WriteAsync($"{command.SomeProperty}");

		//and
		context.Respond(new TRequest
            {
                Command = "Hello!"
            });
    }
}

using Consumer with Microsoft.Extensions.DependencyInjection:

new HostBuilder ()
    .ConfigureServices ((hostContext, services) =>
    {
        services.AddMetroBus (x =>
        {
            x.AddConsumer<TCommandConsumer>();
            x.AddConsumer<TEventConsumer>();
        });

        services.AddSingleton<IBusControl> (provider => MetroBusInitializer.Instance
                .UseRabbitMq (string rabbitMqUri, string rabbitMqUserName, string rabbitMqPassword)
                .RegisterConsumer<TCommandConsumer>("foo.command.queue", provider)
                .RegisterConsumer<TEventConsumer>("foo.event.queue", provider)
                .Build ())
            .BuildServiceProvider ();

        services.AddHostedService<BusService> ();
    })
    .RunConsoleAsync ().Wait ();


public class BusService : IHostedService
{
    private readonly IBusControl _busControl;

    public BusService(IBusControl busControl)
    {
        _busControl = busControl;
    }

    public Task StartAsync(CancellationToken cancellationToken)
    {
        return _busControl.StartAsync(cancellationToken);
    }

    public Task StopAsync(CancellationToken cancellationToken)
    {
        return _busControl.StopAsync(cancellationToken);
    }
}

PS: Publisher and Consumer services must be used same TCommand or TEvent interfaces. This is important for MassTransit integration. Also one other thing is rabbitMqUri parameter must start with "rabbitmq://" prefix.

There are several options you can set via fluent interface:

  • .UseRetryPolicy().UseIncrementalRetryPolicy(int retryLimit, TimeSpan? initialIntervalTime, TimeSpan? intervalIncrementTime, params Exception[] retryOnSpecificExceptionType).Then()
  • .UseCircuitBreaker(int tripThreshold, int activeThreshold, TimeSpan? resetInterval)
  • .UseRateLimiter(int rateLimit, TimeSpan? interval)
  • .UseMessageScheduler()
  • .UseDelayedExchangeMessageScheduler()
  • .UseConcurrentConsumerLimit(int concurrencyLimit)
You can’t perform that action at this time.