kafka woooooo

Program.cs

@@ -1,5 +1,7 @@
-using System;
+using Confluent.Kafka;
 using Microsoft.Extensions.Configuration;
+using System;
+using System.Threading;
 
 namespace directors_assistant
 {
@@ -7,11 +9,66 @@ namespace directors_assistant
     {
         static void Main(string[] args)
         {
+
             var config = new ConfigurationBuilder()
                 .AddJsonFile("appsettings.json", true, true)
                 .Build();
-            var name = config["name"];
-            Console.WriteLine($"Hello {name}!");
+
+            Library.Load(config["kafka location"]);
+            var producerConfig = new ProducerConfig
+            {
+                BootstrapServers = config["bootstrap servers"]
+            };
+            var consumerConfig = new ConsumerConfig
+            {
+                BootstrapServers = config["bootstrap servers"],
+                GroupId = config["handling group"],
+            };
+
+            using (var producer = new ProducerBuilder<Null, string>(producerConfig).Build())
+            {
+                try
+                {
+                    producer.Produce(config["topic"], new Message<Null, string> { Value = "" + config["handling group"] });
+                }
+                catch (ProduceException<Null, string> e)
+                {
+                    Console.WriteLine($"Delivery failed: {e.Error.Reason}");
+                }
+
+
+
+                using (var consumer = new ConsumerBuilder<Ignore, string>(consumerConfig).Build())
+                {
+                    consumer.Subscribe(config["topic"]);
+
+                    CancellationTokenSource cts = new CancellationTokenSource();
+                    Console.CancelKeyPress += (_, e) =>
+                    {
+                        e.Cancel = true; // prevent the process from terminating.
+                        cts.Cancel();
+                    };
+                    try
+                    {
+                        while (true)
+                        {
+                            try
+                            {
+                                var cr = consumer.Consume(cts.Token);
+                                Console.WriteLine($"Consumed message '{cr.Message.Value}' at: '{cr.TopicPartitionOffset}'.");
+                            }
+                            catch (ConsumeException e)
+                            {
+                                Console.WriteLine($"Error occured: {e.Error.Reason}");
+                            }
+                        }
+                    }
+                    catch (OperationCanceledException)
+                    {
+                        consumer.Close();
+                    }
+                }
+            }
         }
     }
 }

README.md

@@ -1,3 +1,7 @@
 # directors-assistant
 
 listen on kafka for commands, execute without question
+
+
+Depends on librdkafka.redist, but confluent's nuget package doesn't know that.
+and worse, it sounds like the point where it just mysteriously loses an exception is where it would try to load it itself

appsettings.example.json

@@ -1,3 +1,7 @@
 {
-    "name": "guy who didn't configure"
+    "name": "guy who didn't configure",
+    "bootstrap servers": "localhost:9092",
+    "handling group": "foo",
+    "kafka location": "/usr/lib/librdkafka.so",
+    "topic": "SM"
 }

directors-assistant.csproj

@@ -7,6 +7,7 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="confluent.kafka" Version="1.7.0" />
     <PackageReference Include="Microsoft.Extensions.Configuration" Version="5.0.0" />
     <PackageReference Include="Microsoft.Extensions.Configuration.FileExtensions" Version="5.0.0" />
     <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="5.0.0" />