### KafkaFlow Setup with Dependency Injection

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/getting-started/installation.md

Example of configuring KafkaFlow in the application's Startup.cs using Dependency Injection.

```csharp
using KafkaFlow;
using KafkaFlow.Producers;
using KafkaFlow.Serializer;

public void ConfigureServices(IServiceCollection services)
{
    services.AddKafka(kafka => kafka
        .UseConsoleLog()
        .AddCluster(cluster => cluster
            .WithBrokers(new[] { "localhost:9092" })
            .AddConsumer(consumer => consumer
                .Topic("sample-topic")
                .WithGroupId("sample-group")
                .WithBufferSize(100)
                .WithWorkersCount(10)
                .AddMiddlewares(middlewares => middlewares
                    .AddDeserializer<JsonCoreDeserializer>()
                    .AddTypedHandlers(handlers => handlers
                        .AddHandler<SampleMessageHandler>())
                )
            )
            .AddProducer("producer-name", producer => producer
                .DefaultTopic("sample-topic")
                .AddMiddlewares(middlewares => middlewares
                    .AddSerializer<JsonCoreSerializer>()
                )
            )
        )
    );
}

public void Configure(
    IApplicationBuilder app,
    IWebHostEnvironment env,
    IHostApplicationLifetime lifetime)
{
    var kafkaBus = app.ApplicationServices.CreateKafkaBus();

    lifetime.ApplicationStarted.Register(() => kafkaBus.StartAsync(lifetime.ApplicationStopped));
}
```

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

### Install KafkaFlow packages for Producer

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/getting-started/create-your-first-application.md

Inside the _Producer_ project directory, run the following commands to install the required packages.

```bash
dotnet add package KafkaFlow
dotnet add package KafkaFlow.Microsoft.DependencyInjection
dotnet add package KafkaFlow.LogHandler.Console
dotnet add package KafkaFlow.Serializer.JsonCore
dotnet add package Microsoft.Extensions.DependencyInjection
```

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

### Startup Class Setup

Source: https://github.com/farfetch/kafkaflow/wiki/Setup

Example of configuring KafkaFlow using the Startup class in .NET Core.

```csharp
public void ConfigureServices(IServiceCollection services)
{
    services.AddKafka(kafka => kafka
        .UseConsoleLog()
        .AddCluster(cluster => cluster
            .WithBrokers(new[] { "localhost:9092" })
            .AddConsumer(consumer => consumer
                .Topic("sample-topic")
                .WithGroupId("sample-group")
                .WithBufferSize(100)
                .WithWorkersCount(10)
                .AddMiddlewares(middlewares => middlewares
                    .AddSerializer<NewtonsoftJsonMessageSerializer>()
                    .AddTypedHandlers(handlers => handlers
                        .AddHandler<SampleMessageHandler>())
                )
            )
            .AddProducer("producer-name", producer => producer
                .DefaultTopic("sample-topic")
                .AddMiddlewares(middlewares => middlewares
                    .AddSerializer<NewtonsoftJsonMessageSerializer>()
                )
            )
        )
    );
}

public void Configure(
    IApplicationBuilder app,
    IWebHostEnvironment env,
    IHostApplicationLifetime lifetime)
{
    var kafkaBus = app.ApplicationServices.CreateKafkaBus();

    lifetime.ApplicationStarted.Register(() => kafkaBus.StartAsync(lifetime.ApplicationStopped));
}
```

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

### Create message sender

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/getting-started/create-your-first-application.md

Replace the content of the _Program.cs_ with the following example:

```csharp
using Microsoft.Extensions.DependencyInjection;
using KafkaFlow.Producers;
using KafkaFlow.Serializer;
using KafkaFlow;
using Producer;

var services = new ServiceCollection();

const string topicName = "sample-topic";
const string producerName = "say-hello";

services.AddKafka(
    kafka => kafka
        .UseConsoleLog()
        .AddCluster(
            cluster => cluster
                .WithBrokers(new[] { "localhost:9092" })
                .CreateTopicIfNotExists(topicName, 1, 1)
                .AddProducer(
                    producerName,
                    producer => producer
                        .DefaultTopic(topicName)
                        .AddMiddlewares(m =>
                            m.AddSerializer<JsonCoreSerializer>()
                            )
                )
        )
);

var serviceProvider = services.BuildServiceProvider();

var producer = serviceProvider
    .GetRequiredService<IProducerAccessor>()
    .GetProducer(producerName);

await producer.ProduceAsync(
                   topicName,
                   Guid.NewGuid().ToString(),
                   new HelloMessage { Text = "Hello!" });


Console.WriteLine("Message sent!");

```

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

### Create the Message contract

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/getting-started/create-your-first-application.md

