SlimMessageBus is a client façade for message brokers for .NET. It comes with implementations for specific brokers (Apache Kafka, Azure EventHub, MQTT/Mosquitto, Redis Pub/Sub) and in-memory message passing (in-process communication). SlimMessageBus additionally provides request-response implementation over message queues.
IConsumer<in TMessage>
- subscriber in pub/sub (or queue consumer)IRequestHandler<in TRequest, TResponse>
- request handler in request-responseIPublishBus
- publisher in pub/sub (or queue producer)IRequestResponseBus
- sender in req/respIMessageBus
- extends IPublishBus
and IRequestResponseBus
IRequestMessage<TResponse>
- marker for request messagesMessageBus
- static accessor for current context IMessageBus
Name | Description | NuGet |
---|---|---|
SlimMessageBus |
The core API for SlimMessageBus | |
Transport providers | ||
.Host.Kafka |
Transport provider for Apache Kafka | |
.Host.AzureServiceBus |
Transport provider for Azure Service Bus | |
.Host.AzureEventHub |
Transport provider for Azure Event Hubs | |
.Host.Redis |
Transport provider for Redis | |
.Host.Memory |
Transport provider implementation for in-process (in memory) message passing (no messaging infrastructure required) | |
.Host.Hybrid |
Bus implementation that composes the bus out of other transport providers and performs message routing | |
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 | |
IoC Container | ||
.Host.MsDependencyInjection |
DI adapter for Microsoft.Extensions.DependencyInjection | |
.Host.AspNetCore |
Integration for ASP.NET Core (DI adapter, config helpers) | |
.Host.Autofac |
DI adapter for Autofac container | |
.Host.Unity |
DI adapter for Unity container | |
Plugins | ||
.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 |
Typically your application components (business logic, domain) only need to depend on SlimMessageBus
which is the facade, and ultimately your application hosting layer (ASP.NET, Windows Service, Console App) will reference and configure the other packages (SlimMessageBus.Host.*
) which are the providers and plugins.
Some service (or domain layer) sends 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 (or topics - depending on what the chosen transport provider supports). The sender side sends a request message:
var messageResponse = await bus.Send(new MessageRequest());
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 back:
public class MessageRequestHandler : IRequestHandler<MessageRequest, MessageResponse>
{
public async Task<MessageResponse> OnHandle(MessageRequest request)
{
// handle the request message and return response
}
}
The bus will ask the chosen DI to provide the consumer instances (SomeMessageConsumer
, MessageRequestHandler
).
When Microsoft.Extensions.DependencyInjection
is used, then SMB can be configured (requires either MsDependencyInjection
or AspNetCore
plugins):
IServiceCollection services;
services.AddSlimMessageBus(mbb =>
{
mbb
.AddChildBus("Bus1", (builder) =>
{
builder
.Produce<SomeMessage>(x => x.DefaultTopic("some-topic"))
.Consume<SomeMessage>(x => x
.Topic("some-topic")
.WithConsumer<SomeMessageConsumer>()
//.KafkaGroup("some-kafka-consumer-group") // Kafka provider specific
//.SubscriptionName("some-azure-sb-topic-subscription") // Azure ServiceBus provider specific
);
// ...
.WithProviderKafka(new KafkaMessageBusSettings("localhost:9092")); // requires SlimMessageBus.Host.Kafka package
// Use Azure Service Bus transport provider
//.WithProviderServiceBus(...)
// Use Azure Azure Event Hub transport provider
//.WithProviderEventHub(...)
// Use Redis transport provider
//.WithProviderRedis(...)
// Use in-memory transport provider
//.WithProviderMemory(...)
})
// Add other bus transports, if needed
//.AddChildBus("Bus2", (builder) => {})
.WithSerializer(new JsonMessageSerializer()) // requires SlimMessageBus.Host.Serialization.Json package
.WithProviderHybrid(); // requires SlimMessageBus.Host.Hybrid package
},
// Option 1 (optional)
addConsumersFromAssembly: new[] { Assembly.GetExecutingAssembly() }, // auto discover consumers and register into DI (see next section)
addInterceptorsFromAssembly: new[] { Assembly.GetExecutingAssembly() }, // auto discover interceptors and register into DI (see next section)
addConfiguratorsFromAssembly: new[] { Assembly.GetExecutingAssembly() } // auto discover modular configuration and register into DI (see next section)
);
// Option 2 (optional)
services.AddMessageBusConsumersFromAssembly(Assembly.GetExecutingAssembly());
services.AddMessageBusInterceptorsFromAssembly(Assembly.GetExecutingAssembly());
services.AddMessageBusConfiguratorsFromAssembly(Assembly.GetExecutingAssembly());
See the Dependency Resolver for more information.
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 app domain process (no external message broker is required).
The domain event is a simple POCO:
// domain event
public record OrderSubmittedEvent(Order Order, DateTime Timestamp);
The event handler implements the IConsumer<T>
interface:
// domain event handler
public class OrderSubmittedHandler : IConsumer<OrderSubmittedEvent>
{
public Task OnHandle(OrderSubmittedEvent e)
{
Console.WriteLine("Customer {0} {1} just placed an order for:", e.Order.Customer.Firstname, e.Order.Customer.Lastname);
foreach (var orderLine in e.Order.Lines)
{
Console.WriteLine("- {0}x {1}", orderLine.Quantity, orderLine.ProductId);
}
Console.WriteLine("Generating a shipping order...");
return Task.Delay(1000);
}
}
The domain event handler (consumer) is obtained from the dependency resolver at the time of event publication. It can be scoped (per web request, per unit of work) as configured in your favorite DI container.
Somewhere in your domain 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;
return MessageBus.Current.Publish(new OrderSubmittedEvent(this)); // raise domain event
}
}
Some sample logic executed in your domain:
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 might actually be configured to resolve a scoped IMessageBus
instance (web request-scoped or pick up message scope from an external bus).
The SlimMessageBus
configuration for in-memory provider looks like this:
IServiceCollection services; // for MsDependencyInjection or AspNetCore
// Cofigure the message bus
services.AddSlimMessageBus(mbb =>
{
mbb
.WithProviderMemory()
.AutoDeclareFrom(Assembly.GetExecutingAssembly()); // Find types that implement IConsumer<T> and IRequestHandler<T, R> and declare producers and consumers for them
},
addConsumersFromAssembly: new[] { Assembly.GetExecutingAssembly() } // Auto discover consumers and register inside DI container
);
In Startup.cs
for the ASP.NET project, set up the MessageBus.Current
helper (if you want to use it):
public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
{
// Set the MessageBus provider, so that IMessageBus are resolved from the current request scope
MessageBus.SetProvider(MessageBusCurrentProviderBuilder.Create().From(app).Build());
}
See the complete sample for ASP.NET Core where the handler and bus are web-request scoped.
See sample.
SlimMessageBus
is "slim"
Microsoft.Extensions.Logging.Abstractions
so that you can connect to your favorite logger provider.cd src
dotnet build
dotnet pack --output ../dist
NuGet packaged end up in dist
folder
To run tests you need to update the respective appsettings.json
to match your own 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
Thanks to Gravity9 for providing an Azure subscription that allows to run our integration test infrastructure.
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.