Home

Awesome

SlimMessageBus <!-- omit in toc -->

SlimMessageBus is a client façade for message brokers for .NET. It comes with implementations for specific brokers (RabbitMQ, Kafka, Azure EventHub, MQTT, Redis Pub/Sub) and in-memory message passing (in-process communication). SlimMessageBus additionally provides request-response implementation over message queues.

Gitter GitHub license Build Maintainability Rating Coverage Duplicated Lines (%) Vulnerabilities Quality Gate Status

The v2 release is available (see migration guide). The v3 release is under construction.

Key elements of SlimMessageBus

Docs

Packages

NameDescriptionNuGet
SlimMessageBusThe core API for SlimMessageBusNuGet
Transport providers
.Host.AzureEventHubTransport provider for Azure Event HubsNuGet
.Host.AzureServiceBusTransport provider for Azure Service BusNuGet
.Host.KafkaTransport provider for Apache KafkaNuGet
.Host.MemoryTransport provider implementation for in-process (in memory) message passing (no messaging infrastructure required)NuGet
.Host.MQTTTransport provider for MQTTNuGet
.Host.NATSTransport provider for NATSNuGet
.Host.RabbitMQTransport provider for RabbitMQNuGet
.Host.RedisTransport provider for RedisNuGet
.Host.Sql (pending)Transport provider implementation for SQL database message passingNuGet
Serialization
.Host.Serialization.JsonSerialization plugin for JSON (Newtonsoft.Json library)NuGet
.Host.Serialization.SystemTextJsonSerialization plugin for JSON (System.Text.Json library)NuGet
.Host.Serialization.AvroSerialization plugin for Avro (Apache.Avro library)NuGet
.Host.Serialization.HybridPlugin that delegates serialization to other serializers based on message typeNuGet
.Host.Serialization.GoogleProtobufSerialization plugin for Google ProtobufNuGet
Plugins
.Host.AspNetCoreIntegration for ASP.NET CoreNuGet
.Host.InterceptorCore interface for interceptorsNuGet
.Host.FluentValidationValidation for messages based on FluentValidationNuGet
.Host.Outbox.SqlTransactional Outbox using SQLNuGet
.Host.Outbox.DbContextTransactional Outbox using EF DbContextNuGet
.Host.AsyncApiAsyncAPI specification generation via SaunterNuGet

Typically the application layers (domain model, business logic) only need to depend on SlimMessageBus which is the facade, and ultimately the application hosting layer (ASP.NET, Console App, Windows Service) will reference and configure the other packages (SlimMessageBus.Host.*) which are the messaging transport providers and additional plugins.

Samples

Basic usage

Some service (or domain layer) publishes a message:

IMessageBus bus; // injected

await bus.Publish(new SomeMessage());

Another service (or application layer) handles the message:

public class SomeMessageConsumer : IConsumer<SomeMessage>
{
   public async Task OnHandle(SomeMessage message)
   {
       // handle the message
   }
}

Note: It is also possible to avoid having to implement the interface IConsumer<T> (see here).

The bus also supports request-response implemented via queues, topics or in-memory - depending on the chosen transport provider. The sender side sends a request message:

var response = await bus.Send(new SomeRequest());

Note: It is possible to configure the bus to timeout a request when the response does not arrive within the allotted time (see here).

The receiving side handles the request and replies:

public class SomeRequestHandler : IRequestHandler<SomeRequest, SomeResponse>
{
   public async Task<SomeResponse> OnHandle(SomeRequest request)
   {
      // handle the request message and return a response
      return new SomeResponse { /* ... */ };
   }
}

The bus will ask the DI container to provide the consumer instances (SomeMessageConsumer, SomeRequestHandler).

There is also support for one-way request-response.

Configuration

The Microsoft.Extensions.DependencyInjection is used to compose the bus:

// IServiceCollection services;

services.AddSlimMessageBus(mbb =>
{
   mbb
      // First child bus - in this example Kafka transport
      .AddChildBus("Bus1", (builder) =>
      {
         builder
            .Produce<SomeMessage>(x => x.DefaultTopic("some-topic"))
            .Consume<SomeMessage>(x => x.Topic("some-topic")
               //.WithConsumer<SomeMessageConsumer>() // Optional: can be skipped as IConsumer<SomeMessage> will be resolved from DI
               //.KafkaGroup("some-kafka-consumer-group") // Kafka: Consumer Group
               //.SubscriptionName("some-azure-sb-topic-subscription") // Azure ServiceBus: Subscription Name
            );
            // ...
            // Use Kafka transport provider (requires SlimMessageBus.Host.Kafka package)
            .WithProviderKafka(cfg => { cfg.BrokerList = "localhost:9092"; }); // requires SlimMessageBus.Host.Kafka package
            // Use Azure Service Bus transport provider
            //.WithProviderServiceBus(cfg => { ... }) // requires SlimMessageBus.Host.AzureServiceBus package
            // Use Azure Event Hub transport provider
            //.WithProviderEventHub(cfg => { ... }) // requires SlimMessageBus.Host.AzureEventHub package
            // Use Redis transport provider
            //.WithProviderRedis(cfg => { ... }) // requires SlimMessageBus.Host.Redis package
            // Use RabbitMQ transport provider
            //.WithProviderRabbitMQ(cfg => { ... }) // requires SlimMessageBus.Host.RabbitMQ package
            // Use in-memory transport provider
            //.WithProviderMemory(cfg => { ... }) // requires SlimMessageBus.Host.Memory package
      })

      // Add other bus transports (as child bus), if needed
      //.AddChildBus("Bus2", (builder) => {  })

      // Scan assembly for consumers, handlers, interceptors, and register into MSDI
      .AddServicesFromAssemblyContaining<SomeMessageConsumer>()
      //.AddServicesFromAssembly(Assembly.GetExecutingAssembly());

      // Add JSON serializer
      .AddJsonSerializer(); // requires SlimMessageBus.Host.Serialization.Json or SlimMessageBus.Host.Serialization.SystemTextJson package
});