Add a new class file named _HelloMessage.cs_ and add the following example:

```csharp
namespace Producer;

public class HelloMessage
{
    public string Text { get; set; } = default!;
}
```

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

### Hosted Service Setup

Source: https://github.com/farfetch/kafkaflow/wiki/Setup

Example of configuring KafkaFlow as a Hosted Service in .NET Core 2.1 and later.

```csharp
public static void Main(string[] args)
{
    Host
        .CreateDefaultBuilder(args)
        .ConfigureServices((hostContext, services) =>
        {
            services.AddKafkaFlowHostedService(kafka => kafka
                .UseConsoleLog()
                .AddCluster(cluster => cluster
                    .WithBrokers(new[] { "localhost:9092" })
                    .AddConsumer(consumer => consumer
                        .Topic("sample-topic")
                        .WithGroupId("sample-group")
                        .WithBufferSize(100)
                        .WithWorkersCount(10)
                        .AddMiddlewares(middlewares => middlewares
                            .AddSerializer<NewtonsoftJsonMessageSerializer>()
                            .AddTypedHandlers(handlers => handlers
                                .AddHandler<SampleMessageHandler>())
                        )
                    )
                    .AddProducer("producer-name", producer => producer
                        .DefaultTopic("sample-topic")
                        .AddMiddlewares(middlewares => middlewares
                            .AddSerializer<NewtonsoftJsonMessageSerializer>()
                        )
                    )
                )
            );
        })
        .Build()
        .Run();
}
```

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

### Create a Message Handler

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/getting-started/create-your-first-application.md

Create a new class file named _HelloMessageHandler.cs_ and add the following example.

```csharp
using KafkaFlow;
using Producer;

namespace Consumer;

public class HelloMessageHandler : IMessageHandler<HelloMessage>
{
    public Task Handle(IMessageContext context, HelloMessage message)
    {
        Console.WriteLine(
            "Partition: {0} | Offset: {1} | Message: {2}",
            context.ConsumerContext.Partition,
            context.ConsumerContext.Offset,
            message.Text);

        return Task.CompletedTask;
    }
}
```

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

### Unity DI Container Setup

Source: https://github.com/farfetch/kafkaflow/wiki/Setup

Example of configuring KafkaFlow with Unity DI, including adding consumers, producers, serializers, and handlers.

```csharp
var configurator = new KafkaFlowConfigurator(
    new UnityDependencyConfigurator(unityContainer),
    kafka => kafka
        .UseConsoleLog()
        .AddCluster(cluster => cluster
            .WithBrokers(new[] { "localhost:9092" })
            .AddConsumer(consumer => consumer
                .Topic("sample-topic")
                .WithGroupId("sample-group")
                .WithBufferSize(100)
                .WithWorkersCount(10)
                .AddMiddlewares(middlewares => middlewares
                    .AddSerializer<NewtonsoftJsonMessageSerializer>()
                    .AddTypedHandlers(handlers => handlers
                        .AddHandler<SampleMessageHandler>())
                )
            )
            .AddProducer("producer-name", producer => producer
                .DefaultTopic("sample-topic")
                .AddMiddlewares(middlewares => middlewares
                    .AddSerializer<NewtonsoftJsonMessageSerializer>()
                )
            )
        )
);

var bus = configurator.CreateBus(new UnityDependencyResolver(unityContainer));

// Call when your app starts
await bus.StartAsync();

// Call when your app stops
await bus.StopAsync();
```

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

### Installation

Source: https://github.com/farfetch/kafkaflow/blob/master/website/README.md

Installs project dependencies using Yarn.

```bash
$ yarn
```

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

### Start the Kafka cluster

Source: https://github.com/farfetch/kafkaflow/blob/master/samples/KafkaFlow.Sample/README.md

Command to start the Kafka cluster using Docker Compose.

```bash
docker-compose up -d
```

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

### Message Consumer Program

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/getting-started/create-your-first-application.md

This C# code snippet shows how to configure and start a KafkaFlow message consumer, including setting up the Kafka cluster, topic, consumer group, and deserialization.

```csharp
using KafkaFlow;
using KafkaFlow.Serializer;
using Microsoft.Extensions.DependencyInjection;
using Consumer;

const string topicName = "sample-topic";
var services = new ServiceCollection();

services.AddKafka(kafka => kafka
    .UseConsoleLog()
    .AddCluster(cluster => cluster
        .WithBrokers(new[] { "localhost:9092" })
        .CreateTopicIfNotExists(topicName, 1, 1)
        .AddConsumer(consumer => consumer
            .Topic(topicName)
            .WithGroupId("sample-group")
            .WithBufferSize(100)
            .WithWorkersCount(10)
            .AddMiddlewares(middlewares => middlewares
                .AddDeserializer<JsonCoreDeserializer>()
                .AddTypedHandlers(h => h.AddHandler<HelloMessageHandler>())
            )
        )
    )
);

var serviceProvider = services.BuildServiceProvider();

var bus = serviceProvider.CreateKafkaBus();

await bus.StartAsync();

Console.ReadKey();

await bus.StopAsync();
```

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

