SlimMessageBus

https://github.com/zarusz/slimmessagebus
Admin
Lightweight message bus interface for .NET (pub/sub and request-response) with transport plugins for...
Tokens:63,771
Snippets:823
Trust Score:8.3
License:Apache-2.0
Update:3 months ago
Show doc for...
Tokens:
Raw
### 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<CustomerContext>(...)
```

--------------------------------

### 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<Program>();
});

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<string>;

public class PingRequestHandler : IRequestHandler<PingRequest, string>
{
    public Task<string> 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<PingConsumer>();
    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<string>;

public class PingRequestHandler : IRequestHandler<PingRequest, string>
{
    public async Task<string> 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<SomeMessage>(x => x
      .Topic("some-topic")
      .WithConsumer<SomeMessageConsumer>()
      .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<EchoRequest, EchoResponse>
{
   public Task<EchoResponse> 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<MultiplyRequest>(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<OrderEvent>(x => x
    .Queue("orders-queue")
    .ExchangeBinding("orders-exchange", routingKey: "orders.created")
    .WithConsumer<OrderCreatedConsumer>());
```

--------------------------------

### 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<OrderEvent>(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<OrderCreatedConsumer>());
});
```

--------------------------------

### 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<SomeRequest, Void>
{
   public async Task<Void> OnHandle(TRequest request, Func<Task<Void>> 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
/// <summary>
/// Event when a customer is created within the domain.
/// </summary>
/// <param name="Id"></param>
/// <param name="Firstname"></param>
/// <param name="Lastname"></param>
public record CustomerCreatedEvent(Guid Id, string Firstname, string Lastname) : CustomerEvent(Id);

public class CustomerCreatedEventConsumer : IConsumer<CustomerCreatedEvent>
{
    /// <summary>
    /// Upon the <see cref="CustomerCreatedEvent"/> will store it with the database.
    /// </summary>
    /// <param name="message"></param>
    /// <returns></returns>
    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<OrderCreatedEvent>(x => x.DefaultPath("orders-topic"))
               .Consume<OrderCreatedEvent>(x => x.Path("orders-topic")
                  //.WithConsumer<OrderCreatedEventConsumer>() // Optional: can be skipped as IConsumer<OrderCreatedEvent> 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<CreateCustomerCommand>(x => x.DefaultPath("customer-requests"))
               .Handle<CreateCustomerCommand, CreateCustomerCommandResult>(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<OrderCreatedEventConsumer>();
});

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<CustomerContext>(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<AzureServiceBusMessageBusSettings>(new AzureServiceBusMessageBusSettings("..."))
        .AddOutbox();

    // Configure child buses, e.g., for consuming commands with transactions
    mbb.AddBus<MemoryMessageBusSettings>(new MemoryMessageBusSettings())
        .UsePostgreSqlTransaction(); // Or UseSqlTransaction()
});

// Ensure your DbContext is registered
builder.Services.AddDbContext<CustomerContext>(options =>
    options.UseSqlServer(configuration.GetConnectionString("DefaultConnection")));
```

--------------------------------

### 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<PingMessage>(x =>
{
    x.Topic(topic)
        .WithConsumer<PingConsumer>()
        .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<PingConsumer>();
    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;
};
```

--------------------------------

### 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<RegionEvent>(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<RegionEvent>(x => x
        .Queue("na-cities-queue")
        .ExchangeBinding("regions", routingKey: "regions.na.cities.*")  // * matches exactly one city
        .WithConsumer<NorthAmericaCitiesConsumer>());

    // Consumer 2: Match all events in the regions exchange
    mbb.Consume<RegionEvent>(x => x
        .Queue("all-regions-queue")
        .ExchangeBinding("regions", routingKey: "regions.#")  // # matches zero or more segments
        .WithConsumer<AllRegionsConsumer>());

    // Consumer 3: Match all audit events with any number of segments
    mbb.Consume<AuditEvent>(x => x
        .Queue("audit-queue")
        .ExchangeBinding("audit", routingKey: "audit.events.#")
        .WithConsumer<AuditConsumer>());

    // Consumer 4: Complex pattern - match region events ending with specific pattern
    mbb.Consume<RegionEvent>(x => x
        .Queue("region-reports-queue")
        .ExchangeBinding("regions", routingKey: "regions.*.reports.*")  // matches regions.{country}.reports.{type}
        .WithConsumer<RegionReportsConsumer>());
});
```

--------------------------------

### 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);
```

--------------------------------

### 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
```

--------------------------------

### 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 <joe.smith@email.com>
```

--------------------------------

### 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
<PropertyGroup>
  <GenerateDocumentationFile>true</GenerateDocumentationFile>
</PropertyGroup>
```

--------------------------------

### 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<T>.OnHandle()   ← e.g. starts a DB transaction
         │  (calls next())
         ▼
 IRequestHandlerInterceptor<TReq,TResp>.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<PingConsumer>()
        .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<IConfiguration>();
    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<CreateCustomerCommand, CreateCustomerCommandResult>
{
    public Task<CreateCustomerCommandResult> 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
   };
});
```

--------------------------------

### 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 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<TMessage>(x => x
   .WithConsumer<TConsumer>()
   .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<TMessage> : IConsumerInterceptor<TMessage>
{
   public async Task<object> OnHandle(TMessage message, Func<Task<object>> 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<BaseMessage>(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"}
```

--------------------------------

### 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<CreateCustomerCommand, Guid>(s =>
                {
                    s.Topic("samples.outbox/customer-events", t =>
                    {
                        t.WithHandler<CreateCustomerCommandHandler, CreateCustomerCommand>()
                            .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<CustomerCreatedEvent>(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<CustomerContext>(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<CustomerContext>(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;
    }
});
```

--------------------------------

### 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 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<EchoRequest>(x =>
    {
        // The requests should be send to "test-echo" exchange
        x.Exchange("test-echo", exchangeType: ExchangeType.Fanout);
    })
    .Handle<EchoRequest, EchoResponse>(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<EchoRequestHandler>()) 
    .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<CreateCustomerCommand, Guid>
{
    public async Task<Guid> 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<PingMessage>
{
    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;
    }
}
```
Tokens:
Raw