The configuration can be modularized.

Use Case: Domain Events (in-process pub/sub messaging)

This example shows how SlimMessageBus and SlimMessageBus.Host.Memory can be used to implement the Domain Events pattern. The provider passes messages in the same process (no external message broker is required).

The domain event is a simple POCO:

// domain event
public record OrderSubmittedEvent(Order Order, DateTime Timestamp);

The domain event handler implements the IConsumer<T> interface:

// domain event handler
public class OrderSubmittedHandler : IConsumer<OrderSubmittedEvent>
{
   public Task OnHandle(OrderSubmittedEvent e)
   {
      // ...
   }
}

The domain event handler (consumer) is obtained from the MSDI at the time of event publication. The event publish enlists in the ongoing scope (web request scope, external message scope of the ongoing message).

In the domain model layer, the domain event gets raised:

// aggregate root
public class Order
{
   public Customer Customer { get; }
   public OrderState State { get; private set; }

   private IList<OrderLine> lines = new List<OrderLine>();
   public IEnumerable<OrderLine> Lines => lines.AsEnumerable();

   public Order(Customer customer)
   {
      Customer = customer;
      State = OrderState.New;
   }

   public OrderLine Add(string productId, int quantity) { }

   public Task Submit()
   {
      State = OrderState.Submitted;

      // Raise domain event
      return MessageBus.Current.Publish(new OrderSubmittedEvent(this));
   }
}

Sample logic executed by the client of the domain model:

var john = new Customer("John", "Whick");

var order = new Order(john);
order.Add("id_machine_gun", 2);
order.Add("id_grenade", 4);

await order.Submit(); // events fired here

Notice the static MessageBus.Current property is configured to resolve a scoped IMessageBus instance (web request-scoped or pick-up message scope from a currently processed message).

The SlimMessageBus configuration for the in-memory provider looks like this:

//IServiceCollection services;

// Configure the message bus
services.AddSlimMessageBus(mbb =>
{
   mbb.WithProviderMemory();
   // Find types that implement IConsumer<T> and IRequestHandler<T, R> and declare producers and consumers on the mbb
   mbb.AutoDeclareFrom(Assembly.GetExecutingAssembly());
   // Scan assembly for consumers, handlers, interceptors, and register into MSDI
   mbb.AddServicesFromAssemblyContaining<OrderSubmittedHandler>();
});

For the ASP.NET project, set up the MessageBus.Current helper (if you want to use it, and pick up the current web-request scope):

services.AddSlimMessageBus(mbb =>
{
   // ...
   mbb.AddAspNet(); // requires SlimMessageBus.Host.AspNetCore package
});
services.AddHttpContextAccessor(); // This is required by the SlimMessageBus.Host.AspNetCore plugin

See the complete sample for ASP.NET Core where the handler and bus are web-request scoped.

Use Case: MediatR replacement

The SlimMessageBus in-memory provider can replace the need to use MediatR library:

See the CQRS and FluentValidation samples.

Use Case: Request-response over Kafka topics

See sample.

Features

Principles

License

Apache License 2.0

Build

cd src
dotnet build
dotnet pack --output ../dist

NuGet packages end up in dist folder

Testing

To run tests you need to update the secrets.txt to match your cloud infrastructure or local infrastructure. SMB has some message brokers set up on Azure for integration tests (secrets not shared).

Run all tests:

dotnet test

Run all tests except integration tests that require local/cloud infrastructure:

dotnet test --filter Category!=Integration

Credits

Thanks to Gravity9 for providing an Azure subscription that allows running the integration test infrastructure.

<a href="https://www.gravity9.com/" target="_blank"><img src="https://uploads-ssl.webflow.com/5ce7ef1205884e25c3d2daa4/5f71f56c89fd4db58dd214d3_Gravity9_logo.svg" width="100" alt="Gravity9"></a>

Thanks to the following service cloud providers for providing free instances for our integration tests:

If you want to help and sponsor, please write to me.