### Local Development

Source: https://github.com/farfetch/kafkaflow/blob/master/website/README.md

Starts a local development server for live previewing changes.

```bash
$ yarn start
```

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

### Install KafkaFlow using .NET CLI

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/getting-started/installation.md

Command to add the required KafkaFlow NuGet packages to your project.

```shell
dotnet add package KafkaFlow
dotnet add package KafkaFlow.Microsoft.DependencyInjection
dotnet add package KafkaFlow.LogHandler.Console
```

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

### Create Consumer Project

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/getting-started/create-your-first-application.md

Run the following command to create a Console Project named _Consumer_.

```bash
dotnet new console --name Consumer
```

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

### Producer Configuration with Compression

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/compression.md

Example of configuring a Kafka producer with Gzip compression within the KafkaFlow service setup.

```csharp
services.AddKafka(kafka => kafka
    .AddCluster(cluster => cluster
        .WithBrokers(new[] { "localhost:9092" })
        .AddProducer<ProductEventsProducer>(producer => producer
            .WithCompression(CompressionType.Gzip)
                ...
            )
        )
    )
);
```

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

### Create Producer Project

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/getting-started/create-your-first-application.md

Run the following command to create a Console Project named _Producer_.

```bash
dotnet new console --name Producer
```

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

### Configure KafkaFlow using Startup.cs

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/configuration.md

Configures KafkaFlow dependencies in `ConfigureServices` and registers an event to start the Kafka Bus on application start in `Configure`.

```csharp
public void ConfigureServices(IServiceCollection services)
{
    services.AddKafka(kafka => kafka
        .UseConsoleLog()
        .AddCluster(cluster => cluster
            .WithBrokers(new[] { "localhost:9092" })
            ...
        )
    );
}

public void Configure(
    IApplicationBuilder app,
    IWebHostEnvironment env,
    IHostApplicationLifetime lifetime)
{
    var kafkaBus = app.ApplicationServices.CreateKafkaBus();

    lifetime.ApplicationStarted.Register(() => kafkaBus.StartAsync(lifetime.ApplicationStopped));
}
```

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

### Start Consumer Command

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/admin/web-api.md

C# command to start a consumer by its name.

```csharp
var consumerAdmin = provider.GetService<IConsumerAdmin>();
await consumerAdmin.StartConsumerAsync(consumerName);
```

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

### Dynamic Worker Configuration Example

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/consumers/dynamic-workers-configuration.md

Example of configuring dynamic worker count based on peak hours.

```csharp
.AddConsumer(
    consumer => consumer
        ...
        .WithWorkersCount(
            (context, resolver) =>
            {
                // Implement a custom logic to calculate the number of workers
                if (IsPeakHour(DateTime.UtcNow))
                {
                    return Task.FromResult(10); // High worker count during peak hours
                }
                else
                {
                    return Task.FromResult(2); // Lower worker count during off-peak hours
                }
            }, 
            TimeSpan.FromMinutes(15)); // Evaluate the worker count every 15 minutes
        ...
        )
)
```

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

### HandlingMiddleware Example

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/middlewares/batch-consume-middleware.md

Example of a HandlingMiddleware that accesses the message batch using GetMessagesBatch.

```csharp
using KafkaFlow;

internal class HandlingMiddleware : IMessageMiddleware
{
    public Task Invoke(IMessageContext context, MiddlewareDelegate next)
    {
        var batch = context.GetMessagesBatch();
        
        (...)

        return Task.CompletedTask;
    }
}
```

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

### Basic Gzip Compression

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/compression.md

Example of configuring Gzip compression for a producer.

```csharp
.WithCompression(CompressionType.Gzip)
```

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

### Run the Producer

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/getting-started/create-your-first-application.md

This bash command shows how to run the message producer application from the command line.

```bash
dotnet run --project Producer/Producer.csproj
```

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

### Start Jaeger

Source: https://github.com/farfetch/kafkaflow/blob/master/samples/KafkaFlow.Sample.OpenTelemetry/README.md

Command to run Jaeger all-in-one for trace collection.

