### Start Docker Compose

Source: https://github.com/beagle1984/silverback/blob/master/samples/README.md

Start the docker compose file in the repository root.

```bash
cd ..
docker-compose up -d
```

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

### Add MQTT Integration Package

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/connecting.md

Install the Silverback MQTT integration package using the .NET CLI.

```powershell
dotnet add package Silverback.Integration.MQTT
```

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

### Quick Kafka Integration Example

Source: https://github.com/beagle1984/silverback/blob/master/README.md

This snippet demonstrates how to configure Silverback for Kafka integration, including adding Kafka clients, specifying bootstrap servers, and configuring producers and consumers for a specific topic.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddKafka())
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddProducer(producer => producer
            .Produce<MyMessage>(endpoint => endpoint.ProduceTo("my-topic")))
        .AddConsumer(consumer => consumer
            .Consume<MyMessage>(endpoint => endpoint.ConsumeFrom("my-topic"))));
```

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

### Consumer Management Service Example

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/connecting.md

This C# service demonstrates how to interact with the IConsumerCollection to get total consumed messages, start/stop individual consumers, and restart disconnected consumers.

```csharp
public class ConsumerManagementService
{
    private readonly IConsumerCollection _consumers;

    public ConsumerManagementService(IConsumerCollection consumers)
    {
        _consumers = consumers;
    }

    public int GetTotalConsumedMessages()
    {
        int totalCount = 0;

        foreach (IConsumer in _consumers
        {
            totalCount += consumer.StatusInfo.ConsumedMessagesCount;
        }
        
        return totalCount;
    }
    
    public async ValueTask Start(string consumerName)
    {
        await _consumers[consumerName].StartAsync();
    }
    
    public async ValueTask Stop(string consumerName)
    {
        await _consumers[consumerName].StopAsync();
    }

    public async ValueTask RestartDisconnectedConsumers()
    {
        foreach (IConsumer consumer in _consumers)
        {
            if (consumer.Client.Status == ClientStatus.Disconnected)
            {
                await consumer.Client.ConnectAsync();
            }
        }
    }
}
```

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

### MyClientsConfigurator Implementation for Kafka

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/connecting.md

Example implementation of IBrokerClientsConfigurator for configuring Kafka producers and consumers.

```csharp
public class MyClientsConfigurator : IBrokerClientsConfigurator
{
    public void Configure(BrokerClientsConfigurationBuilder builder)
    {
        builder.AddKafkaClients(clients => clients
            .WithBootstrapServers("PLAINTEXT://localhost:9092")
            .AddProducer(producer => producer
                .Produce<MyMessage>(endpoint => endpoint
                    .ProduceTo("my-topic")))
            .AddConsumer(consumer => consumer
                .WithGroupId("consumer1")
                .AutoResetOffsetToEarliest()
                .Consume<MyOtherMessage>(endpoint => endpoint
                    .ConsumeFrom("my-other-topic")
                    .EnableBatchProcessing(100, TimeSpan.FromSeconds(5))
                    .OnError(policy => policy.Retry(3).ThenSkip()))));
    }
}
```

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

### Add Kafka Integration Package

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/connecting.md

Install the Silverback Kafka integration package using the .NET CLI.

```powershell
dotnet add package Silverback.Integration.Kafka
```

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

### Provisioning Kafka Offset Store Tables

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/consuming/kafka/offset.md

Initialize the necessary tables for Kafka offset storage using SilverbackStorageInitializer. Examples are provided for SQLite and PostgreSQL.

```csharp
await storageInitializer.CreateSqliteKafkaOffsetStoreAsync(connectionString);
// or
await storageInitializer.CreatePostgreSqlKafkaOffsetStoreAsync(connectionString);
```

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

### Enable Newtonsoft.Json Serializer

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/producing/serialization.md

Use the Newtonsoft.Json serializer for producing messages. Ensure the Silverback.Newtonsoft package is installed.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddKafka())
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddProducer("producer1", producer => producer
            .Produce<MyMessage>("endpoint1", endpoint => endpoint
                .ProduceTo("my-topic")
                .SerializeAsJsonUsingNewtonsoft())));
```

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

### Static Kafka Partition Assignment from Topic Metadata

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/consuming/kafka/subscription.md

Dynamically select Kafka partitions based on topic metadata using LINQ expressions. This example consumes even-numbered partitions starting from the beginning.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddKafka())
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddConsumer("consumer1", consumer => consumer
            .Consume<MyMessage>("endpoint1", endpoint => endpoint
                .ConsumeFrom(
                    "my-topic",
                    partitions => partitions
                        .Where(p => p.Partition % 2 == 0)
                        .Select(p => new TopicPartitionOffset(p, Offset.Beginning))))));
```

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

### Start Confluent Control Center Docker Container

Source: https://github.com/beagle1984/silverback/blob/master/Kafka.md

Launches the Confluent Control Center container, configuring it to connect to Zookeeper and Kafka.

```bash
docker run -d --name=control-center --net=confluent --ulimit nofile=16384:16384 -p 9021:9021 -v /tmp/control-center/data:/var/lib/confluent-control-center -e CONTROL_CENTER_ZOOKEEPER_CONNECT=zookeeper:2181 -e CONTROL_CENTER_BOOTSTRAP_SERVERS=kafka:9092 -e CONTROL_CENTER_REPLICATION_FACTOR=1 -e CONTROL_CENTER_MONITORING_INTERCEPTOR_TOPIC_PARTITIONS=1 -e CONTROL_CENTER_INTERNAL_TOPICS_PARTITIONS=1 -e CONTROL_CENTER_STREAMS_NUM_STREAM_THREADS=2 -e CONTROL_CENTER_CONNECT_CLUSTER=http://kafka-connect:8082 confluentinc/cp-enterprise-control-center:5.0.1
```

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

### Custom Partitioning with Partition Selector

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/producing/kafka/partitioning.md

Gain explicit control over partition selection by using the `ProduceTo` overload that accepts a partition selector. This example routes messages based on a modulo operation on the message ID.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddKafka())
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddProducer("producer1", producer => producer
            .Produce<MyMessage>("endpoint1", endpoint => endpoint
                .ProduceTo("my-topic", message => message.Id % 3))));
```

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

