diff --git a/README.md b/README.md index 2a91c7c..166b43c 100644 --- a/README.md +++ b/README.md @@ -43,4 +43,20 @@ root@tools:/ exit #turn of the docker containers to save on resources $ docker-compose down -v zookeeper kafka tools -``` \ No newline at end of file +``` + +# Optional .NET Kafka Webserver + +This repo also includes a small ASP.NET Core webserver for C# developers who want to experiment with Kafka from a browser instead of using the prebuilt Node webservers. + +```bash +# Start Kafka, create the course topics, and run the .NET webserver +$ docker-compose up -d zookeeper kafka create-topics webserver-dotnet +``` + +Open in your browser. + +* Use **Send** (`/produce`) to write a keyed string message into the `dotnet-messages` Kafka topic. +* Use **Consume** (`/consume`) to view messages consumed from the same topic. The page refreshes automatically every two seconds. + +The implementation lives in `webserver-dotnet/Program.cs` and is intentionally small so it can be modified during class. diff --git a/docker-compose.yml b/docker-compose.yml index f2cec8b..65f247b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -221,6 +221,19 @@ services: done; \ npm run start'" + webserver-dotnet: + build: ./webserver-dotnet + container_name: webserver-dotnet + hostname: webserver-dotnet + ports: + - 3005:8080 + environment: + KAFKA_BOOTSTRAP_SERVERS: kafka:9092 + KAFKA_TOPIC: dotnet-messages + depends_on: + - kafka + - create-topics + postgres: image: postgres:11.2-alpine restart: always @@ -246,6 +259,7 @@ services: kafka-topics --bootstrap-server kafka:9092 --create --topic driver-augmented-avro --partitions 3 --replication-factor 1 ; kafka-topics --bootstrap-server kafka:9092 --create --topic _confluent-monitoring --partitions 12 --replication-factor 1 --config retention.ms=259200000 --config message.timestamp.type=LogAppendTime ; kafka-topics --bootstrap-server kafka:9092 --create --topic driver-positions-string-avro --partitions 3 --replication-factor 1 ; + kafka-topics --bootstrap-server kafka:9092 --create --topic dotnet-messages --partitions 3 --replication-factor 1 ; true' tools: diff --git a/webserver-dotnet/.dockerignore b/webserver-dotnet/.dockerignore new file mode 100644 index 0000000..cd42ee3 --- /dev/null +++ b/webserver-dotnet/.dockerignore @@ -0,0 +1,2 @@ +bin/ +obj/ diff --git a/webserver-dotnet/Dockerfile b/webserver-dotnet/Dockerfile new file mode 100644 index 0000000..cea4efb --- /dev/null +++ b/webserver-dotnet/Dockerfile @@ -0,0 +1,12 @@ +FROM mcr.microsoft.com/dotnet/sdk:8.0 AS build +WORKDIR /src +COPY webserver-dotnet.csproj ./ +RUN dotnet restore +COPY . ./ +RUN dotnet publish webserver-dotnet.csproj -c Release -o /app + +FROM mcr.microsoft.com/dotnet/aspnet:8.0 +WORKDIR /app +ENV ASPNETCORE_URLS=http://0.0.0.0:8080 +COPY --from=build /app ./ +ENTRYPOINT ["dotnet", "webserver-dotnet.dll"] diff --git a/webserver-dotnet/Program.cs b/webserver-dotnet/Program.cs new file mode 100644 index 0000000..20a9523 --- /dev/null +++ b/webserver-dotnet/Program.cs @@ -0,0 +1,208 @@ +using System.Collections.Concurrent; +using Confluent.Kafka; + +var builder = WebApplication.CreateBuilder(args); + +var bootstrapServers = Environment.GetEnvironmentVariable("KAFKA_BOOTSTRAP_SERVERS") ?? "kafka:9092"; +var topic = Environment.GetEnvironmentVariable("KAFKA_TOPIC") ?? "dotnet-messages"; +var groupId = Environment.GetEnvironmentVariable("KAFKA_GROUP_ID") ?? $"dotnet-webserver-{Environment.MachineName}"; + +builder.Services.AddSingleton(new KafkaSettings(bootstrapServers, topic, groupId)); +builder.Services.AddSingleton(); +builder.Services.AddSingleton>(_ => +{ + var config = new ProducerConfig + { + BootstrapServers = bootstrapServers, + ClientId = "dotnet-webserver", + }; + + return new ProducerBuilder(config).Build(); +}); +builder.Services.AddHostedService(); + +var app = builder.Build(); + +app.MapGet("/", () => Results.Redirect("/produce")); + +app.MapGet("/produce", (KafkaSettings settings) => Results.Content(Html.Page("Send Messages", $""" +