```bash
docker run --rm --name jaeger \
  -e COLLECTOR_ZIPKIN_HOST_PORT=:9411 \
  -p 6831:6831/udp \
  -p 6832:6832/udp \
  -p 5778:5778 \
  -p 16686:16686 \
  -p 4317:4317 \
  -p 4318:4318 \
  -p 14250:14250 \
  -p 14268:14268 \
  -p 14269:14269 \
  -p 9411:9411 \
  jaegertracing/all-in-one:1.51
```

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

### Run the Consumer

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/getting-started/create-your-first-application.md

This bash command shows how to run the message consumer application from the command line.

```bash
dotnet run --project Consumer/Consumer.csproj
```

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

### Configure Dashboard with .NET 6

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/admin/dashboard.md

Example of configuring the KafkaFlow Dashboard within a .NET 6 application, including Kafka setup and enabling admin/telemetry messages.

```csharp
using KafkaFlow;
using KafkaFlow.Admin.Dashboard;

var builder = WebApplication.CreateBuilder(args);

builder.Services
    .AddKafka(kafka => kafka
        .AddCluster(cluster => cluster
                .WithBrokers(new[] { "localhost:9092" })
                .AddConsumer(consumer => consumer
                    .Topic("mytopic")
                    .WithGroupId("g1")
                    .WithWorkersCount(1)
                    .WithBufferSize(10)
                )
                .EnableAdminMessages("kafka-flow.admin")
                .EnableTelemetry("kafka-flow.admin") // you can use the same topic used in EnableAdminMessages, if need it
        ))
    .AddControllers();

var app = builder.Build();

app.MapControllers();
app.UseKafkaFlowDashboard();

var kafkaBus = app.Services.CreateKafkaBus();
await kafkaBus.StartAsync();

await app.RunAsync();
```

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

### Dynamic Worker Pool Scaling Configuration Example

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/migration/from-v2-to-v3.md

Example of configuring dynamic worker count calculation using `WithWorkersCount` override.

```csharp
.AddConsumer(
    consumerBuilder => consumerBuilder
        .Topic("test-topic")
        .WithGroupId("group1")
        .WithWorkersCount((context, resolver) => {
          // Use whatever logic to determine the worker count
          return GetWorkerCount(context, resolver);
        }, TimeSpan.FromMinutes(15))
)
```

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

### Initialize Kafka Cluster

Source: https://github.com/farfetch/kafkaflow/blob/master/CONTRIBUTING.md

Command to start a Kafka cluster using Docker.

```bash
make init_broker
```

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

### Add required package references for ASP.NET Core Startup

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/configuration.md

Installs the necessary KafkaFlow packages for integration with ASP.NET Core applications.

```bash
dotnet add package KafkaFlow
dotnet add package KafkaFlow.Microsoft.DependencyInjection
```

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

### Install Swashbuckle Package

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/admin/web-api.md

Command to add the Swashbuckle.AspNetCore package using dotnet CLI.

```bash
dotnet add package Swashbuckle.AspNetCore
```

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

### Console Logger Configuration

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/logging.md

Example of configuring KafkaFlow to use the Console LogHandler.

```csharp
services.AddKafka(
    kafka => kafka
        .UseConsoleLog()
        ...

```

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

### Gzip Compression with Level

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/compression.md

Example of configuring Gzip compression with a specific compression level.

```csharp
.WithCompression(CompressionType.Gzip, 5)
```

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

### Expose Swagger UI

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/admin/web-api.md

C# code example for exposing the generated Swagger documentation.

```csharp
app.UseSwagger();
app.UseSwaggerUI(c =>
{
    c.SwaggerEndpoint("/swagger/kafka-flow/swagger.json", "KafkaFlow Admin");
});
```

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

### Basic Consumer Configuration

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/consumers/add-consumers.md

Example of adding a consumer, configuring the topic, consumer group, buffer size, and worker count.

```csharp
using KafkaFlow;
using Microsoft.Extensions.DependencyInjection;

services.AddKafka(kafka => kafka
    .AddCluster(cluster => cluster
        .WithBrokers(new[] { "localhost:9092" })
        .AddConsumer(consumer => consumer
            .Topic("topic-name")
            .WithGroupId("sample-group")
            .WithBufferSize(100)
            .WithWorkersCount(10)
            .AddMiddlewares(middlewares => middlewares
                ...
            )
        )
    )
);
```

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

### Unity or other DI framework integration

Source: https://github.com/farfetch/kafkaflow/wiki/Dependency-Injection

Example of how to configure KafkaFlow with the Unity DI container.

```csharp
var container = new UnityContainer();

var configurator = new KafkaFlowConfigurator(
    // Install KafkaFlow.Unity package
    new UnityDependencyConfigurator(container),
    kafka => kafka
        .AddCluster( ... )
);

//Call bus.StartAsync() when your app starts and bus.StopAsync() when your app stops
var bus = configurator.CreateBus(new UnityDependencyResolver(container));
```

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