### Configure Kafka Connection Mode to AfterStartup

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/connecting.md

Sets the Kafka client connection mode to be established after the application has started.

```csharp
services
    .AddSilverback()
    .WithConnectionToMessageBroker(options => options
        .AddKafka()
        .WithConnectionOptions(new BrokerClientConnectionOptions
        {
            Mode = BrokerClientConnectionMode.AfterStartup
        }));
```

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

### Enlist Database Transaction with Publisher

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/producing/outbox.md

Manually enlist the database transaction with the publisher to ensure atomic updates. This example uses Entity Framework but applies to other database access libraries.

```csharp
await using (IDbContextTransaction transaction = await dbContext.Database.BeginTransactionAsync())
{
    publisher.EnlistDbTransaction(transaction.GetDbTransaction());

    await publisher.PublishAsync(...);
    await publisher.PublishAsync(...);
    await publisher.PublishAsync(...);

    await transaction.CommitAsync();
}
```

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

### Start Kafka Container

Source: https://github.com/beagle1984/silverback/wiki/Kafka-deployment-on-Docker

Launches a Kafka container in detached mode, connecting to Zookeeper and exposing Kafka on port 9092. Requires Zookeeper to be running.

```bash
docker run -d --net=confluent --name=kafka -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092 -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 -p 9092:9092 confluentinc/cp-kafka:5.0.1
```

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

### Implementing a Sorted Behavior

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/bus.md

Example of a behavior that implements ISorted to control its execution order within the pipeline. The SortIndex property determines the order.

```csharp
public class SortedBehavior : IBehavior, ISorted
{
    public int SortIndex => 120;
    
    public Task<IReadOnlyCollection<object?>> HandleAsync(
        object message,
        MessageHandler next)
    {
        return next(message);
    }
}
```

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

### Example appsettings.json for Kafka Configuration

Source: https://github.com/beagle1984/silverback/wiki/External-Configuration

Define Kafka-specific inbound and outbound endpoints, including consumer and producer configurations, error policies, and retry mechanisms. This JSON structure is read by Silverback's configuration system.

```json
{
  "Silverback": {
    "Using": [ "Silverback.Integration.Kafka" ],
    "Inbound": [
      {
        "Endpoint": {
          "Type": "KafkaConsumerEndpoint",
          "Name": "catalog-events",
          "Configuration": {
            "BootstrapServers": "PLAINTEXT://kafka:9092",
            "ClientId": "basket-service",
            "AutoOffsetReset": "Earliest"
          }
        },
        "ErrorPolicies": [
          {
            "Type": "Retry",
            "MaxFailedAttempts": 5,
            "DelayIncrement": "00:00:30"
          },
          {
            "Type": "Move",
            "Endpoint": {
              "Type": "KafkaProducerEndpoint",
              "Name": "basket-failedmessages"
            }
          }
        ]
      }
    ],
    "Outbound": [
      {
        "MessageType": "IIntegrationEvent",
        "Endpoint": {
          "Type": "KafkaProducerEndpoint",
          "Name": "basket-events",
          "Configuration": {
            "BootstrapServers": "PLAINTEXT://kafka:9092",
            "ClientId": "basket-service"
          }
        }
      }
    ]
  }
}
```

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

### Configure Kafka Key in Endpoint

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/producing/kafka/partitioning.md

Define the Kafka key in the endpoint configuration using `SetKafkaKey` for consistent routing. This example shows how to set up a producer with a specific bootstrap server and topic.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddKafka())
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddProducer("producer1", producer => producer
            .Produce<MyMessage>("endpoint1", endpoint => endpoint
                .ProduceTo("my-topic")
                .SetKafkaKey(message => message.Id))));
```

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

### Enable MQTT Batch Processing

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/consuming/batch-processing.md

Enable batch processing for MQTT consumers by calling EnableBatchProcessing on the endpoint configuration. This example sets a batch size of 1000 messages and a timeout of 30 seconds.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddMqtt())
    .AddMqttClients(clients => clients
        .ConnectViaTcp("localhost")
        .AddClient("my-client", client => client
            .WithClientId("client1")
            .Consume<MyMessage>("endpoint1", endpoint => endpoint
                .ConsumeFrom("messages/my")
                .EnableBatchProcessing(1000, TimeSpan.FromSeconds(30))));
```

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

### Configure Entity Framework Kafka Offset Store

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/consuming/kafka/offset.md

Integrate Silverback's Kafka offset storage with Entity Framework. This example shows how to configure the DbContext and register the EF offset store. Ensure your AppDbContext includes the SilverbackStoredOffset DbSet.

```csharp
services
    .AddDbContextFactory<AppDbContext>(options => options.UseNpgsql(connectionString))
    .AddSilverback()
    .WithConnectionToMessageBroker(options => options
        .AddKafka()
        .AddEntityFrameworkKafkaOffsetStore())
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddConsumer(consumer => consumer
            .WithGroupId("my-group")
            .DisableOffsetsCommit()
            .StoreOffsetsClientSide(offsetStore => offsetStore.UseEntityFramework<AppDbContext>())
            .Consume(endpoint => endpoint
                .ConsumeFrom("my-topic"))));

public class AppDbContext : DbContext
{
    // ...your entities...

    public DbSet<SilverbackStoredOffset> KafkaOffsets { get; set; } = null!;
}
```

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

### Consumer Startup Configuration

Source: https://github.com/beagle1984/silverback/blob/master/docs/samples/mqtt/binaryfile-streaming.md

Configures the startup services for the MQTT consumer application.

