KafkaService 1.0.2

Kafka Service

Consumer side settings

Program.cs file

builder.Services.AddSingleton<WebSocketConnectionManager>();
builder.Services.AddSingleton<IKafaProducer, KafkaProducer>();
builder.Services.AddSingleton<IKafkaMessageHandler, ConsumerKafkaHandler>();
builder.Services.AddSingleton<IKafkaConsumer, KafkaConsumer>();
builder.Services.AddHostedService<KafkaConsumer>();

appsettings.join

"Kafka": {
"BootstrapServers": "localhost:9092",
"Topics": [ "PostRecord"]
},

KafkaHandler.cs file

public class ConsumerKafkaHandler : IKafkaMessageHandler
{
    private readonly IKafaProducer kafaProducer;

    public ConsumerKafkaHandler(IKafaProducer kafaProducer)
    {
        this.kafaProducer = kafaProducer;
    }

    public async Task<Task> HandleMessageAsync(string topic, string message)
    {
        await Task.Delay(0);
        Console.WriteLine($"Consumer: Received message on topic {topic}: {message}");
        MyApplicationMethod(message);
        switch (topic)
        {
            case "PostRecord":
                var res = kafaProducer.ProduceAsync("PostRecord_ACK", "PostRecord_ACK", message);
                break;
            default:
                Console.WriteLine($"Unknown topic: {topic}");
                break;
        }
        return Task.CompletedTask;
    }


    private void MyApplicationMethod(string msg)
    {
        //Console.WriteLine($"Handling message in app: {msg}");
    }
}

Producer side settings

Program.cs file

builder.Services.AddSingleton<WebSocketConnectionManager>();
builder.Services.AddSingleton<IKafkaMessageHandler, ProducerKafkaHandler>();
builder.Services.AddSingleton<IKafkaConsumer, KafkaConsumer>();
builder.Services.AddSingleton<IKafaProducer,KafkaProducer>();
builder.Services.AddHostedService<KafkaConsumer>();

app.UseWebSockets();

appsettings.join

"Kafka": {
"BootstrapServers": "localhost:9092",
"Topics": [ "PostRecord_ACK" ]
},

KafkaHandler.cs file

public class ProducerKafkaHandler : IKafkaMessageHandler
{
    private readonly WebSocketConnectionManager _connectionManager;

    public MyKafkaHandler(WebSocketConnectionManager connectionManager)
    {
        _connectionManager = connectionManager;
    }

    public async Task<Task> HandleMessageAsync(string topic, string message)
    {
        // Send ack via WebSocket to client
        await _connectionManager.SendMessageAsync("PostRecord", $"Saved {topic} : {message}");
        //_connectionManager.RemoveSocket("PostRecord");
        // Call your application's logic here
        Console.WriteLine($"Producer: Received message on topic {topic}: {message}");
        MyApplicationMethod(message);
        return Task.CompletedTask;
    }


    private void MyApplicationMethod(string msg)
    {
       
    }
}

Producer Controller

[Route("api/[controller]")]
[ApiController]
public class ProducerController : ControllerBase
{
    private readonly IKafaProducer kafaProducer;

    public ProducerController(IKafaProducer kafaProducer)
    {
        this.kafaProducer = kafaProducer;
    }

    [HttpPost]
    public Task<IActionResult> Post([FromBody] PostRequest request)
    {
        var message = JsonSerializer.Serialize(request);

        var res = kafaProducer.ProduceAsync("PostRecord", "PostRecord", message);
        return Task.FromResult<IActionResult>(Ok("Record Updated Successfully..."));
    }
}

WebSocketController

[ApiController]
[Route("ws")]
public class WebSocketController : ControllerBase
{
    private readonly WebSocketConnectionManager _connectionManager;

    public WebSocketController(WebSocketConnectionManager connectionManager)
    {
        _connectionManager = connectionManager;
    }

    [HttpGet()]
    public async Task Get()
    {
        string clientId = "PostRecord";
        if (HttpContext.WebSockets.IsWebSocketRequest)
        {
            var socket = await HttpContext.WebSockets.AcceptWebSocketAsync();
            _connectionManager.AddSocket(clientId, socket);

            var buffer = new byte[1024 * 4];
            while (socket.State == WebSocketState.Open)
            {
                var result = await socket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);

                if (result.MessageType == WebSocketMessageType.Close)
                {
                    _connectionManager.RemoveSocket(clientId);
                    await socket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closed", CancellationToken.None);
                }
            }
        }
        else
        {
            HttpContext.Response.StatusCode = 400;
        }
    }
}