Overview

This library provides extensions for MassTransit to support:

• Hosting with Microsoft.Extensions.Hosting (i.e. IHostedService)
• Dependency Injection with Microsoft.Extensions.DependencyInjection (i.e. IServiceProvider)

This library was designed to make no assumptions how to configure MassTransit bus instances and provides the developer with flexible hosting implementation options.

• MassTransit.Extensions.Hosting
  • Provides extensions common to all MassTransit transports
  • Includes an in-memory bus

Transport specific libraries:

• MassTransit.Extensions.Hosting.RabbitMq
• MassTransit.Extensions.Hosting.ActiveMq
• MassTransit.Extensions.Hosting.AmazonSqs
• MassTransit.Extensions.Hosting.AzureServiceBus
• MassTransit.Extensions.Hosting.AzureServiceBusCore
• MassTransit.Extensions.Hosting.Http

Problem Statement

The core MassTransit library only provides abstractions and no implementation unless the MassTransit.Host package is used. Unfortunatly the MassTransit.Host package makes many assumptions and forces the developer with many potentially unwanted conventions such as:

• Autofac as the DI container
  • No ability to modify the registrations in ContainerBuilder
• A prebuilt Windows Service executable using Topshelf
  • No ability to modify the Topshelf service configuration
• log4net as the logging provider
• Configuration settings from assembly config files
  • web.config is not supported

None of the items mentioned above are bad or wrong, just potentially not intended to be used in every host implementation. The individual libraries such as Autofac, log4net and Topshelf are in fact extremly helpful, just not always needed in every implementation.

Also the MassTransit.Host is not usable in other hosting environments such as Web Applications, Console Applications, etc since it provides a prebuilt Topshelf executable. Instead it would be convenient to use the same style of design for MassTransit bus instances but agnostic to their hosting environment.

Proposed Solution

This library uses the new Generic Host pattern from ASP.NET Core as the glue for building MassTransit applications. Other than using the hosting and dependency injection abstractions, this library makes no assumptions on DI containers, logging providers, configuration providers, and the hosting environment.

Features

• Fluent configuration interfaces
• No assumptions on configuration (i.e. the developer is in full control)
• Access to the existing MassTransit configuration interfaces (i.e. IBusFactoryConfigurator)
• Ability to retrieve the bus host by connection name after startup
• Supports all the available MassTransit transports

Usage

Step 1) Add NuGet Package(s)

Choose one (or multiple) of the following transports:

PM> Install-Package MassTransit.Extensions.Hosting.RabbitMq

PM> Install-Package MassTransit.Extensions.Hosting.ActiveMq

PM> Install-Package MassTransit.Extensions.Hosting.AmazonSqs

PM> Install-Package MassTransit.Extensions.Hosting.AzureServiceBus

PM> Install-Package MassTransit.Extensions.Hosting.AzureServiceBusCore

PM> Install-Package MassTransit.Extensions.Hosting.Http

Step 2) Add MassTransit Services

public void ConfigureServices(IServiceCollection services)
{
    services.AddMassTransit(busBuilder =>
    {
        // ...
    });
}

Step 3) Configure Bus Transports

public void ConfigureServices(IServiceCollection services)
{
    services.AddMassTransit(busBuilder =>
    {
        busBuilder.UseInMemory("connection-name-1", hostBuilder => 
        {
            hostBuilder.UseServiceScope();

            hostBuilder.AddConfigurator(configureBus =>
            {
                configureBus.UseRetry(r => r.Immediate(3));
            });

            // ...
        });

        busBuilder.UseRabbitMq("connection-name-2", "127.0.0.1", "/vhost", hostBuilder => 
        {
            hostBuilder.UseCredentials("guest", "guest");

            // ...
        });

        busBuilder.UseActiveMq("connection-name-3", "127.0.0.1", hostBuilder => 
        {
            hostBuilder.UseSsl();
            hostBuilder.UseCredentials("guest", "guest");

            // ...
        });

        busBuilder.UseAmazonSqs("connection-name-4", hostBuilder => 
        {
            hostBuilder.UseRegion(RegionEndpoint.USEast1);
            hostBuilder.UseCredentials(new EnvironmentVariablesAWSCredentials());

            // ...
        });

        busBuilder.UseAzureServiceBus("connection-name-5", new Uri("sb://namespace.servicebus.windows.net/scope"), hostBuilder =>
        {
            hostBuilder.UseServiceScope();

            hostBuilder.UseTokenProvider(TokenProvider.CreateSimpleWebTokenProvider("token-test"));

            hostBuilder.AddConfigurator(configureBus =>
            {
                configureBus.SelectBasicTier();
            });
        });

        busBuilder.UseHttp("connection-name-6", new Uri("http://127.0.0.1:8080"), hostBuilder =>
        {
            hostBuilder.UseServiceScope();
            hostBuilder.UseMethod(HttpMethod.Post);
            hostBuilder.AddReceiveEndpoint("example-queue-6", endpointBuilder =>
            {
                endpointBuilder.AddConsumer<ExampleConsumer>();
            });
        });

    });
}