```csharp
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.Hosting;

namespace Silverback.Samples.Mqtt.BinaryFileStreaming.Consumer
{
    public class Startup
    {
        public static void Main(string[] args)
        {
            CreateHostBuilder(args).Build().Run();
        }

        public static IHostBuilder CreateHostBuilder(string[] args) =>
            Host.CreateDefaultBuilder(args)
                .ConfigureWebHostDefaults(webBuilder =>
                {
                    webBuilder.UseStartup<Startup>();
                });

        public void ConfigureServices(Microsoft.Extensions.DependencyInjection.IServiceCollection services)
        {
            services.AddControllers();
            services.AddSilverback();
        }

        public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
        {
            app.UseRouting();
            app.UseEndpoints(endpoints => endpoints.MapControllers());
        }
    }
}
```

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

### Producer Startup Configuration

Source: https://github.com/beagle1984/silverback/blob/master/docs/samples/mqtt/binaryfile-streaming.md

Configures the startup services for the MQTT producer application.

```csharp
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.Hosting;

namespace Silverback.Samples.Mqtt.BinaryFileStreaming.Producer
{
    public class Startup
    {
        public static void Main(string[] args)
        {
            CreateHostBuilder(args).Build().Run();
        }

        public static IHostBuilder CreateHostBuilder(string[] args) =>
            Host.CreateDefaultBuilder(args)
                .ConfigureWebHostDefaults(webBuilder =>
                {
                    webBuilder.UseStartup<Startup>();
                });

        public void ConfigureServices(Microsoft.Extensions.DependencyInjection.IServiceCollection services)
        {
            services.AddControllers();
            services.AddSilverback();
        }

        public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
        {
            app.UseRouting();
            app.UseEndpoints(endpoints => endpoints.MapControllers());
        }
    }
}
```

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

### Custom Producer Behavior Example

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/producing/behaviors.md

Implement a custom producer behavior by inheriting from IProducerBehavior and overriding the HandleAsync method. This example adds a custom header to each outbound message.

```csharp
public class CustomHeadersProducerBehavior : IProducerBehavior
{
    public int SortIndex => 1000;

    public async ValueTask HandleAsync(
        ProducerPipelineContext context,
        ProducerBehaviorHandler next,
        CancellationToken cancellationToken)
    {
        context.Envelope.Headers.Add("generated-by", "silverback");

        await next(context);
    }
}
```

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

### Consumer Startup Configuration

Source: https://github.com/beagle1984/silverback/blob/master/docs/samples/kafka/basic.md

Configures the consumer application, including Silverback and Kafka settings.

```csharp
using Microsoft.Extensions.DependencyInjection;
using Silverback.Messaging.Configuration;
using Silverback.Messaging.Messages;
using System;
using System.Threading.Tasks;

namespace Basic.Consumer
{
    public class Startup
    {
        public void ConfigureServices(IServiceCollection services)
        {
            services.AddSilverback(
                    (options, builder) => builder
                        .WithConnectionToMessageBroker(broker => broker.AddKafka())
                        .AddSingleton<SampleMessageSubscriber>()
                );
        }

        public void Configure()
        {
            // Not needed for this sample.
        }
    }
}
```

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

### Consumer Startup Configuration

Source: https://github.com/beagle1984/silverback/blob/master/docs/samples/mqtt/basic.md

Configures the startup process for the MQTT consumer. This sets up the necessary services for receiving and processing messages.

```csharp
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.DependencyInjection;
using Silverback.Messaging.Configuration;
using Silverback.Samples.Mqtt.Basic.Common;

namespace Silverback.Samples.Mqtt.Basic.Consumer
{
    public class Startup
    {
        public void ConfigureServices(IServiceCollection services)
        {
            services.AddSilverback(options => {
                options.UseMqtt();
            });
        }

        public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
        {
            app.UseRouting();

            app.UseEndpoints(endpoints => {
                endpoints.MapGet("/", async context => {
                    await context.Response.WriteAsync("Consumer is running!");
                });
            });
        }
    }
}
```

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

### Consumer Startup Configuration

Source: https://github.com/beagle1984/silverback/blob/master/docs/samples/mqtt/basic-v3.md

Configures the consumer application startup, including services and endpoints.

```csharp
using Microsoft.Extensions.DependencyInjection;
using Silverback.Messaging.Configuration;
using System;

namespace Basic.ConsumerV3
{
    public class Startup
    {
        public void ConfigureServices(IServiceCollection services)
        {
            services.AddSilverback(
                    configure: adv => adv.WithConnectionToMessageBroker(
                        "mqtt://localhost:1883",
                        configure: mqtt => mqtt.UseMqtt311()
                    )
                )
                .WithLoopbackQueue(QueueType.Queue);
        }
    }
}
```

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

### Start Zookeeper Container

Source: https://github.com/beagle1984/silverback/wiki/Kafka-deployment-on-Docker

Launches a Zookeeper container in detached mode on the 'confluent' network. This is a prerequisite for Kafka.

```bash
docker run -d --net=confluent --name=zookeeper -e ZOOKEEPER_CLIENT_PORT=2181 confluentinc/cp-zookeeper:5.0.1
```

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

### Producer Startup Configuration

Source: https://github.com/beagle1984/silverback/blob/master/docs/samples/mqtt/basic.md

Configures the startup process for the MQTT producer. This includes setting up necessary services and configurations for message publishing.

```csharp
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.DependencyInjection;
using Silverback.Messaging.Configuration;
using Silverback.Samples.Mqtt.Basic.Common;

namespace Silverback.Samples.Mqtt.Basic.Producer
{
    public class Startup
    {
        public void ConfigureServices(IServiceCollection services)
        {
            services.AddSilverback(options => {
                options.UseMqtt();
            });
        }

        public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
        {
            app.UseRouting();

            app.UseEndpoints(endpoints => {
                endpoints.MapGet("/", async context => {
                    await context.Response.WriteAsync("Producer is running!");
                });
            });
        }
    }
}
```

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

### Create a Mocked MQTT Producer on-the-fly

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/testing.md

