### Configure Outbox Using DbContext in Startup Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/plugin_outbox.md Example of configuring the Outbox.DbContext plugin and integrating it with an application's DbContext. This setup ensures messages are delivered via the outbox mechanism. ```csharp services.AddOutboxUsingDbContext(...) ``` -------------------------------- ### Install SlimMessageBus and Providers Source: https://github.com/zarusz/slimmessagebus/blob/master/README.md Install the core SlimMessageBus package and any necessary transport provider and serialization plugins using the .NET CLI. ```bash dotnet add package SlimMessageBus # Add specific transport provider, e.g. Kafka dotnet add package SlimMessageBus.Host.Kafka # Add serialization plugin dotnet add package SlimMessageBus.Host.Serialization.SystemTextJson ``` -------------------------------- ### MediatR Setup in Program.cs Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/UseCases/ReplaceMediatR.md Demonstrates a typical setup for MediatR in a .NET application's Program.cs file, including service registration and basic endpoint definition. ```csharp using MediatR; var builder = WebApplication.CreateBuilder(args); // Register MediatR services builder.Services.AddMediatR(cfg => { cfg.RegisterServicesFromAssemblyContaining(); }); var app = builder.Build(); app.MapGet("/ping", async (IMediator mediator) => { return await mediator.Send(new PingRequest()); }); app.Run(); // Define a simple request and handler public record PingRequest : IRequest; public class PingRequestHandler : IRequestHandler { public Task Handle(PingRequest request, CancellationToken cancellationToken) { return Task.FromResult("Pong"); } } ``` -------------------------------- ### Install Avro .NET Tools Source: https://github.com/zarusz/slimmessagebus/blob/master/src/Samples/Sample.Serialization.MessagesAvro/README.md Install the Apache.Avro.Tools globally using the .NET CLI. This tool is required for generating C# classes from Avro IDL files. ```cmd dotnet tool install -g Apache.Avro.Tools ``` -------------------------------- ### Start Local Infrastructure Containers Source: https://github.com/zarusz/slimmessagebus/blob/master/src/Infrastructure/README.md Use this command to create and host local containers for message bus testing. It ensures clean instances by not using volumes. ```bash src/Infrastructure> docker compose up --force-recreate -V ``` ```bash > ./infrastructure.sh ``` ```powershell > .\infrastructure.ps1 ``` -------------------------------- ### Configure SQL Transport Provider Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_sql.md Configure the SQL transport provider for SlimMessageBus using the `.WithProviderSql()` method. This example shows basic setup within the message bus builder. ```csharp services.AddSlimMessageBus(mbb => { mbb.WithProviderSql(cfg => { // ToDo }); mbb.AddServicesFromAssemblyContaining(); mbb.AddJsonSerializer(); }); ``` -------------------------------- ### Configure GoogleProtobuf Serializer Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/serialization.md Installs the GoogleProtobuf serializer for IMessage objects. Ensure the Google.Protobuf library is available. ```csharp mbb.AddGoogleProtobufSerializer(); ``` -------------------------------- ### SlimMessageBus Setup with In-Memory Provider in Program.cs Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/UseCases/ReplaceMediatR.md Shows how to configure SlimMessageBus with the in-memory provider in Program.cs, including service registration and endpoint definition. ```csharp using SlimMessageBus; using SlimMessageBus.Host.Memory; var builder = WebApplication.CreateBuilder(args); // Configure SlimMessageBus with In-Memory provider builder.Services.AddSlimMessageBus(mbb => { mbb.WithProviderMemory().AutoDeclareFrom(typeof(Program).Assembly); }); var app = builder.Build(); app.MapGet("/ping", async (IMessageBus bus) => { return await bus.Send(new PingRequest()); }); app.Run(); // Define request and handler compatible with SlimMessageBus public record PingRequest : IRequestMessage; public class PingRequestHandler : IRequestHandler { public async Task OnHandle(PingRequest request, CancellationToken cancellationToken) { return "Pong"; } } ``` -------------------------------- ### Kafka Consumer Debug Logging Example Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_kafka.md Example log output demonstrating Kafka consumer lifecycle events at Debug level. This helps in understanding partition assignments, message reception, and offset commits during rebalancing or normal operation. ```log [00:03:06 INF] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Subscribing to topics: 4p5ma6io-test-ping [00:03:06 INF] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Consumer loop started [00:03:12 DBG] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Assigned partition, Topic: 4p5ma6io-test-ping, Partition: [0] [00:03:12 INF] SlimMessageBus.Host.Kafka.KafkaPartitionConsumer Creating consumer for Group: subscriber, Topic: 4p5ma6io-test-ping, Partition: [0] [00:03:12 DBG] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Assigned partition, Topic: 4p5ma6io-test-ping, Partition: [1] [00:03:12 INF] SlimMessageBus.Host.Kafka.KafkaPartitionConsumer Creating consumer for Group: subscriber, Topic: 4p5ma6io-test-ping, Partition: [1] ... [00:03:15 DBG] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Received message with Topic: 4p5ma6io-test-ping, Partition: [1], Offset: 98578, payload size: 57 [00:03:15 INF] SlimMessageBus.Host.Kafka.Test.KafkaMessageBusIt.PingConsumer Got message 073 on topic 4p5ma6io-test-ping. [00:03:15 DBG] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Received message with Topic: 4p5ma6io-test-ping, Partition: [1], Offset: 98579, payload size: 57 [00:03:15 INF] SlimMessageBus.Host.Kafka.Test.KafkaMessageBusIt.PingConsumer Got message 075 on topic 4p5ma6io-test-ping. [00:03:16 DBG] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Reached end of partition, Topic: 4p5ma6io-test-ping, Partition: [0], Offset: 100403 [00:03:16 DBG] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Commit Offset, Topic: 4p5ma6io-test-ping, Partition: [0], Offset: 100402 [00:03:16 DBG] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Reached end of partition, Topic: 4p5ma6io-test-ping, Partition: [1], Offset: 98580 [00:03:16 DBG] SlimMessageBus.Host.Kafka.KafkaGroupConsumer Group [subscriber]: Commit Offset, Topic: 4p5ma6io-test-ping, Partition: [1], Offset: 98579 ``` -------------------------------- ### Configure Redis Transport with Connection String Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_redis.md Configure the Redis transport provider for SlimMessageBus using a connection string. Ensure the SlimMessageBus.Host.Redis package is installed. This setup also includes JSON serialization and service registration. ```csharp var connectionString = "server1:6379,server2:6379" // Redis connection string services.AddSlimMessageBus(mbb => { // requires SlimMessageBus.Host.Redis package mbb.WithProviderRedis(cfg => { cfg.ConnectionString = connectionString; }); mbb.AddJsonSerializer(); mbb.AddServicesFromAssembly(Assembly.GetExecutingAssembly()); }); ``` -------------------------------- ### Add MongoDB Outbox Package Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/plugin_outbox_mongodb.md Install the required NuGet package for MongoDB outbox support. ```bash dotnet add package SlimMessageBus.Host.Outbox.MongoDb ``` -------------------------------- ### Configure Service Bus Topology Provisioning Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_azure_servicebus.md Customize topology provisioning settings for the Service Bus provider. This example disables consumer creation of topics but allows creation of subscriptions and filters, while enabling validation of subscription filters. ```csharp mbb.WithProviderServiceBus(cfg => { cfg.TopologyProvisioning = new ServiceBusTopologySettings { Enabled = true, CanConsumerCreateTopic = false, // the consumers will not be able to provision a missing topic CanConsumerCreateSubscription = true, // the consumers will not be able to add a missing subscription if needed CanConsumerCreateSubscriptionFilter = true, // the consumers will not be able to add a missing filter on subscription CanConsumerValidateSubscriptionFilters = true, // any deviations from the expected will be logged }; ... }); ``` -------------------------------- ### Add MassTransit NuGet Packages Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/UseCases/ReplaceMassTransit.md Install the necessary MassTransit packages for Azure Service Bus integration. ```shell dotnet add package MassTransit dotnet add package MassTransit.Azure.ServiceBus.Core ``` -------------------------------- ### Manually Start and Stop Consumers Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/intro.md Inject IConsumerControl to manually start or stop message consumers after the bus has been created. Requires the SlimMessageBus.Host package. ```csharp IConsumerControl consumerControl = (IConsumerControl)bus; // Need to reference SlimMessageBus.Host package // Start message consumers await consumerControl.Start(); // or // Stop message consumers await consumerControl.Stop(); ``` -------------------------------- ### Consumer-Specific Subscription Filter and Topic Options Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_azure_servicebus.md Configure specific topic creation options and subscription filters for a consumer. This example sets a correlation filter and overrides topic creation settings. ```csharp mbb.Consume(x => x .Topic("some-topic") .WithConsumer() .SubscriptionName("some-service") .SubscriptionCorrelationFilter(correlationId: "some-id") // this will create a rule to filter messages with a CorrelationId of 'some-id' .CreateTopicOptions((options) => { options.RequiresDuplicateDetection = false; }) ); ``` -------------------------------- ### Example Handler for Auto-Declaration Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_memory.md An example of a request handler that would be automatically discovered and configured by AutoDeclareFrom. The bus automatically registers a producer for EchoRequest and a handler for EchoRequest/EchoResponse. ```csharp public class EchoRequestHandler : IRequestHandler { public Task OnHandle(EchoRequest request, CancellationToken cancellationToken) { // ... } } ``` -------------------------------- ### Kafka Producer with Message Key Provider Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_kafka.md Configure a `KeyProvider` to generate a message key for partitioning. The key must be a `byte[]`. This example uses the sum of Left and Right properties as the key. ```csharp // MessageBusBuilder mbb; mbb .Produce(x => { x.DefaultTopic("topic1"); // Message key could be set for the message x.KeyProvider((request, topic) => Encoding.ASCII.GetBytes((request.Left + request.Right).ToString())); }) .WithProviderKafka(cfg => cfg.BrokerList = kafkaBrokers); ``` -------------------------------- ### Add SlimMessageBus NuGet Packages Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/UseCases/ReplaceMassTransit.md Install the required SlimMessageBus packages for Azure Service Bus and System.Text.Json serialization. ```shell dotnet add package SlimMessageBus dotnet add package SlimMessageBus.Host.AzureServiceBus dotnet add package SlimMessageBus.Host.Serialization.SystemTextJson ``` -------------------------------- ### Consume with Exact Routing Key Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_rabbitmq.md Example of binding a consumer to a specific queue and exchange with an exact routing key for direct or topic exchanges. ```csharp mbb.Consume(x => x .Queue("orders-queue") .ExchangeBinding("orders-exchange", routingKey: "orders.created") .WithConsumer()); ``` -------------------------------- ### Add Outbox Using Direct SQL Connection for PostgreSQL Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/plugin_outbox.t.md Use this to add the Outbox.Sql plugin for direct SQL connection with PostgreSQL. Ensure the SlimMessageBus.Host.Outbox.PostgreSql package is installed. ```csharp using SlimMessageBus.Host.Outbox.PostgreSql; ``` -------------------------------- ### Configure Unrecognized Messages to DLX with Consumer Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_rabbitmq.md Example demonstrating how to configure the bus to send unrecognized messages to a Dead Letter Exchange while also setting up a consumer for specific events. This aids in investigating routing issues. ```cs services.AddSlimMessageBus(mbb => { mbb.WithProviderRabbitMQ(cfg => { // Send unrecognized messages to DLX for investigation cfg.MessageUnrecognizedRoutingKeyHandler = (transportMessage) => RabbitMqMessageConfirmOptions.Nack; }); mbb.Consume(x => x .Queue("orders-queue") .ExchangeBinding("orders", routingKey: "orders.created") // Unrecognized messages will be routed to this DLX .DeadLetterExchange("orders-dlq", exchangeType: ExchangeType.Direct) .WithConsumer()); }); ``` -------------------------------- ### Configure Saunter for AsyncAPI Schema Generation Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/plugin_asyncapi.md Add Saunter services to your application, configuring the AsyncAPI document information such as version and description. This should be done after SlimMessageBus setup. ```csharp // Add Saunter to the application services. builder.Services.AddAsyncApiSchemaGeneration(options => { options.AsyncApi = new AsyncApiDocument { Info = new Info("SlimMessageBus AsyncAPI Sample API", "1.0.0") { Description = "This is a sample of the SlimMessageBus AsyncAPI plugin", License = new License("Apache 2.0") { Url = "https://www.apache.org/licenses/LICENSE-2.0" } } }; }); ``` -------------------------------- ### Request Handler Interceptor Example Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/intro.md Demonstrates implementing an interceptor for request handlers that do not return a response, using `Void` as the response type. ```csharp public class SomeRequestInterceptor : IRequestHandlerInterceptor { public async Task OnHandle(TRequest request, Func> next, IConsumerContext context) { // ... pre-handling logic // Handle the actual request await next(); // ... post-handling logic // The return value is not used return null; } } ``` -------------------------------- ### Generate Documentation from Code Comments Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/plugin_asyncapi.md Example of C# code demonstrating how XML comments on message types and consumer methods are used to generate documentation within the AsyncAPI specification. Ensure GenerateDocumentationFile is enabled in your project settings. ```csharp /// /// Event when a customer is created within the domain. /// /// /// /// public record CustomerCreatedEvent(Guid Id, string Firstname, string Lastname) : CustomerEvent(Id); public class CustomerCreatedEventConsumer : IConsumer { /// /// Upon the will store it with the database. /// /// /// public Task OnHandle(CustomerCreatedEvent message, CancellationToken cancellationToken) { } ``` -------------------------------- ### Configure SlimMessageBus with Kafka Transport Source: https://github.com/zarusz/slimmessagebus/blob/master/README.md Configure SlimMessageBus with Kafka transport, including message routing for publish-subscribe and request-response patterns. This example also sets up JSON serialization and registers consumers from an assembly. ```csharp services.AddSlimMessageBus(mbb => { mbb.AddChildBus("Bus1", builder => { builder // the pub-sub events .Produce(x => x.DefaultPath("orders-topic")) .Consume(x => x.Path("orders-topic") //.WithConsumer() // Optional: can be skipped as IConsumer will be resolved from DI //.KafkaGroup("kafka-consumer-group") // Kafka: Consumer Group //.SubscriptionName("azure-sb-topic-subscription") // Azure ServiceBus: Subscription Name ) // the request-response .Produce(x => x.DefaultPath("customer-requests")) .Handle(x => x.Path("customer-requests")) // Use Kafka transport provider (requires SlimMessageBus.Host.Kafka package) .WithProviderKafka(cfg => cfg.BrokerList = "localhost:9092"); // 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 for in memory domain events), if needed //.AddChildBus("Bus2", (builder) => { }) .AddJsonSerializer() // requires SlimMessageBus.Host.Serialization.SystemTextJson or SlimMessageBus.Host.Serialization.Json package .AddServicesFromAssemblyContaining(); }); The configuration can be [modularized](docs/intro.md#modularization-of-configuration) (for modular monoliths). ``` -------------------------------- ### Add Outbox Using DbContext for PostgreSQL Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/plugin_outbox.t.md Use this to add the Outbox.DbContext plugin when integrating with Entity Framework and PostgreSQL. Ensure the SlimMessageBus.Host.Outbox.PostgreSql.DbContext package is installed. ```csharp using SlimMessageBus.Host.Outbox.PostgreSql.DbContext; ``` -------------------------------- ### Add Outbox Using Direct SQL Connection for SQL Server Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/plugin_outbox.t.md Use this to add the Outbox.Sql plugin for direct SQL connection with SQL Server. Ensure the SlimMessageBus.Host.Outbox.Sql package is installed. ```csharp using SlimMessageBus.Host.Outbox.Sql; ``` -------------------------------- ### Configure Outbox with Entity Framework DbContext Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/plugin_outbox.t.md Configure the SlimMessageBus container to use the Outbox.DbContext plugin. Replace `CustomerContext` with your application's specific Entity Framework DbContext. This setup is typically done during application startup. ```csharp builder.Services.AddOutboxUsingDbContext(mbb => { // Configure outbox options here if needed }); builder.Services.AddSlimMessageBus(mbb => { // Configure message bus with outbox support mbb.UseOutbox(); // Configure child buses, e.g., for publishing events mbb.AddBus(new AzureServiceBusMessageBusSettings("...")) .AddOutbox(); // Configure child buses, e.g., for consuming commands with transactions mbb.AddBus(new MemoryMessageBusSettings()) .UsePostgreSqlTransaction(); // Or UseSqlTransaction() }); // Ensure your DbContext is registered builder.Services.AddDbContext(options => options.UseSqlServer(configuration.GetConnectionString("DefaultConnection"))); ``` -------------------------------- ### Add Outbox Using DbContext for SQL Server Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/plugin_outbox.t.md Use this to add the Outbox.DbContext plugin when integrating with Entity Framework and SQL Server. Ensure the SlimMessageBus.Host.Outbox.Sql.DbContext package is installed. ```csharp using SlimMessageBus.Host.Outbox.Sql.DbContext; ``` -------------------------------- ### Configure Kafka Consumer Offset Commits Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_kafka.md Configure when Kafka consumer offsets are committed using CheckpointEvery for message count or CheckpointAfter for time intervals. This example sets both. ```csharp mbb.Consume(x => { x.Topic(topic) .WithConsumer() .KafkaGroup("subscriber") .Instances(2) .CheckpointEvery(1000) .CheckpointAfter(TimeSpan.FromSeconds(600)); }); ``` -------------------------------- ### Configure MQTT Provider with TCP Server and TLS Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_mqtt.md Configure the MQTT transport provider with connection details for a TCP server, including TLS options and credentials. This setup is essential for establishing a connection to the MQTT broker. ```csharp services.AddSlimMessageBus(mbb => { mbb.WithProviderMqtt(cfg => { cfg.ClientBuilder .WithTcpServer(configuration["Mqtt:Server"], int.Parse(configuration["Mqtt:Port"])) .WithTlsOptions(o => o.UseTls()) .WithCredentials(configuration["Mqtt:Username"], configuration["Mqtt:Password"]) // Use MQTTv5 to use message headers (if the broker supports it) .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500); }); mbb.AddServicesFromAssemblyContaining(); mbb.AddJsonSerializer(); }); ``` -------------------------------- ### Custom Unrecognized Routing Key Handler Logic Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_rabbitmq.md Implement custom logic to handle unrecognized messages based on routing key or message content. This example logs the unrecognized key and routes specific keys to a DLX while acknowledging others. ```cs cfg.MessageUnrecognizedRoutingKeyHandler = (transportMessage) => { var routingKey = transportMessage.RoutingKey; // Log for monitoring and alerting logger.LogWarning("Unrecognized routing key: {RoutingKey} on exchange: {Exchange}", routingKey, transportMessage.Exchange); // Route to DLX if it looks like a potential typo or misconfiguration if (routingKey.StartsWith("orders.")) { return RabbitMqMessageConfirmOptions.Nack; // Send to DLX for investigation } // Ack and discard messages from other exchanges return RabbitMqMessageConfirmOptions.Ack; }; ``` -------------------------------- ### Build and Run Tests Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/Maintainers/build.md Navigate to the src directory and execute the build and test commands. Use the filter to exclude integration tests. ```bash cd src dotnet build dotnet test --filter Category!=Integration ``` ```bash dotnet test ``` ```bash dotnet test --filter Category!=Integration ``` -------------------------------- ### Disable Auto Consumer Start Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/intro.md Set this option to false to prevent consumers from starting automatically when the bus is created. Default is true. ```csharp mbb.AutoStartConsumersEnabled(false); ``` -------------------------------- ### Configure Producer and Consumers with Wildcard Routing Keys Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_rabbitmq.md Demonstrates configuring a producer with a dynamic routing key provider and multiple consumers using wildcard patterns ('*' and '#') for topic exchanges. ```csharp services.AddSlimMessageBus(mbb => { // Producer sends messages with specific routing keys mbb.Produce(x => x .Exchange("regions", exchangeType: ExchangeType.Topic) .RoutingKeyProvider((m, p) => $"regions.{m.Country}.cities.{m.City}")); // Consumer 1: Match all cities in North America mbb.Consume(x => x .Queue("na-cities-queue") .ExchangeBinding("regions", routingKey: "regions.na.cities.*") // * matches exactly one city .WithConsumer()); // Consumer 2: Match all events in the regions exchange mbb.Consume(x => x .Queue("all-regions-queue") .ExchangeBinding("regions", routingKey: "regions.#") // # matches zero or more segments .WithConsumer()); // Consumer 3: Match all audit events with any number of segments mbb.Consume(x => x .Queue("audit-queue") .ExchangeBinding("audit", routingKey: "audit.events.#") .WithConsumer()); // Consumer 4: Complex pattern - match region events ending with specific pattern mbb.Consume(x => x .Queue("region-reports-queue") .ExchangeBinding("regions", routingKey: "regions.*.reports.*") // matches regions.{country}.reports.{type} .WithConsumer()); }); ``` -------------------------------- ### Configure Kafka Provider Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_kafka.md Set up the Kafka provider with a list of brokers and configure producer/consumer specific settings. ```cs services.AddSlimMessageBus(mbb => { // ... mbb.WithProviderKafka(cfg => { cfg.BrokerList = kafkaBrokers; cfg.ProducerConfig = (config) => { // adjust the producer config }; cfg.ConsumerConfig = (config) => { // adjust the consumer config }; }); }); ``` -------------------------------- ### Configure Default Topology Creation Options Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_azure_servicebus.md Set bus-wide default creation options for queues, topics, and subscriptions using delegates. These options are applied before resource creation. ```csharp mbb.WithProviderServiceBus(cfg => { //cfg.ConnectionString = serviceBusConnectionString; cfg.TopologyProvisioning = new ServiceBusTopologySettings { CreateQueueOptions = (options) => { options.EnablePartitioning = true; options.RequiresDuplicateDetection = true; options.DuplicateDetectionHistoryTimeWindow = TimeSpan.FromMinutes(5); }, CreateTopicOptions = (options) => { options.EnablePartitioning = true; options.RequiresDuplicateDetection = true; options.DuplicateDetectionHistoryTimeWindow = TimeSpan.FromMinutes(5); }, CreateSubscriptionOptions = (options) => { options.LockDuration = TimeSpan.FromMinutes(5); }, CreateSubscriptionFilterOptions = (options) => { }, }; }); ``` -------------------------------- ### Git Commit Sign-off Example Source: https://github.com/zarusz/slimmessagebus/blob/master/CONTRIBUTING.md This is an example of the required sign-off line to be added to every git commit message. It certifies that you wrote the code or have the right to pass it on. ```git Signed-off-by: Joe Smith ``` -------------------------------- ### Client Logic for Submitting an Order Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/UseCases/DomainEvents.md Sample logic demonstrating the creation and submission of an order, which in turn raises domain events. ```csharp var john = new Customer("John", "Wick"); var order = new Order(john); order.Add("id_machine_gun", 2); order.Add("id_grenade", 4); await order.Submit(); // events fired here ``` -------------------------------- ### Custom RabbitMQ Topology Initialization Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_rabbitmq.md Use `UseTopologyInitializer` to perform custom cleanup or apply specific topology elements before or instead of the default SlimMessageBus inferred topology. Call `applyDefaultTopology()` to include the default behavior. ```csharp services.AddSlimMessageBus((mbb) => { mbb.WithProviderRabbitMQ(cfg => { cfg.UseTopologyInitializer((channel, applyDefaultTopology) => { // perform some cleanup if needed channel.QueueDelete("subscriber-0", ifUnused: true, ifEmpty: false); channel.QueueDelete("subscriber-1", ifUnused: true, ifEmpty: false); channel.ExchangeDelete("test-ping", ifUnused: true); channel.ExchangeDelete("subscriber-dlq", ifUnused: true); // apply default SMB inferred topology applyDefaultTopology(); }); }); }); ``` -------------------------------- ### Add JSON Serializer to SlimMessageBus Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/serialization.md Install the SlimMessageBus.Host.Serialization.SystemTextJson NuGet package and add the JSON serializer to the bus configuration. ```csharp mbb.AddJsonSerializer(); ``` -------------------------------- ### Enable XML Documentation File Generation Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/plugin_asyncapi.md Configure your project to generate an XML documentation file. This is essential for the AsyncAPI plugin to extract comments and remarks from your code. ```xml true ``` -------------------------------- ### Import SQS Host Package Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_amazon_sqs.md Import the necessary package for using the Amazon SQS transport provider with SlimMessageBus. ```csharp using SlimMessageBus.Host.AmazonSQS; ``` -------------------------------- ### Add AsyncAPI Document Generator to SlimMessageBus Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/plugin_asyncapi.md Register the IDocumentGenerator for the Saunter library during SlimMessageBus setup. This is a required step for enabling AsyncAPI document generation. ```csharp services.AddSlimMessageBus(mbb => { // Register the IDocumentGenerator for Saunter library mbb.AddAsyncApiDocumentGenerator(); }); ``` -------------------------------- ### Consumer DI Resolution Order Diagram Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/intro.md Illustrates the sequence of events from DI scope creation to consumer execution, highlighting when interceptors run and when the consumer instance is resolved. ```text Per-message DI scope created │ ▼ Interceptors resolved from DI │ ▼ IConsumerInterceptor.OnHandle() ← e.g. starts a DB transaction │ (calls next()) ▼ IRequestHandlerInterceptor.OnHandle() │ (calls next()) ▼ Consumer resolved from DI ← scoped constructor deps see the open transaction │ ▼ Consumer.OnHandle() / Handler.OnHandle() │ ▼ Interceptors unwind (commit / rollback) ``` -------------------------------- ### Configure Memory Provider with Asynchronous Publish Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_memory.md Set up the memory message bus provider, enabling asynchronous publishing and JSON serialization. This configuration allows consumers to process messages in the background, decoupling them from the publishing unit of work. ```csharp services.AddSlimMessageBus(mbb => { mbb .WithProviderMemory(cfg => { cfg.EnableMessageSerialization = _enableSerialization; cfg.EnableBlockingPublish = _enableBlockingPublish; }) .AddServicesFromAssemblyContaining() .AddJsonSerializer(); }); ``` -------------------------------- ### Configure Default Session Settings Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_azure_servicebus.md Sets default session configuration values for all consumers that enable sessions. This avoids repetitive configuration within each consumer setup. ```csharp mbb.WithProviderServiceBus(cfg => { //cfg.ConnectionString = ""; cfg.SessionIdleTimeout = TimeSpan.FromSeconds(5); cfg.MaxConcurrentSessions = 10; }); ``` -------------------------------- ### Configure Outbox with Direct SQL Connection Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/plugin_outbox.t.md Configure the SlimMessageBus container to use the Outbox.Sql plugin with a direct SQL connection. This requires registering a `SqlConnection` instance in the service container. The `PollBatchSize` option can be configured. ```csharp builder.Services.AddSlimMessageBus(mbb => { // Alternatively, if we were not using EF, we could use a SqlConnection mbb.AddOutboxUsingSql(opts => { opts.PollBatchSize = 100; }); }); // SMB requires the SqlConnection to be registered in the container builder.Services.AddTransient(svp => { var configuration = svp.GetRequiredService(); var connectionString = configuration.GetConnectionString("DefaultConnection"); return new SqlConnection(connectionString); }); ``` -------------------------------- ### Add Consumers with Type Filter Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/intro.md Enables filtering discovered types during assembly auto-registration based on a predicate. This example filters for types within a specific namespace. ```csharp services.AddSlimMessageBus(mbb => { // Register the found types that contain DomainEventHandlers in the namespace mbb.AddConsumersFromAssembly(Assembly.GetExecutingAssembly(), filter: (type) => type.Namespace.Contains("DomainEventHandlers")); }; ``` -------------------------------- ### Implement a Command Handler Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/plugin_fluent_validation.md Provides a basic command handler for the CreateCustomerCommand. This handler is executed after the command has been successfully validated. ```csharp public class CreateCustomerCommandHandler : IRequestHandler { public Task OnHandle(CreateCustomerCommand command, CancellationToken cancellationToken) { return Task.FromResult(new CreateCustomerCommandResult(Guid.NewGuid())); } } ``` -------------------------------- ### Add Consumers with Custom Lifetime Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/intro.md Allows overriding the default service lifetime for discovered consumers when using assembly auto-registration. This example registers them with a Scoped lifetime. ```csharp services.AddSlimMessageBus(mbb => { // Register the found types as Scoped lifetime in MSDI mbb.AddConsumersFromAssembly(Assembly.GetExecutingAssembly(), consumerLifetime: ServiceLifetime.Scoped); }; ``` -------------------------------- ### Configure Topology Provisioning Permissions Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_azure_servicebus.md Specify which entities (producers, consumers) are allowed to create queues, topics, and subscriptions. This controls the scope of topology provisioning. ```csharp mbb.WithProviderServiceBus(cfg => { //cfg.ConnectionString = serviceBusConnectionString; cfg.TopologyProvisioning = new ServiceBusTopologySettings { Enabled = true, CanProducerCreateQueue = true, // only declared producers will be used to provision queues CanProducerCreateTopic = true, // only declared producers will be used to provision topics CanConsumerCreateQueue = false, // the consumers will not be able to provision a missing queue CanConsumerCreateTopic = false, // the consumers will not be able to provision a missing topic CanConsumerCreateSubscription = true, // but the consumers will add the missing subscription if needed CanConsumerCreateSubscriptionFilter = true, // but the consumers will add the missing filter on subscription if needed }; }); ``` -------------------------------- ### Configure Consumer Settings with FIFO and Max Message Count Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_amazon_sqs.md Configure a consumer with FIFO enabled, maximum message count, and number of instances for an SQS queue. ```csharp mbb.Consume(x => x .WithConsumer() .Queue("some-queue") .EnableFifo() .MaxMessageCount(10) .Instances(1)); ``` -------------------------------- ### Generic Logging Consumer Interceptor Implementation Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/intro.md An example implementation of a generic consumer interceptor that logs information about arriving messages and their processing time. This interceptor can be applied to any message type. ```csharp public class LoggingConsumerInterceptor : IConsumerInterceptor { public async Task OnHandle(TMessage message, Func> next, IConsumerContext context) { _logger.LogInformation("Message of type {MessageType} arrived...", message.GetType()); var stopwatch = Stopwatch.StartNew(); var result = await next(); _logger.LogInformation("Message of type {MessageType} processing took {MessageProcessingTime}", message.GetType(), stopwatch.Elapsed); return result; } } ``` -------------------------------- ### Polymorphic Message Serialization Example Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/serialization.md Demonstrates how `useActualTypeOnSerialize` affects the serialized JSON output for derived message types. Note that System.Text.Json requires `[JsonDerivedType]` attributes for proper deserialization. ```csharp public class BaseMessage { public string Id { get; set; } } public class DerivedMessage : BaseMessage { public string ExtraData { get; set; } } // Producer configuration mbb.Produce(x => x.DefaultTopic("my-topic")); // When publishing var message = new DerivedMessage { Id = "123", ExtraData = "test" }; await bus.Publish(message); // With useActualTypeOnSerialize: true (default) // Serialized JSON: {"id":"123","extraData":"test"} // With useActualTypeOnSerialize: false // Serialized JSON: {"id":"123"} ``` -------------------------------- ### Generate Avro C# Classes Source: https://github.com/zarusz/slimmessagebus/blob/master/src/Samples/Sample.Serialization.MessagesAvro/README.md Execute the gen.ps1 script in the Tools directory to generate C# classes and .avpr files from your .avdl definitions. Ensure you have Java SDK 1.8 installed. ```powershell cd Tools .\gen.ps1 ``` -------------------------------- ### Configure SlimMessageBus with SQL Server and Azure Service Bus Outbox Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/plugin_outbox.md This snippet shows the configuration of SlimMessageBus, including setting up memory and Azure Service Bus child buses. It demonstrates how to enable the outbox pattern for Azure Service Bus and configure SQL Server DbContext for outbox persistence. ```csharp builder.Services.AddSlimMessageBus(mbb => { mbb.PerMessageScopeEnabled(false); mbb .AddChildBus("Memory", mbb => { mbb.WithProviderMemory() .AutoDeclareFrom(Assembly.GetExecutingAssembly(), consumerTypeFilter: t => t.Name.EndsWith("CommandHandler")); //.UseTransactionScope(messageTypeFilter: t => t.Name.EndsWith("Command")) // Consumers/Handlers will be wrapped in a TransactionScope //.UseSqlTransaction(messageTypeFilter: t => t.Name.EndsWith("Command")); // Consumers/Handlers will be wrapped in a SqlTransaction ending with Command switch (dbProvider) { case DbProvider.SqlServer: mbb.UseSqlTransaction(messageTypeFilter: t => t.Name.EndsWith("Command")); // Consumers/Handlers will be wrapped in a SqlTransaction ending with Command break; case DbProvider.PostgreSql: mbb.UsePostgreSqlTransaction(messageTypeFilter: t => t.Name.EndsWith("Command")); // Consumers/Handlers will be wrapped in a SqlTransaction ending with Command break; } }) .AddChildBus("AzureSB", mbb => { mbb .Handle(s => { s.Topic("samples.outbox/customer-events", t => { t.WithHandler() .SubscriptionName("CreateCustomer"); }); }) .WithProviderServiceBus(cfg => { cfg.ConnectionString = Secrets.Service.PopulateSecrets(configuration["Azure:ServiceBus"]); cfg.TopologyProvisioning.CanProducerCreateTopic = true; cfg.TopologyProvisioning.CanConsumerCreateQueue = true; cfg.TopologyProvisioning.CanConsumerReplaceSubscriptionFilters = true; }) .Produce(x => { x.DefaultTopic("samples.outbox/customer-events"); // OR if you want just this producer to sent via outbox // x.UseOutbox(); }) // All outgoing messages from this bus will go out via an outbox .UseOutbox(/* messageTypeFilter: t => t.Name.EndsWith("Command") */); // Additionally, can apply filter do determine messages that should go out via outbox }) .AddServicesFromAssembly(Assembly.GetExecutingAssembly()) .AddJsonSerializer() .AddAspNet(); switch (dbProvider) { case DbProvider.SqlServer: SlimMessageBus.Host.Outbox.Sql.DbContext.MessageBusBuilderExtensions.AddOutboxUsingDbContext(mbb, opts => { opts.PollBatchSize = 500; opts.PollIdleSleep = TimeSpan.FromSeconds(10); opts.MessageCleanup.Interval = TimeSpan.FromSeconds(10); opts.MessageCleanup.Age = TimeSpan.FromMinutes(1); //opts.SqlSettings.TransactionIsolationLevel = System.Data.IsolationLevel.RepeatableRead; //opts.SqlSettings.Dialect = SqlDialect.SqlServer; }); break; case DbProvider.PostgreSql: SlimMessageBus.Host.Outbox.PostgreSql.DbContext.MessageBusBuilderExtensions.AddOutboxUsingDbContext(mbb, opts => { opts.PollBatchSize = 500; opts.PollIdleSleep = TimeSpan.FromSeconds(10); opts.MessageCleanup.Interval = TimeSpan.FromSeconds(10); opts.MessageCleanup.Age = TimeSpan.FromMinutes(1); //opts.SqlSettings.TransactionIsolationLevel = System.Data.IsolationLevel.RepeatableRead; //opts.SqlSettings.Dialect = SqlDialect.SqlServer; }); break; } }); ``` -------------------------------- ### Configure RabbitMQ Request-Response with SlimMessageBus Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_rabbitmq.md Sets up a request-response flow using fanout exchanges. Configure producer exchange, handler queue, bindings, dead-letter exchange, and response handling with timeouts. ```csharp services.AddSlimMessageBus((mbb) => { // ... mbb.Produce(x => { // The requests should be send to "test-echo" exchange x.Exchange("test-echo", exchangeType: ExchangeType.Fanout); }) .Handle(x => x // Declare the queue for the handler .Queue("echo-request-handler") // Bind the queue to the "test-echo" exchange .ExchangeBinding("test-echo") // If the request handling fails, the failed messages will be routed to the DLQ exchange .DeadLetterExchange("echo-request-handler-dlq") .WithHandler()) .ExpectRequestResponses(x => { // Tell the handler to which exchange send the responses to x.ReplyToExchange("test-echo-resp", ExchangeType.Fanout); // Which queue to use to read responses from x.Queue("test-echo-resp-queue"); // Bind to the reply to exchange x.ExchangeBinding(); // Timeout if the response doesn't arrive within 60 seconds x.DefaultTimeout(TimeSpan.FromSeconds(60)); }); }); ``` -------------------------------- ### Redis Transport Lifecycle Hook: OnDatabaseConnected Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_redis.md Implement a hook to execute logic after a connection to the Redis database is established. This example demonstrates clearing specific Redis lists upon connection. ```csharp mbb.WithProviderRedis(cfg => { cfg.OnDatabaseConnected = (database) => { // Upon connect clear the redis list with the specified keys database.KeyDelete("test-echo-queue"); database.KeyDelete("test-echo-queue-resp"); }; }); ``` -------------------------------- ### CreateCustomerCommandHandler with Transactional Outbox Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/plugin_outbox.md This command handler demonstrates how to publish an event using the message bus after successfully saving a customer to the database. The handler is designed to be transactional, ensuring that the database operation and the message publication are atomic. ```csharp public record CreateCustomerCommandHandler(IMessageBus Bus, CustomerContext CustomerContext) : IRequestHandler { public async Task OnHandle(CreateCustomerCommand request, CancellationToken cancellationToken) { // Note: This handler will be already wrapped in a transaction: see Program.cs and .UseTransactionScope() / .UseSqlTransaction() var customer = new Customer(request.Firstname, request.Lastname); await CustomerContext.Customers.AddAsync(customer); await CustomerContext.SaveChangesAsync(); // Announce to anyone outside of this micro-service that a customer has been created (this will go out via an transactional outbox) await Bus.Publish(new CustomerCreatedEvent(customer.Id, customer.Firstname, customer.Lastname)); return customer.Id; } } ``` -------------------------------- ### ASP.NET Minimal API Endpoint Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/plugin_fluent_validation.t.md Example of an ASP.NET Minimal API endpoint that accepts a command and sends it via SlimMessageBus. If producer-side validation is enabled, invalid commands will result in a `ValidationException`. ```csharp // Using minimal APIs var app = builder.Build(); app.MapPost("/customer", (CreateCustomerCommand command, IMessageBus bus) => bus.Send(command)); await app.RunAsync(); ``` -------------------------------- ### Custom Message Type to Topic Converter Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/provider_memory.md Customize how virtual topic names are generated when using AutoDeclareFrom by providing a custom messageTypeToTopicConverter function. This example uses the message type's FullName. ```csharp services.AddSlimMessageBus(builder => builder.WithProviderMemory() .AutoDeclareFrom(Assembly.GetExecutingAssembly(), messageTypeToTopicConverter: messageType => messageType.FullName) ); ``` -------------------------------- ### Accessing Consumer Context via Constructor Injection Source: https://github.com/zarusz/slimmessagebus/blob/master/docs/intro.md The recommended way to access IConsumerContext is through constructor injection. This provides access to message path, headers, and transport-specific metadata like Kafka partitions. ```csharp public class PingConsumer(IConsumerContext context) : IConsumer { public Task OnHandle(PingMessage message, CancellationToken cancellationToken) { var path = context.Path; // topic or queue name var headers = context.Headers; // message headers // Example: Kafka-specific metadata (requires SlimMessageBus.Host.Kafka) var transportMessage = context.GetTransportMessage(); var partition = transportMessage.TopicPartition.Partition; return Task.CompletedTask; } } ```