SlimMessageBus

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.

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

Key elements of SlimMessageBus
Docs
Packages
Samples
Features
Principles
License
Build
Testing
Credits

Key elements of SlimMessageBus

Consumers:
- IConsumer<in TMessage> - subscriber in pub/sub (or queue consumer)
- IRequestHandler<in TRequest, TResponse> & IRequestHandler<in TRequest> - request handler in request-response
Producers:
- IPublishBus - publisher in pub/sub (or queue producer)
- IRequestResponseBus - sender in req/resp
- IMessageBus - extends IPublishBus and IRequestResponseBus
Misc:
- IRequest<out TResponse> & IRequest - a marker for request messages
- MessageBus - static accessor for current context IMessageBus

Docs

Introduction
Providers:
- Apache Kafka
- Azure EventHubs
- Azure ServiceBus
- Hybrid
- MQTT
- Memory
- NATS
- RabbitMQ
- Redis
- SQL
Plugins:

Packages

Name	Description	NuGet
`SlimMessageBus`	The core API for SlimMessageBus
Transport providers
`.Host.AzureEventHub`	Transport provider for Azure Event Hubs
`.Host.AzureServiceBus`	Transport provider for Azure Service Bus
`.Host.Kafka`	Transport provider for Apache Kafka
`.Host.Memory`	Transport provider implementation for in-process (in memory) message passing (no messaging infrastructure required)
`.Host.MQTT`	Transport provider for MQTT
`.Host.NATS`	Transport provider for NATS
`.Host.RabbitMQ`	Transport provider for RabbitMQ
`.Host.Redis`	Transport provider for Redis
`.Host.Sql` (pending)	Transport provider implementation for SQL database message passing
Serialization
`.Host.Serialization.Json`	Serialization plugin for JSON (Newtonsoft.Json library)
`.Host.Serialization.SystemTextJson`	Serialization plugin for JSON (System.Text.Json library)
`.Host.Serialization.Avro`	Serialization plugin for Avro (Apache.Avro library)
`.Host.Serialization.Hybrid`	Plugin that delegates serialization to other serializers based on message type
`.Host.Serialization.GoogleProtobuf`	Serialization plugin for Google Protobuf
Plugins
`.Host.AspNetCore`	Integration for ASP.NET Core
`.Host.Interceptor`	Core interface for interceptors
`.Host.FluentValidation`	Validation for messages based on FluentValidation
`.Host.Outbox.Sql`	Transactional Outbox using SQL
`.Host.Outbox.DbContext`	Transactional Outbox using EF DbContext
`.Host.AsyncApi`	AsyncAPI specification generation via Saunter

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:

It has similar semantics and has the interceptor pipeline enabling the addition of custom behavior.
The generic interceptors can introduce common behavior like logging, authorization or audit of messages.
The FluentValidation plugin can introduce request/command/query validation.
The external communication can be layered on top of SlimMessageBus which allows having one library for in-memory and out-of-process messaging (Hybrid Provider).

See the CQRS and FluentValidation samples.

Use Case: Request-response over Kafka topics

See sample.

Features

Types of messaging patterns supported:
- Publish-subscribe
- Request-response
- Queues
- A hybrid of the above (e.g. Kafka with multiple topic consumers in one group)
Modern async/await syntax and TPL
Fluent configuration
SourceLink support
Because SlimMessageBus is a facade, chosen messaging transports can be swapped without impacting the overall application architecture.

Principles

The core of SlimMessageBus is "slim"
- Simple, common and friendly API to work with messaging systems
- No external dependencies.
- The core interface can be used in the domain model (e.g. Domain Events)
Plugin architecture:
- Message serialization (JSON, Avro, Protobuf)
- Use your favorite messaging broker as a provider by simply pulling a NuGet package
- Add transactional outbox pattern or message validation
No threads created (pure TPL)
Async/Await support
Fluent configuration
Logging is done via Microsoft.Extensions.Logging.Abstractions so that you can connect to your favorite logger provider.

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.

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

Redis - Redis Labs
Kafka - CloudKarafka
MQTT - HiveMQ
RabbitMQ - CloudAMQP

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

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

README.md

README.md

SlimMessageBus

Key elements of SlimMessageBus

Docs

Packages

Samples

Basic usage

Configuration

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

Use Case: MediatR replacement

Use Case: Request-response over Kafka topics

Features

Principles

License

Build

Testing

Credits

Files

README.md

Latest commit

History

README.md

File metadata and controls

SlimMessageBus

Key elements of SlimMessageBus

Docs

Packages

Samples

Basic usage

Configuration

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

Use Case: MediatR replacement

Use Case: Request-response over Kafka topics

Features

Principles

License

Build

Testing

Credits