Create an MQTT producer directly using `IMqttTestingHelper.GetProducer` to simulate external message publishing to a specific topic.

```csharp
var producer = Helper.GetProducer(
    config => config
        .WithClientId("external-producer")
        .ProduceTo("test/topic"));

await producer.ProduceAsync(new SomeMessage());
```

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

### Configure MQTT Clients and Endpoints

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/connecting.md

Set up MQTT clients, including connection details, topics, QoS levels, and last will messages.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddMqtt())
    .AddMqttClients(clients => clients
        .ConnectViaTcp("localhost")
        .AddClient(client => client
            .WithClientId("my.client")
            .Produce<MyMessage>(endpoint => endpoint
                .ProduceTo("messages/my")
                .WithAtLeastOnceQoS()
                .Retain()
                .IgnoreNoMatchingSubscribersError())
            .Consume(endpoint => endpoint
                .ConsumeFrom("messages/other")
                    .WithAtLeastOnceQoS()
                    .OnError(policy => policy.Skip()))
            .SendLastWillMessage<TestamentMessage>(lastWill => lastWill
                .SendMessage(new TestamentMessage(){ ... })
                .ProduceTo("testaments"))));
```

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

### Run MQTT Basic Consumer

Source: https://github.com/beagle1984/silverback/blob/master/samples/README.md

Run the basic MQTT consumer application.

```bash
dotnet run --project ./samples/MQTT/Basic.Consumer/.
```

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

### Check Control Center Container Status

Source: https://github.com/beagle1984/silverback/blob/master/Kafka.md

Filters and displays log messages from the Control Center container to confirm it has started.

```bash
docker logs control-center | grep Started
```

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

### MQTT Test-Only Configuration

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/testing.md

Configures Silverback with a mocked MQTT broker for testing. This is suitable for tests where you control the ServiceCollection setup.

```csharp
services
    .AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddMockedMqtt())
    .AddMqttClients(
        clients => clients
            .AddProducer(
                producer => producer
                    .Produce<SomeMessage>(endpoint => endpoint.ProduceTo("test/topic")))
            .AddConsumer(
                consumer => consumer
                    .Consume(endpoint => endpoint.ConsumeFrom("test/topic"))));
```

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

### Run Kafka Basic Consumer

Source: https://github.com/beagle1984/silverback/blob/master/samples/README.md

Run the basic Kafka consumer application.

```bash
dotnet run --project ./samples/Kafka/Basic.Consumer/.
```

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

### Producer Startup Configuration

Source: https://github.com/beagle1984/silverback/blob/master/docs/samples/kafka/basic.md

Configures the producer application, including Silverback and Kafka settings.

```csharp
using Microsoft.Extensions.DependencyInjection;
using Silverback.Messaging.Configuration;
using Silverback.Messaging.Messages;
using System;
using System.Threading.Tasks;

namespace Basic.Producer
{
    public class Startup
    {
        public void ConfigureServices(IServiceCollection services)
        {
            services.AddSilverback(
                    (options, builder) => builder
                        .WithConnectionToMessageBroker(broker => broker.AddKafka())
                        .AddSingleton<ProducerBackgroundService>()
                );
        }

        public void Configure()
        {
            // Not needed for this sample.
        }
    }
}
```

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

### Run MQTT Basic Producer

Source: https://github.com/beagle1984/silverback/blob/master/samples/README.md

Run the basic MQTT producer application.

```bash
dotnet run --project ./samples/MQTT/Basic.Producer/.
```

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

### Configure StringMessage Consumer

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/consuming/deserialization.md

Example of configuring a Kafka consumer to consume StringMessage. The string deserializer is used by default for StringMessage or derived types.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddKafka())
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddConsumer("consumer1", consumer => consumer
            .Consume<StringMessage<MyMessage>>("endpoint1", endpoint => endpoint
                .ConsumeFrom("my-topic"))));
```

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

### Configure Kafka Clients and Endpoints

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/connecting.md

Set up Kafka producers and consumers, including bootstrap servers, topics, and error handling policies.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddKafka())
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddProducer(producer => producer
            .Produce<MyMessage>(endpoint => endpoint
                .ProduceTo("my-topic"))
            .Produce<MyMessage2>(endpoint => endpoint
                .ProduceTo("my-topic-2")))
        .AddConsumer(consumer => consumer
            .WithGroupId("consumer1")
            .AutoResetOffsetToEarliest()
            .Consume<MyMessage3>(endpoint => endpoint
                .ConsumeFrom("my-topic-3")
                .OnError(policy => policy.Retry(3).ThenSkip()))
            .Consume<MyMessage4>(endpoint => endpoint
                .ConsumeFrom("my-topic-4")
                .EnableBatchProcessing(100, TimeSpan.FromSeconds(5))
                .OnError(policy => policy.Skip()))));
```

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

### Implement Custom Consumer Behavior

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/consuming/behaviors.md

Example of a custom consumer behavior implementing the IConsumerBehavior interface. The HandleAsync method is invoked for each consumed message.

```csharp
public class CustomConsumerBehavior : IConsumerBehavior
{
    public int SortIndex => 300;

    public async ValueTask HandleAsync(
        ConsumerPipelineContext context,
        ConsumerBehaviorHandler next,
        CancellationToken cancellationToken)
    {
        // ...

        await next(context);
    }
}
```

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

### Producer Startup Configuration

Source: https://github.com/beagle1984/silverback/blob/master/docs/samples/mqtt/basic-v3.md

Configures the producer application startup, including services and endpoints.

```csharp
using Microsoft.Extensions.DependencyInjection;
using Silverback.Messaging.Configuration;
using System;

namespace Basic.ProducerV3
{
    public class Startup
    {
        public void ConfigureServices(IServiceCollection services)
        {
            services.AddSilverback(
                    configure: adv => adv.WithConnectionToMessageBroker(
                        "mqtt://localhost:1883",
                        configure: mqtt => mqtt.UseMqtt311()
                    )
                )
                .WithLoopbackQueue(QueueType.Queue);
        }
    }
}
```

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

### Implementing a Custom Tracing Behavior

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/bus.md

Example of a custom behavior that implements tracing logic for message processing. Behaviors allow for custom middleware-like pipelines.

```csharp
public class TracingBehavior : IBehavior
{
    private readonly ITracer _tracer;

