SimpleKafkaLibrary 1.0.5

Kafka Implementation

in appsettings.json put your configuration

Configuration:

"Kafka": {
    "bootstrapservers": "localhost:9092",
    "UseKafka": true,
    "UseEncryptedData":false,
    "EncryptedKey":"xxxxxx" 
    "FlushProducerInSeconds": 2,
    "ConsumedInSeconds": 2
}

Note that you can use Confluent.Kafka configuration and EncryptedKey is optional if you want to encrypt the data

Producer Usage Sample Producer DI

 public class CreateUserCommandHandler : IRequestHandler<CreateUserCommand, PayloadResponse<UsersDto>>
 {
    private readonly IMessageProducers _messageProducers;

    public CreateUserCommandHandler(IMessageProducers messageProducers)
    {
       _messageProducers = messageProducers
    }

    public async Task<PayloadResponse<UsersDto>> Handle(CreateUserCommand request, CancellationToken cancellationToken)
        {
            ...

            await _messageProducers.ProduceAsync<UsersDto>("User", newUser).ConfigureAwait(false);

            return ResponseStatus<UsersDto>.Create<PayloadResponse<UsersDto>>(ResponseCodes.SUCCESSFUL, _messageProvider.GetMessage(ResponseCodes.SUCCESSFUL), newUser);
        }

 }

Consumer Usage

Sample Consumer class

Path: src\Core\SmartCleanArchitecture.Application\kafkaConsumer\Consumer.cs

    public class Consumer : ConsumerBase
    {

        public Consumer(KafkaConfig configuration, IMessageProducers producers, IMessageAdmin messageAdmin) : base("test1", configuration, messageAdmin)
        {

        }

        public override async Task Invoke()
        {

            await ConsumeAsync<string>("testTopic", (value) =>
            {  
                // put your action here
                Console.WriteLine(value);
            });

            await base.Invoke();
        }
    }

Note: - You can create multiple Consumer classes and configure them in the configuration file. - The groupId is test1. - The topic to be consumed is testTopic.

services.AddKafkaServices<Consumer>(kafkaConfig);

OR

services.AddKafkaServices(cfg =>
{
    cfg.Configure(configuration.GetSection("Kafka"));
    cfg.RegisterConsumer<Consumer>();
    cfg.RegisterConsumer<Consumer2>(); 
});

OR

services.AddKafkaServices(cfg =>
{
    cfg.Configure(configuration.GetSection("Kafka"));

    cfg.RegisterConsumer(Assembly.GetExecutingAssembly());

});

License

This project is licensed with the MIT license.