ConfluentKafkaAspNetAdapter 0.2.2

Introduction

This library is used to more easily work with Confluents .NET client. It assumes that specific avro schemas are utilized.

Configuration

Necessary

The library is dependant on two configuration properties: MessageBroker:Uri and SchemaRegistry:Uri, they can either be set in the appsettings.json file or as Environmental Variables

Optional

The following values can be configured optionally:

Exempels

To use this library, add the nuget package to the project that will use it. Within the ConfigureServices method in Startup.cs you are then able to use the extension methods AddKafkaReceiver and AddKafkaPublisher.

Syncronous Kafka Reciever

A syncronous kafka reciever is added to the DI IoC container as follows:

public void ConfigureServices(IServiceCollection services)
{
    services.AddKafkaReceiver<MessageType, KafkaUserMassageReciever>("Topic");
}

The KafkaMessageReciever-type must implement the IKafkaMessageReciever interface e.g:

public class KafkaUserMassageReciever : IKafkaMessageReciever<User>
{
    public void HandleMessage(User message)
    {
        Console.WriteLine(message);
    }
}

This creates a backround service that will create a Reciever of type KafkaUserMessageReciever everytime a message is added to that topic and the HandleMessagemethod will be called.

Asynchronous Kafka Reciever

An asyncronous kafka reciever enables awaiting calls within the HandleMessagemethod and is added to the DI IoC container as follows:

public void ConfigureServices(IServiceCollection services)
{
    services.AddAsyncKafkaReceiver<MessageType, AsyncKafkaMessageReciever>("Topic");
}

The AsyncKafkaMessageReciever-type must implement the IAsyncKafkaMessageReciever interface e.g:

public class AsyncKafkaMessageReciever : IAsyncKafkaMessageReciever<User>
{
    public Task HandleMessage(User message)
    {
        Console.WriteLine(message);
        return Task.CompletedTask;
    }
}

Kafka publisher

A Kafka publisher is added to the DI IoC container as follows:

public void ConfigureServices(IServiceCollection services)
{
    services.AddKafkaPublisher<MessageType>("Topic");
}

The publisher can then be used inside a class as follows:

public class UserManipulator
{
    private readonly IKafkaMessagePublisher<User> kafkaMessagePublisher;
    private User user = new User();

    public UserManipulator(IKafkaMessagePublisher<User> kafkaMessagePublisher)
    {
        this.kafkaMessagePublisher = kafkaMessagePublisher;
    }

    public void AddName(string name)
    {
        user.Name = name;

        kafkaMessagePublisher.Publish(user);
    }
}

A message will then be published to the kafka buss on the specified topic

GroupId

GroupId can be specified for the recievers with the second parameter to the Add[Async]KafkaReciever method. If not specified the groupId is set to

$"{topic}-{Guid.NewGuid}-group"