    public TracingBehavior(ITracer tracer)
    {
        _tracer = tracer;
    }
    
    public async Task<IReadOnlyCollection<object?>> HandleAsync(
        object message,
        MessageHandler next)
    {
        try
        {
            _tracer.TraceProcessing(message);
            object result = await next(message);
            _tracer.TraceProcessed(message);
            return result;
        }
        catch (Exception ex)
        {
            _tracer.TraceError(message, ex);
            throw;
        }
    }
}
```

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

### Consumer Startup Configuration

Source: https://github.com/beagle1984/silverback/blob/master/docs/samples/kafka/batch.md

Configures the consumer application services, including Kafka integration and batch processing.

```csharp
using Microsoft.Extensions.DependencyInjection;
using Silverback.Messaging.Configuration;
using System.Threading.Tasks;

namespace Batch.Consumer
{
    public class Startup
    {
        public Task ConfigureServices(IServiceCollection services)
        {
            services
                .AddSilverback(
                    (opts, provider) => opts.UseKafka(
                        "kafka://localhost:9092"));

            return Task.CompletedTask;
        }
    }
}
```

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

### Initialize SQLite Outbox and Kafka Offset Store

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/storage.md

Use SilverbackStorageInitializer to programmatically create tables for SQLite outbox and Kafka offset storage. Requires a connection string.

```csharp
await using var services = serviceCollection.BuildServiceProvider();
var initializer = new SilverbackStorageInitializer(services);

await initializer.CreateSqliteOutboxAsync(connectionString);
await initializer.CreateSqliteKafkaOffsetStoreAsync(connectionString);
```

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

### Consumer Startup Configuration

Source: https://github.com/beagle1984/silverback/blob/master/docs/samples/kafka/binaryfile-streaming.md

Configures the startup services for the Kafka consumer application.

```csharp
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.DependencyInjection;
using Silverback.Messaging.Configuration;

namespace BinaryFileStreaming.Consumer
{
    public class Startup
    {
        public void ConfigureServices(IServiceCollection services)
        {
            services.AddControllers();
            services.AddSilverback(
                $
```

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

### Configure StringMessage Consumer with Custom Encoding

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/consuming/deserialization.md

Example of configuring a Kafka consumer to consume StringMessage with a specific encoding (Unicode). The default encoding is UTF-8.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddKafka())
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddConsumer("consumer1", consumer => consumer
            .Consume<StringMessage<MyMessage>>("endpoint1", endpoint => endpoint
                .ConsumeFrom("my-topic")
                .ConsumeStrings(deserializer => deserializer
                    .WithEncoding(MessageEncoding.Unicode))))));
```

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

### Configure RawMessage Consumer

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/consuming/deserialization.md

Example of configuring a Kafka consumer to consume RawMessage. The raw deserializer is used by default when consuming RawMessage or derived types.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddKafka())
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddConsumer("consumer1", consumer => consumer
            .Consume<RawMessage<MyMessage>>("endpoint1", endpoint => endpoint
                .ConsumeFrom("my-topic"))));
```

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

### Initialize PostgreSQL Outbox, Kafka Offset Store, and Lock Table

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/storage.md

Use SilverbackStorageInitializer to programmatically create tables for PostgreSQL outbox, Kafka offset storage, and the locks table for table-based locking. Requires a connection string.

```csharp
await initializer.CreatePostgreSqlOutboxAsync(connectionString);
await initializer.CreatePostgreSqlKafkaOffsetStoreAsync(connectionString);
await initializer.CreatePostgreSqlLocksTableAsync(connectionString);
```

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

### Configure SQLite Outbox and Worker

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/producing/outbox.md

Configure Silverback to use the SQLite outbox and outbox worker. This setup is for applications using SQLite for data access.

```csharp
services
    .AddSilverback()
    .WithConnectionToMessageBroker(options => options
        .AddKafka()
        .AddSqliteOutbox()
        .AddOutboxWorker(worker => worker
            .ProcessOutbox(outbox => outbox
                .UseSqlite(connectionString)))))
```

```csharp
.AddKafkaClients(clients => clients
    .WithBootstrapServers("PLAINTEXT://localhost:9092")
    .AddProducer("producer1", producer => producer
        .Produce<MyMessage>("endpoint1", endpoint => endpoint
            .ProduceTo("my-topic")
            .StoreToOutbox(outbox => outbox.UseSqlite(connectionString)))));
```

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

### Producer Startup Configuration

Source: https://github.com/beagle1984/silverback/blob/master/docs/samples/kafka/batch.md

Configures the producer application services, including Kafka integration.

```csharp
using Microsoft.Extensions.DependencyInjection;
using Silverback.Messaging.Configuration;
using System.Threading.Tasks;

namespace Batch.Producer
{
    public class Startup
    {
        public Task ConfigureServices(IServiceCollection services)
        {
            services
                .AddSilverback(
                    (opts, provider) => opts.UseKafka(
                        "kafka://localhost:9092"));

            return Task.CompletedTask;
        }
    }
}
```

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

### Configure PostgreSQL Outbox and Worker

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/producing/outbox.md

Configure Silverback to use the PostgreSQL outbox and outbox worker. This setup is for applications using PostgreSQL for data access.

```csharp
services
    .AddSilverback()
    .WithConnectionToMessageBroker(options => options
        .AddKafka()
        .AddPostgreSqlOutbox()
        .AddOutboxWorker(worker => worker
            .ProcessOutbox(outbox => outbox
                .UsePostgreSql(connectionString)))))