Send a string message to Kafka topic {settings.Topic}.

+
+ + + + + + 

+    

+    
+    """), "text/html"));
+
+app.MapGet("/consume", (KafkaSettings settings) => Results.Content(Html.Page("Consume Messages", $"""
+    
Recent messages consumed from Kafka topic {settings.Topic}.

+    
+    

+    
+    """), "text/html"));
+
+app.MapPost("/api/messages", async Task (ProduceRequest request, IProducer producer, KafkaSettings settings) =>
+{
+    if (string.IsNullOrWhiteSpace(request.Value))
+    {
+        return Results.BadRequest(new { error = "Message value is required." });
+    }
+
+    var result = await producer.ProduceAsync(settings.Topic, new Message
+    {
+        Key = string.IsNullOrWhiteSpace(request.Key) ? null : request.Key,
+        Value = request.Value,
+    });
+
+    return Results.Ok(new
+    {
+        status = "sent",
+        result.Topic,
+        result.Partition,
+        result.Offset,
+        result.Message.Key,
+        result.Message.Value,
+    });
+});
+
+app.MapGet("/api/messages", (MessageStore store) => Results.Ok(store.GetAll()));
+
+app.Lifetime.ApplicationStopping.Register(() =>
+{
+    var producer = app.Services.GetRequiredService>();
+    producer.Flush(TimeSpan.FromSeconds(10));
+    producer.Dispose();
+});
+
+app.Run();
+
+record KafkaSettings(string BootstrapServers, string Topic, string GroupId);
+record ProduceRequest(string? Key, string Value);
+record ConsumedMessage(string? Key, string Value, string Topic, int Partition, long Offset, DateTime Timestamp);
+
+class MessageStore
+{
+    private const int MaxMessages = 50;
+    private readonly ConcurrentQueue messages = new();
+
+    public void Add(ConsumedMessage message)
+    {
+        messages.Enqueue(message);
+        while (messages.Count > MaxMessages && messages.TryDequeue(out _))
+        {
+        }
+    }
+
+    public ConsumedMessage[] GetAll() => messages.Reverse().ToArray();
+}
+
+class KafkaConsumerService(KafkaSettings settings, MessageStore store, ILogger logger) : BackgroundService
+{
+    protected override Task ExecuteAsync(CancellationToken stoppingToken) => Task.Run(() => Consume(stoppingToken), stoppingToken);
+
+    private void Consume(CancellationToken stoppingToken)
+    {
+        var config = new ConsumerConfig
+        {
+            BootstrapServers = settings.BootstrapServers,
+            GroupId = settings.GroupId,
+            AutoOffsetReset = AutoOffsetReset.Earliest,
+            EnableAutoCommit = true,
+        };
+
+        using var consumer = new ConsumerBuilder(config).Build();
+        consumer.Subscribe(settings.Topic);
+        logger.LogInformation("Consuming topic {Topic} from {BootstrapServers}", settings.Topic, settings.BootstrapServers);
+
+        try
+        {
+            while (!stoppingToken.IsCancellationRequested)
+            {
+                var result = consumer.Consume(stoppingToken);
+                store.Add(new ConsumedMessage(
+                    result.Message.Key,
+                    result.Message.Value,
+                    result.Topic,
+                    result.Partition.Value,
+                    result.Offset.Value,
+                    result.Message.Timestamp.UtcDateTime));
+            }
+        }
+        catch (OperationCanceledException)
+        {
+            logger.LogInformation("Kafka consumer is stopping.");
+        }
+        finally
+        {
+            consumer.Close();
+        }
+    }
+}
+
+static class Html
+{
+    public static string Page(string title, string body) => $$"""
+    
+    
+    
+      
+      
+      {{title}}
+      
+    
+    
+      
+      

+        
{{title}}

+        {{body}}
+      

+    
+    
+    """;
+}
diff --git a/webserver-dotnet/webserver-dotnet.csproj b/webserver-dotnet/webserver-dotnet.csproj
new file mode 100644
index 0000000..11c0c45
--- /dev/null
+++ b/webserver-dotnet/webserver-dotnet.csproj
@@ -0,0 +1,10 @@
+
+  
+    net8.0
+    enable
+    enable
+  
+  
+    
+  
+