Step 4) Configure Receive Endpoints

public void ConfigureServices(IServiceCollection services)
{
    services.AddMassTransit(busBuilder =>
    {
        busBuilder.UseInMemory("connection-name-1", hostBuilder => 
        {
            // ...

            hostBuilder.AddReceiveEndpoint("example-queue-1", endpointBuilder =>
            {
                endpointBuilder.AddConfigurator(configureEndpoint =>
                {
                    configureEndpoint.UseRetry(r => r.Immediate(3));
                });

                // ...
            });
        });
    });
}

Step 5) Configure Consumers

public void ConfigureServices(IServiceCollection services)
{
    services.AddMassTransit(busBuilder =>
    {
        busBuilder.UseInMemory("connection-name-1", hostBuilder => 
        {
            // ...

            hostBuilder.AddReceiveEndpoint("example-queue-1", endpointBuilder =>
            {
                endpointBuilder.AddConsumer<ExampleConsumer>(configureConsumer =>
                {
                    configureConsumer.UseRateLimit(10);

                    // ...
                });
            });
        });
    });
}

Step 6) Bus Manager

public async Task Run(IServiceProvider serviceProvider)
{
    var busManager = serviceProvider.GetRequiredService<IBusManager>();

    // start all bus instances
    await busManager.StartAsync().ConfigureAwait(false);

    // get a reference to a named bus instance
    // and publish a message
    IBus bus = busManager.GetBus("connection-name-1");
    await bus.Publish(/* ... */).ConfigureAwait(false);

    // ...

    // stop all bus instances
    await busManager.StopAsync().ConfigureAwait(false);
}

Example Console Host

public static class Program
{
    private static void Main(string[] args)
    {
        // use the new generic host from ASP.NET Core
        // see for more info: https://github.com/aspnet/Hosting/issues/1163
        new HostBuilder()
            .ConfigureHostConfiguration(config => config.AddEnvironmentVariables())
            .ConfigureAppConfiguration((context, builder) => ConfigureAppConfiguration(context, builder, args))
            .ConfigureServices(ConfigureServices)
            .Build()
            .Run();
    }

    private static void ConfigureAppConfiguration(HostBuilderContext context, IConfigurationBuilder configurationBuilder, string[] args)
    {
        var environmentName = context.HostingEnvironment.EnvironmentName;

        configurationBuilder.AddJsonFile("appsettings.json", optional: true);
        configurationBuilder.AddJsonFile($"appsettings.{environmentName}.json", optional: true);

        configurationBuilder.AddCommandLine(args);
    }

    private static void ConfigureServices(HostBuilderContext context, IServiceCollection services)
    {
        services.AddTransient<IWidgetService, WidgetService>();
        // add other DI services...

        // optionally use configuration for any settings
        var configuration = context.Configuration;

        // the following adds IBusManager which is also an IHostedService that is started/stopped by HostBuilder
        services.AddMassTransit(busBuilder =>
        {
            // configure RabbitMQ
            busBuilder.UseRabbitMq(configuration.GetSection("MassTransit:RabbitMq"), hostBuilder =>
            {
                // use scopes for all downstream filters and consumers
                // i.e. per-request lifetime
                hostBuilder.UseServiceScope();

                // example adding an optional configurator to the bus
                // using IRabbitMqBusFactoryConfigurator
                hostBuilder.Add