### Native Compressor Configuration

Source: https://github.com/farfetch/kafkaflow/wiki/Compressor

Example of configuring native Gzip compression for a producer in KafkaFlow.

```csharp
public class Startup
{
    public void ConfigureServices(IServiceCollection services)
    {
        services.AddKafka(kafka => kafka
            .AddCluster(cluster => cluster
                .WithBrokers(new[] { "localhost:9092" })
                .AddProducer<ProductEventsProducer>(producer => producer
                    .WithCompression(CompressionType.Gzip)
                        ...
                    )
                )
            )
        );
    }
}
```

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

### Add required package references for Hosted Service

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/configuration.md

Installs the necessary KafkaFlow packages for using it as a Hosted Service in .NET applications.

```bash
dotnet add package KafkaFlow
dotnet add package KafkaFlow.Extensions.Hosting
dotnet add package KafkaFlow.Microsoft.DependencyInjection
dotnet add package Microsoft.Extensions.Hosting
```

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

### Basic OpenTelemetry Instrumentation Setup

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/open-telemetry.md

Adds OpenTelemetry instrumentation to KafkaFlow configuration.

```csharp
services.AddKafka(
    kafka => kafka
        .AddCluster(...)
        .AddOpenTelemetryInstrumentation()
);
```

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

### Configure Application

Source: https://github.com/farfetch/kafkaflow/wiki/Dashboard

Example of configuring the application pipeline to use KafkaFlow Dashboard.

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

