Chd.Library.Kafka 9.0.4

Kafka Helper Library For .Net Core

Chd (cleverly handle difficulty) library helps you cleverly handle difficulty, writing code fastly and do your application stable.

📝 Table of Contents

• About
• Usage
• Injection Kafka Producer Into The Project
• Producing Some Messages To Kafka
• Apsettings Configurations
• Consuming Messages
• Injection Kafka Consumer Into The Project
• Authors
• Acknowledgments

🧐 About

Kafka is primarily used to build real-time streaming data pipelines and applications that adapt to the data streams.

🏁 Getting Started

Apache Kafka is an open-source distributed event streaming platform used by thousands of companies for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications. This library simplfy Kafka using on .net core applications.

Prerequisites

You must use .net core 9.0 or higher

🎈 Usage

After we inject the Chd.Kafka library into the project, we can produce and consume messages from Kafka. We can use the MessageProducer class for producing messages and the ConsumerServiceBase class for consuming messages. We can create multiple consumer services for consuming messages from different topics. We can also inject services into the consumer services. We can use the OnMessageDelivered method for processing the messages and the OnErrorOccured method for processing the errors.

💉 Injection Kafka Producer Into The Project

   var builder = WebApplication.CreateBuilder(args);
   ....
   ....
   builder.Services.AddKafkaProducer();

🚀 Producing Some Messages To Kafka

We will produce two messages for Kafka on different topics. We will create a KafkaTestController which inherits ControllerBase class. We will inject the MessageProducer class into the KafkaTestController. We will produce two messages for Kafka on different topics. We will use the Produce method of the MessageProducer class for producing messages. We will use the AddMessagesToKafkaQueue method for testing on swagger or etc.

     public class KafkaTestController : ControllerBase
    {
        private readonly MessageProducer _messageProducer;
        public KafkaTestController(MessageProducer messageProducer)
        {
            _messageProducer = messageProducer;
        }
        [Route("AddMessagesToKafkaQueue")]
        [HttpGet]
        public string? AddMessagesToKafkaQueue()//For testing on swagger or etc..
        {
            _messageProducer.Produce("topic1", new EmailMessage { Subject = "Test1 Subject", Content = "Test1 Content", To = "test1@gmail.com" });
            Thread.Sleep(3000);
            _messageProducer.Produce("topic2", new EmailMessage { Subject = "Test2 Subject", Content = "Test2 Content", To = "test2@gmail.com" });
            Thread.Sleep(3000);
            return "Ok";
        }
    }

⚙️ Apsettings Configurations

You must add code below in appsettings.json

   "Kafka": {
    "Host": "192.168.1.1", //change ip
    "Port": "6501" //change port
  },

You can see the message in Kafka queue after running this code. Topics will be created automatically. We have come to the next stage, let's consume the messages. Open another solution for consume the produced messages. Let's say we have two consumers who consume different topics.

🛠️ Consuming Messages

We will create two Kafka consumer service which inherits "ConsumerServiceBase" class

    public class Consumer1 : ConsumerServiceBase
    {
        public Consumer1()
        {
        }

        public override void OnMessageDelivered(string message)
        {
        //Add break point here
        }

        public override void OnErrorOccured(Error error)
        {
         //Add break point here
        }
    }
    public class Consumer2 : ConsumerServiceBase
    {
        public Consumer2()
        {
        }

        public override void OnMessageDelivered(string message)
        {
         //Add break point here
        }

        public override void OnErrorOccured(Error error)
        {
         //Add break point here
        }
    }

We should create two consumer service which consume different topics. We can now consume the messages we have previously produced.

💉 Injection Kafka Consumer Into The Project

   var builder = WebApplication.CreateBuilder(args);
   ....
   ....
   builder.Services.AddKafkaConsumer<Consumer1>("topic1");
   builder.Services.AddKafkaConsumer<Consumer2>("topic2");

Now when we run the application we will consume the messages produced in kafka.

✍️ Authors

• Mehmet Yoldaş - Linkedin

See also the list of contributors who participated in this project.

🎉 Acknowledgements

Thank you for using my library.