```

```csharp
.AddKafkaClients(clients => clients
    .WithBootstrapServers("PLAINTEXT://localhost:9092")
    .AddProducer("producer1", producer => producer
        .Produce<MyMessage>("endpoint1", endpoint => endpoint
            .ProduceTo("my-topic")
            .StoreToOutbox(outbox => outbox
                .UsePostgreSql(connectionString))))));
```

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

### Static Kafka Partition Assignment

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/consuming/kafka/subscription.md

Manually assign specific Kafka partitions to a consumer endpoint. This example assigns partitions 0, 3, and 5 of 'my-topic'.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddKafka())
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddConsumer("consumer1", consumer => consumer
            .Consume<MyMessage>("endpoint1", endpoint => endpoint
                .ConsumeFrom("my-topic", 0, 3, 5)))));
```

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

### Consumer Startup Configuration

Source: https://github.com/beagle1984/silverback/blob/master/docs/samples/kafka/avro.md

Configures the startup process for the Kafka Avro consumer application.

```csharp
using Avro.Common;
using Microsoft.Extensions.DependencyInjection;
using Silverback.Messaging.Configuration;

namespace Avro.Consumer
{
    public class Startup
    {
        public void ConfigureServices(IServiceCollection services)
        {
            services.AddSilverback(
                    configure => configure
                        .WithConnectionToMessageBroker(broker => broker.UseKafka())
                )
                .WithAvro(avro => avro.UseSchemaRegistry("http://localhost:8081"))
                .WithConsumer<AvroMessage>();
        }
    }
}
```

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

### Configure BinaryMessage Consumer

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/consuming/deserialization.md

Example of configuring a Kafka consumer to consume a custom binary message type. The binary deserializer is used by default for types implementing IBinaryMessage.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddKafka())
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddConsumer("consumer1", consumer => consumer
            .Consume<MyBinaryMessage>("endpoint1", endpoint => endpoint
                .ConsumeFrom("my-topic"))));
```

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

### Run Kafka Basic Producer

Source: https://github.com/beagle1984/silverback/blob/master/samples/README.md

Run the basic Kafka producer application.

```bash
dotnet run --project ./samples/Kafka/Basic.Producer/.
```

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

### Producer Startup Configuration

Source: https://github.com/beagle1984/silverback/blob/master/docs/samples/kafka/binaryfile-streaming.md

Configures the startup services for the Kafka producer application.

```csharp
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.DependencyInjection;
using Silverback.Messaging.Configuration;