```

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

### Add required package references for other DI Containers

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/configuration.md

Installs the necessary KafkaFlow packages for integration with the Unity DI container.

```bash
dotnet add package KafkaFlow
dotnet add package KafkaFlow.Unity
```

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

### Add a reference to the Producer

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/getting-started/create-your-first-application.md

In order to access the message contract, add a reference to the Producer Project. Inside the _Consumer_ project directory, run the following commands to add the reference.

```bash
dotnet add reference ../Producer
```

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

### Custom Logger Configuration

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/logging.md

Example of configuring KafkaFlow to use a custom ILogHandler implementation.

```csharp
services.AddKafka(
    kafka => kafka
        .UseLogHandler<YourLogHandler>()
        ...

```

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

### AddBatching Configuration

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/middlewares/batch-consume-middleware.md

Example of configuring the Batch Consume Middleware using the AddBatching extension method.

```csharp
services.AddKafka(kafka => kafka
    .AddCluster(cluster => cluster
        .WithBrokers(new[] { "localhost:9092" })
        .AddConsumer(
            consumerBuilder => consumerBuilder
            ...
            .AddMiddlewares(
                middlewares => middlewares
                    ...
                    .AddBatching(100, TimeSpan.FromSeconds(10)) // Configuration of the BatchConsumeMiddleware
                    .Add<HandlingMiddleware>() // Middleware to process the batch
            )
        )
    )
);
```

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

### Compressor Middleware Configuration

Source: https://github.com/farfetch/kafkaflow/wiki/Compressor

Example of adding the GzipMessageCompressor middleware to a producer in KafkaFlow.

```csharp
public class Startup
{
    public void ConfigureServices(IServiceCollection services)
    {
        services.AddKafka(kafka => kafka
            .AddCluster(cluster => cluster
                .WithBrokers(new[] { "localhost:9092" })
                .AddProducer<ProductEventsProducer>(producer => producer
                    ...
                    .AddMiddlewares(middlewares => middlewares
                        ...
                        .AddCompressor<GzipMessageCompressor>()
                        // or
                        .AddCompressor(resolver => new GzipMessageCompressor(...))
                        ...
                    )
                )
            )
        );
    }
}
```

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

### Install KafkaFlow Admin Packages

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/admin/web-api.md

Command to add necessary KafkaFlow Admin packages using dotnet CLI.

```bash
dotnet add package KafkaFlow.Microsoft.DependencyInjection
dotnet add package KafkaFlow.Admin
dotnet add package KafkaFlow.Admin.WebApi
```

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

### Microsoft Logging Framework Configuration

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/logging.md

Example of configuring KafkaFlow to use the Microsoft Logging Framework.

```csharp
services.AddKafka(
    kafka => kafka
        .UseMicrosoftLog()
        ...

```

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

### Update Package References

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/migration/from-v3-to-v4.md

Example of updating KafkaFlow package references from v3.1.0 to v4.0.0 in a .NET project file.

```xml
<ItemGroup>
-   <PackageReference Include="KafkaFlow" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.Abstractions" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.Admin" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.Admin.Dashboard" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.Admin.WebApi" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.Compressor.Gzip" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.Extensions.Hosting" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.LogHandler.Console" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.LogHandler.Microsoft" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.Microsoft.DependencyInjection" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.OpenTelemetry" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.SchemaRegistry" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.Serializer.JsonCore" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.Serializer.NewtonsoftJson" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.Serializer.ProtobufNet" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.Serializer.SchemaRegistry.ConfluentAvro" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.Serializer.SchemaRegistry.ConfluentJson" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf" Version="3.1.0" />
-   <PackageReference Include="KafkaFlow.Unity" Version="3.1.0" />

+   <PackageReference Include="KafkaFlow" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.Abstractions" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.Admin" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.Admin.Dashboard" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.Admin.WebApi" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.Compressor.Gzip" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.Extensions.Hosting" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.LogHandler.Console" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.LogHandler.Microsoft" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.Microsoft.DependencyInjection" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.OpenTelemetry" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.SchemaRegistry" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.Serializer.JsonCore" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.Serializer.NewtonsoftJson" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.Serializer.ProtobufNet" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.Serializer.SchemaRegistry.ConfluentAvro" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.Serializer.SchemaRegistry.ConfluentJson" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf" Version="4.0.0" />
+   <PackageReference Include="KafkaFlow.Unity" Version="4.0.0" />
</ItemGroup>
```

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

### Registering a custom log handler

Source: https://github.com/farfetch/kafkaflow/wiki/Log-Handler

Example of how to register a custom log handler implementation using `UseLogHandler`.

```csharp
services.AddKafka(
    kafka => kafka
        .UseLogHandler<YourLogHandler>()
        ...
)
```

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

### Consumer Configuration: Before KafkaFlow v3

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/migration/from-v2-to-v3.md

Example of consumer configuration using AddSerializer and AddCompressor before KafkaFlow v3.

```csharp
.AddConsumer(
    consumerBuilder => consumerBuilder
        .Topic("test-topic")
        .WithGroupId("group1")
        .AddMiddlewares(
            middlewares => middlewares
                .AddCompressor<GzipMessageCompressor>()
                .AddSerializer<JsonCoreSerializer>()
        )
)
```

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

### Consumer Configuration: KafkaFlow v3

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/migration/from-v2-to-v3.md

Example of consumer configuration using AddDeserializer and AddDecompressor in KafkaFlow v3.

```csharp
.AddConsumer(
    consumerBuilder => consumerBuilder
        .Topic("test-topic")
        .WithGroupId("group1")
        .AddMiddlewares(
            middlewares => middlewares
                .AddDecompressor<GzipMessageDecompressor>()
                .AddDeserializer<JsonCoreDeserializer>()
        )
)
```

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

### Update KafkaFlow Package References

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/migration/from-v2-to-v3.md

Example of updating KafkaFlow package references from v2.5.0 to v3.0.0 in a .NET project file.

```xml
<ItemGroup>
-   <PackageReference Include="KafkaFlow" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.Abstractions" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.Admin" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.Admin.Dashboard" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.Admin.WebApi" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.BatchConsume" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.Compressor" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.Compressor.Gzip" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.Extensions.Hosting" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.LogHandler.Console" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.LogHandler.Microsoft" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.Microsoft.DependencyInjection" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.OpenTelemetry" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.SchemaRegistry" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.Serializer" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.Serializer.JsonCore" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.Serializer.NewtonsoftJson" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.Serializer.ProtobufNet" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.Serializer.SchemaRegistry.ConfluentAvro" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.Serializer.SchemaRegistry.ConfluentJson" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.TypedHandler" Version="2.5.0" />
-   <PackageReference Include="KafkaFlow.Unity" Version="2.5.0" />

+   <PackageReference Include="KafkaFlow" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.Abstractions" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.Admin" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.Admin.Dashboard" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.Admin.WebApi" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.Compressor.Gzip" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.Extensions.Hosting" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.LogHandler.Console" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.LogHandler.Microsoft" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.Microsoft.DependencyInjection" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.OpenTelemetry" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.SchemaRegistry" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.Serializer.JsonCore" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.Serializer.NewtonsoftJson" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.Serializer.ProtobufNet" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.Serializer.SchemaRegistry.ConfluentAvro" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.Serializer.SchemaRegistry.ConfluentJson" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf" Version="3.0.0" />
+   <PackageReference Include="KafkaFlow.Unity" Version="3.0.0" />
</ItemGroup>
```

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

### Type-based Producer Configuration

Source: https://github.com/farfetch/kafkaflow/wiki/Producers

Example of configuring a producer using its class type (ProductEventsProducer) and how to inject and use the typed producer in a service class.

```csharp
public class Startup
{
    public void ConfigureServices(IServiceCollection services)
    {
        services.AddKafka(kafka => kafka
            .AddCluster(cluster => cluster
                .WithBrokers(new[] { "localhost:9092" })
                .AddProducer<ProductEventsProducer>(
                    producer => producer
                    ...
                )
            )
        );
    }
}

public class ProductEventsProducer : IProductEventsProducer
{
    private readonly IMessageProducer<ProductEventsProducer> producer;

    public ProductEventsProducer(IMessageProducer<ProductEventsProducer> producer)
    {
        this.producer = producer;
    }

    public Task ProduceAync(IProductEvent event) =>
        this.producer.ProduceAsync(event.ProductId.ToString(), event);
}
```

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

### Name-based Producer Configuration

Source: https://github.com/farfetch/kafkaflow/wiki/Producers

Example of configuring a named producer ('product-events') in the application setup and how to access it in a service class.

```csharp
public class Startup
{
    public void ConfigureServices(IServiceCollection services)
    {
        services.AddKafka(kafka => kafka
            .AddCluster(cluster => cluster
                .WithBrokers(new[] { "localhost:9092" })
                .AddProducer(
                    "product-events" //the producer name
                    producer => producer
                    ...
                )
            )
        );
    }
}

public class ProductService : IProductService
{
    private readonly IProducerAccessor producers;

    public ProductEventsProducer(IProducerAccessor producers)
    {
        this.producers = producers;
    }

    public Task CreateProduct(Create)
    {
        ...
        await this.producers["product-events"].ProduceAsync(event.ProductId.ToString(), event);
        ...
    }        
}
```

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

### Run the KafkaFlow Sample

Source: https://github.com/farfetch/kafkaflow/blob/master/samples/KafkaFlow.Sample/README.md

Command to run the KafkaFlow sample application.

```bash
dotnet run
```

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

### Configuring Typed Handlers

Source: https://github.com/farfetch/kafkaflow/wiki/Typed-Handler-Middleware

Example of how to configure the AddTypedHandlers middleware within the KafkaFlow setup, including options for handler lifetime and adding handlers.

```csharp
services.AddKafka(kafka => kafka
    .AddCluster(cluster => cluster
        .WithBrokers(new[] { "localhost:9092" })
        .AddConsumer(consumer => consumer
            ...
            .AddMiddlewares(middlewares => middlewares
                ...
                .AddTypedHandlers(handlers => handlers
                    .WithHandlerLifetime(InstanceLifetime.Singleton)
                    .AddHandler<ProductCreatedHandler>()
                    // or
                    .AddHandlers( ... )
                    // or
                    .AddHandlersFromAssemblyOf<ProductCreatedHandler>()
                )
            )
        )
    )
);
```

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

### Build

Source: https://github.com/farfetch/kafkaflow/blob/master/website/README.md

Generates static content for deployment.

```bash
$ yarn build
```

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

### Deployment (using SSH)

Source: https://github.com/farfetch/kafkaflow/blob/master/website/README.md

Deploys the website using SSH.

```bash
$ USE_SSH=true yarn deploy
```

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

### Build Dashboard UI

Source: https://github.com/farfetch/kafkaflow/blob/master/samples/KafkaFlow.Sample.Dashboard/README.md

Command to build the Dashboard UI using Angular CLI.

```bash
ng build
```

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

### Register Swagger Generator

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/admin/web-api.md

C# code example for registering the Swagger Generator service.

```csharp
using Microsoft.OpenApi.Models;

builder.Services
    .AddSwaggerGen(
        c =>
        {
            c.SwaggerDoc(
                "kafka-flow",
                new OpenApiInfo
                {
                    Title = "KafkaFlow Admin",
                    Version = "kafka-flow",
                });
        });
```

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

### Kafka Authentication Configuration

Source: https://github.com/farfetch/kafkaflow/wiki/Authentication

Example of configuring security information for Kafka brokers in KafkaFlow using C#.

```csharp
services.AddKafka(
    kafka => kafka
        .AddCluster(
            cluster => cluster
                .WithBrokers(new[] {"localhost:9092"})
                .WithSchemaRegistry(config => config.Url = "localhost:8081")
                .WithSecurityInformation(information =>
                {
                  information.SaslMechanism = SaslMechanism.Plain;
                  information.SaslPassword = "pwd";
                  information.SaslUsername = "user";
                  information.SecurityProtocol = SecurityProtocol.SaslPlaintext;
                  information.EnableSslCertificateVerification = true;
                  ...
                })
                ...
            ...

```

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

### Configure Services

Source: https://github.com/farfetch/kafkaflow/wiki/Dashboard

Example of configuring KafkaFlow services, including enabling admin messages and telemetry.

```csharp
 public void ConfigureServices(IServiceCollection services)
 {
    services.AddKafka(kafka => kafka
        .AddCluster(cluster => cluster
            .WithBrokers(new[] { "localhost:9092" })
            .EnableAdminMessages("kafka-flow.admin")
            .EnableTelemetry("kafka-flow.admin") // you can use the same topic used in EnableAdminMessages, if need it
        ));
 }

```

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

### Configure KafkaFlow with Unity DI Container

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/configuration.md

Demonstrates how to use `KafkaFlowConfigurator` with a Unity DI container to set up and manage the Kafka Bus.

```csharp
using KafkaFlow.Configuration;
using KafkaFlow.Unity;
using Unity;


static class Program
{
    public static async Task Main(string[] args)
    {
        var container = new UnityContainer();

        var configurator = new KafkaFlowConfigurator(
            new UnityDependencyConfigurator(container),
            kafka => kafka
                .AddCluster(cluster => cluster
                    .WithBrokers(new[] { "localhost:9092" })
                    ...
                )
        );

        var bus = configurator.CreateBus(new UnityDependencyResolver(container));

        // Call when your app starts
        await bus.StartAsync();

        // Call when your app stops
        await bus.StopAsync();
    }
}
```

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

### Producing a message to a specific topic

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/producers.md

Example demonstrating how to specify the destination topic as the first argument to the ProduceAsync method.

```csharp
await _producers["product-events"]
    .ProduceAsync("products-topic", product.Id.ToString(), product);
```

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

### Manual Partitions Assignment

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/consumers/add-consumers.md

Consumer configuration using ManualAssignPartitions() to specify topic partitions manually.

```csharp
using KafkaFlow;
using KafkaFlow.Serializer;
using Microsoft.Extensions.DependencyInjection;

services.AddKafka(kafka => kafka
    .AddCluster(cluster => cluster
        .WithBrokers(new[] { "localhost:9092" })
        .AddConsumer(consumer => consumer
            .ManualAssignPartitions("topic-name", new[] { 1, 2, 3 })
            ...
        )
    )
);
```

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

### Consumer Lag-Based Worker Balancer Configuration

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/consumers/built-in-workers-algorithms/consumer-lag-based-worker-balancer.md

Example of configuring the WithConsumerLagWorkerBalancer method in KafkaFlow.

```csharp
.AddConsumer(
    consumer => consumer
        ...
        .WithConsumerLagWorkerBalancer(
            50, // The total number of workers to be distributed across all application instances.
            3,  // The minimum number of workers for each application instance.
            20) // The maximum number of workers for each application instance.
        ... 
        )
)
```

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

### Handling No Handler Found event

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/middlewares/typed-handler-middleware.md

Example of configuring a callback to handle events when no specific handler is found.

```csharp
services.AddKafka(kafka => kafka
    .AddCluster(cluster => cluster
        .WithBrokers(new[] { "localhost:9092" })
        .AddConsumer(consumer => consumer
            ...
            .AddMiddlewares(middlewares => middlewares
                ...
                .AddTypedHandlers(handlers => handlers
                    .AddHandler<ProductCreatedHandler>()
                    .WhenNoHandlerFound(context => 
                        Console.WriteLine("Message not handled > Partition: {0} | Offset: {1}",
                            context.ConsumerContext.Partition,
                            context.ConsumerContext.Offset)
                        )
                )
            )
        )
    )
);
```

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

### Automatic Partitions Assignment

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/consumers/add-consumers.md

Consumer configuration using Topic() or Topics() for automatic partition assignment.

```csharp
using KafkaFlow;
using KafkaFlow.Serializer;
using Microsoft.Extensions.DependencyInjection;

services.AddKafka(kafka => kafka
    .AddCluster(cluster => cluster
        .WithBrokers(new[] { "localhost:9092" })
        .AddConsumer(consumer => consumer
            .Topic("topic-name")
            .WithGroupId("sample-group")
            ...
        )
    )
);
```

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

### Configure KafkaFlow in ASP.NET Core

Source: https://github.com/farfetch/kafkaflow/blob/master/website/docs/guides/configuration.md

Registers KafkaFlow dependencies and starts the Kafka Bus before the application runs in an ASP.NET Core application.

```csharp
using KafkaFlow;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddKafka(kafka => kafka
    .AddCluster(cluster => cluster
        .WithBrokers(new[] { "localhost:9092" })
        ...
    )
);

var app = builder.Build();

app.MapGet("/", () => "Hello World!");

var kafkaBus = app.Services.CreateKafkaBus();
await kafkaBus.StartAsync();

app.Run();
```