namespace BinaryFileStreaming.Producer
{
    public class Startup
    {
        public void ConfigureServices(IServiceCollection services)
        {
            services.AddControllers();
            services.AddSilverback(
                $
```

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

### PostgreSQL Advisory Lock with Entity Framework

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/producing/outbox.md

Example of configuring Silverback to use PostgreSQL advisory locks for distributed locking and processing messages from the outbox using Entity Framework.

```csharp
.AddOutboxWorker(worker => worker
    .WithDistributedLock(distributedLock => distributedLock
        .UsePostgreSqlAdvisoryLock(connectionString))
    .ProcessOutbox(outbox => outbox
        .UsePostgreSql(connectionString)))
```

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

### Run MQTT Binary File Streaming Consumer

Source: https://github.com/beagle1984/silverback/blob/master/samples/README.md

Run the MQTT binary file streaming consumer application. Consumed files will be saved in samples/temp.

```bash
dotnet run --project ./samples/MQTT/BinaryFileStreaming.Consumer/.
```

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

### Configure JSON Serializer with Schema Registry

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/producing/serialization.md

Integrate with Confluent Schema Registry using the dedicated JSON serializer. This example shows how to connect to the registry and configure schema auto-registration.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddKafka())
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddProducer("producer1", producer => producer
            .Produce<MyMessage>("endpoint1", endpoint => endpoint
                .ProduceTo("my-topic")
                .SerializeAsJsonUsingSchemaRegistry(serializer => serializer
                    .ConnectToSchemaRegistry("http://localhost:4242")
                    .Configure(
                        config =>
                        {
                            config.AutoRegisterSchemas = false;
                        })))));
```

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

### Producer Startup Configuration

Source: https://github.com/beagle1984/silverback/blob/master/docs/samples/kafka/avro.md

Configures the startup process for the Kafka Avro producer application.

```csharp
using Microsoft.Extensions.DependencyInjection;
using Silverback.Messaging.Configuration;
using System;

namespace Avro.Producer
{
    public class Startup
    {
        public void ConfigureServices(IServiceCollection services)
        {
            services.AddSilverback(
                    configure => configure
                        .WithConnectionToMessageBroker(broker => broker.UseKafka())
                )
                .WithAvro(avro => avro.UseSchemaRegistry("http://localhost:8081"))
                .WithProducer<AvroMessage>();
        }
    }
}
```

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

### Customize Default Header Names

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/default-headers.md

Use the WithCustomHeaderName method to override default header names. This example shows how to customize chunk ID and chunk count headers for Kafka.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddKafka())
    .WithCustomHeaderName(DefaultMessageHeaders.ChunkId, "x-ch-id")
    .WithCustomHeaderName(DefaultMessageHeaders.ChunksCount, "x-ch-cnt")
    .AddBrokerClientsConfigurator<MyClientsConfigurator>();
```

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

### Configure Silverback Endpoints from IConfiguration

Source: https://github.com/beagle1984/silverback/wiki/External-Configuration

Adapt your startup code to read Silverback endpoint configurations from IConfiguration. This method is useful when configuration is managed centrally.

```csharp
public Startup(IConfiguration configuration)
{
   Configuration = configuration;
}

public IConfiguration Configuration { get; }

public void ConfigureServices(IServiceCollection services)
{
   ...
}

public void Configure(IApplicationBuilder app, IBrokerEndpointsConfigurationBuilder endpoints)
{
    endpoints
        .ReadConfig(Configuration, app.ApplicationServices)
        .Broker.Connect();
}
```

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

### Configure Entity Framework Outbox and Worker

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/producing/outbox.md

Configure Silverback to use the Entity Framework outbox and outbox worker. This setup is for applications using Entity Framework for data access.

```csharp
services
    .AddSilverback()
    .WithConnectionToMessageBroker(options => options
        .AddKafka()
        .AddEntityFrameworkOutbox()
        .AddOutboxWorker(worker => worker
            .ProcessOutbox(outbox => outbox
                .UseEntityFramework<AppDbContext>()))))
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddProducer("producer1", producer => producer
            .Produce<MyMessage>("endpoint1", endpoint => endpoint
                .ProduceTo("my-topic")
                .StoreToOutbox(outbox => outbox
                    .UseEntityFramework<AppDbContext>()))));
```

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

### Configure Multiple Brokers

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/connecting.md

Configure both Kafka and MQTT brokers within the same Silverback application.

```csharp
services
    .AddSilverback()
    .WithConnectionToMessageBroker(options => options
        .AddKafka()
        .AddMqtt());
```

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

### Enable Kafka Batch Processing

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/consuming/batch-processing.md

Enable batch processing for Kafka consumers by calling EnableBatchProcessing on the endpoint configuration. This example sets a batch size of 1000 messages and a timeout of 30 seconds.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddKafka())
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddConsumer("consumer1", producer => producer
            .Consume<MyMessage>("endpoint1", endpoint => endpoint
                .ConsumeFrom("my-topic")
                .EnableBatchProcessing(1000, TimeSpan.FromSeconds(30)))));
```

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

### Run MQTT Basic Producer V3

Source: https://github.com/beagle1984/silverback/blob/master/samples/README.md

Run the basic MQTT producer application using protocol version 3.1.0.

```bash
dotnet run --project ./samples/MQTT/Basic.ProducerV3/.
```

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

### Create a Mocked Kafka Producer on-the-fly

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/testing.md

Simulate an external Kafka producer by creating one directly using `IKafkaTestingHelper.GetProducer`. This is useful for testing scenarios where messages originate outside the application's publisher.

```csharp
var producer = Helper.GetProducer(
    config => config
        .WithBootstrapServers("PLAINTEXT://tests")
        .ProduceTo("test-topic"));

await producer.ProduceAsync(new SomeMessage());
```

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

### Entity Framework Transaction Example for Exactly-Once Semantics

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/consuming/kafka/offset.md

Integrate Kafka offset persistence into an Entity Framework transaction. Enlist the EF transaction with KafkaOffsetStoreScope to ensure atomicity between database changes and offset storage.

```csharp
public async Task OnMessageReceivedAsync(MyMessage message, KafkaOffsetStoreScope offsetStoreScope, AppDbContext db)
{
    await using var tx = await db.Database.BeginTransactionAsync();

    offsetStoreScope.EnlistTransaction(tx.GetDbTransaction());

    // ...db.Add / db.Update ...
    await db.SaveChangesAsync();

    await offsetStoreScope.StoreOffsetsAsync();

    await tx.CommitAsync();
}
```

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

### Run MQTT Basic Consumer V3

Source: https://github.com/beagle1984/silverback/blob/master/samples/README.md

Run the basic MQTT consumer application using protocol version 3.1.0.

```bash
dotnet run --project ./samples/MQTT/Basic.ConsumerV3/.
```

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

### Provisioning SQLite Outbox Table

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/producing/outbox.md

Use SilverbackStorageInitializer to create the necessary outbox table for SQLite. This is an alternative to EF migrations for non-EF scenarios.

```csharp
storageInitializer.CreateSqliteOutboxAsync(connectionString);
```

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

### Publish a Query and Get a Result

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/bus.md

Publish a query message and await its result. Use .Single() to extract the result when multiple subscribers might respond. The ICommand and IQuery interfaces can specify the TResult type.

```csharp
public async Task<QueryResult> PublishSomething()
{
    MyQuery query = new MyQuery() { ... };
    QueryResult result = await _publisher.PublishAsync(query);
    return result.Single();
}
```

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

### Executing a Command with Cancellation Token

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/bus.md

Demonstrates how to execute a command and pass a CancellationToken to it. This allows for interrupting the command execution if needed.

```csharp
await _publisher.ExecuteCommandAsync(myCommand, cancellationToken);
```

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

### Configuring Kafka Endpoint for Tombstones

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/producing/kafka/tombstone.md

Configure the Silverback endpoint for a specific message type to ensure both regular messages and tombstones are routed to the same Kafka topic. This setup requires defining bootstrap servers and producer configurations.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddKafka())
    .AddKafkaClients(clients => clients
        .WithBootstrapServers("PLAINTEXT://localhost:9092")
        .AddProducer("producer1", producer => producer
            .Produce<MyMessage>("endpoint1", endpoint => endpoint
                .ProduceTo("my-topic"))));
```

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

### Conditional Error Handling with Move and Skip

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/consuming/basics.md

Applies specific error handling policies based on exception type and message content. This example moves the message to a Kafka topic if it's a MyException and a condition is met, then skips it.

```csharp
.OnError(policy => policy
    .MoveToKafkaTopic(
        moveEndpoint => moveEndpoint.ProduceTo("some-other-topic"),
        movePolicy => movePolicy
            .ApplyTo<MyException>()
            .ApplyWhen((msg, ex) => msg.Xy == myValue))
    .ThenSkip());
```

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

### Create Kafka Topic

Source: https://github.com/beagle1984/silverback/blob/master/Kafka.md

Creates a Kafka topic named 'Topic1' with one partition and a replication factor of one.

```bash
docker run --net=confluent --rm confluentinc/cp-kafka:5.0.1 kafka-topics --create --topic Topic1 --partitions 1 --replication-factor 1 --if-not-exists --zookeeper zookeeper:2181
```

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

### Run MQTT Binary File Streaming Producer

Source: https://github.com/beagle1984/silverback/blob/master/samples/README.md

Run the MQTT binary file streaming producer application. Browse the Swagger UI to fire sample requests.

```bash
dotnet run --project ./samples/MQTT/BinaryFileStreaming.Producer/.
```

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

### Initialize, Publish, and Commit Kafka Transaction

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/producing/kafka/transactions.md

Demonstrates the process of initializing a Kafka transaction, publishing messages, and then committing the transaction.

```csharp
using IKafkaTransaction transaction = publisher.InitKafkaTransaction();

await publisher.PublishEventAsync(new MyMessage());
await publisher.PublishEventAsync(new MyMessage());
await publisher.PublishEventAsync(new AnotherMessage());

transaction.Commit();
```

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

### Basic Silverback Configuration

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/setup.md

This snippet shows a basic configuration of Silverback using dependency injection. It enables the message bus, configures connections to both Kafka and MQTT brokers, and registers a message handler.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options
        .AddKafka()
        .AddMqtt())
    .AddSingletonSubscriber<MyMessageHandler>();
```

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

### Configure Services with Kafka Broker

Source: https://github.com/beagle1984/silverback/wiki/Connecting-a-Message-Broker

Add the Kafka broker to the DI container and configure inbound and outbound connectors.

```csharp
public void ConfigureServices(IServiceCollection services)
{
    ...

    services
        .AddBus()
        .AddBroker<KafkaBroker>(options => options
            .AddInboundConnector()
            .AddOutboundConnector());
    ...
}
```

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

### Configure Services with Multiple Connectors

Source: https://github.com/beagle1984/silverback/wiki/Multiple-Configurations

Configure the service collection to include a Kafka broker with both a specific DbInboundConnector and a default InboundConnector.

```csharp
protected override void ConfigureServices(IServiceCollection services)
{
    services
        .AddBus()
        .AddBroker<KafkaBroker>(options => options
            .AddDbInboundConnector<MyDbContext>()
            .AddInboundConnector());
```

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

### Run Kafka Binary File Streaming Consumer

Source: https://github.com/beagle1984/silverback/blob/master/samples/README.md

Run the Kafka binary file streaming consumer application. Consumed files will be saved in samples/temp.

```bash
dotnet run --project ./samples/Kafka/BinaryFileStreaming.Consumer/.
```

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

### Run Kafka Transactional Consumer

Source: https://github.com/beagle1984/silverback/blob/master/samples/README.md

Run the Kafka transactional consumer application.

```bash
dotnet run --project ./samples/Kafka/TransactionalProducer.Consumer/.
```

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

### Run Kafka Batch Consumer

Source: https://github.com/beagle1984/silverback/blob/master/samples/README.md

Run the batch Kafka consumer application.

```bash
dotnet run --project ./samples/Kafka/Batch.Consumer/.
```

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

### Configure MQTT Client with Dynamic Topic

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/producing/basics.md

Configures an MQTT client to send messages to topics 'my/even' or 'my/odd' based on a header value.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddMqtt())
    .AddMqttClients(clients => clients
        .ConnectViaTcp(...)
        .AddClient(client => client
            .WithClientId("my.client")
            .Produce<MyMessage>(endpoint => endpoint
                .ProduceTo(envelope => envelope.Headers.GetValue<int>("x-random") % 2 == 0 ? "my/even" : "my/odd"))));
```

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

### Run Kafka Avro Consumer

Source: https://github.com/beagle1984/silverback/blob/master/samples/README.md

Run the Kafka Avro consumer application.

```bash
dotnet run --project ./samples/Kafka/Avro.Consumer/.
```

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

### Configure MQTT Client with Formatted Topic

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/producing/basics.md

Configures an MQTT client to send messages to a formatted topic, dynamically generating parts of the topic name.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddMqtt())
    .AddMqttClients(clients => clients
        .ConnectViaTcp(...)
        .AddClient(client => client
            .WithClientId("my.client")
            .Produce<MyMessage>(endpoint => endpoint
                .ProduceTo(
                    "my/{0}",
                    message => message.Id % 2 == 0 ? new object[] { "even" } : new object[] { "odd" }))));
```

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

### Configure MQTT Producer

Source: https://github.com/beagle1984/silverback/blob/master/docs/guides/broker/producing/basics.md

Configures an MQTT producer using Silverback. Connects via TCP to localhost and adds a client named 'my-client' with client ID 'client1'. It produces messages of type MyMessage to endpoint 'endpoint1' on topic 'messages/my' with at-least-once QoS.

```csharp
services.AddSilverback()
    .WithConnectionToMessageBroker(options => options.AddMqtt())
    .AddMqttClients(clients => clients
        .ConnectViaTcp("localhost")
        .AddClient("my-client", client => client
            .WithClientId("client1")
            .Produce<MyMessage>("endpoint1", endpoint => endpoint
                .ProduceTo("messages/my")
                .WithAtLeastOnceQoS())));
```

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

### Configure Kafka Endpoints and Connect

Source: https://github.com/beagle1984/silverback/wiki/Connecting-a-Message-Broker

Define inbound and outbound Kafka endpoints with their respective configurations and connect the broker. Use `AddInbound` for incoming messages and `AddOutbound` for outgoing messages.

```csharp
public void Configure(..., IBrokerEndpointsConfigurationBuilder endpoints)
{
    endpoints
        .AddInbound(
            new KafkaEndpoint("basket-events")
            {
                Configuration = new Confluent.Kafka.ConsumerConfig
                {
                    BootstrapServers = "PLAINTEXT://kafka:9092",
                    GroupId = "order-service"
                }
            })
        .AddInbound(
            new KafkaEndpoint("payment-events")
            {
                Configuration = new Confluent.Kafka.ConsumerConfig
                {
                    BootstrapServers = "PLAINTEXT://kafka:9092",
                    GroupId = "order-service"
                }
            })
        .AddOutbound<IIntegrationEvent>(
            new KafkaEndpoint("order-events")
            {
                Configuration = new Confluent.Kafka.ProducerConfig
                {
                    BootstrapServers = "PLAINTEXT://kafka:9092"
                }
            })
        .Broker.Connect();
}
```