puttiing this on hold because suddenly I hate kafka

I should have just said "cool thing, let me just ctrl+f /.Factory\b/, oh. [delete]"
This commit is contained in:
Adam R. Grey 2021-06-19 04:49:08 -04:00
parent c1c3890211
commit 4aa3162f06
4 changed files with 64 additions and 63 deletions

20
Config.cs Normal file
View File

@ -0,0 +1,20 @@
using System.Collections.Generic;
public class Config
{
public string name { get; set; } = "guy who didn't configure";
public string bootstrap_servers { get; set; } = "localhost:9092";
public string handling_group { get; set; } = null;
public string kafka_location { get; set; } = "/usr/lib/librdkafka.so";
public Dictionary<string, string> topics { get; set; } = new Dictionary<string, string>() {
{"global", "silver_meddlists.global"},
{"directorial", "silver_meddlists.directorial"}
};
public class Command
{
public string name { get; set; }
public string shell { get; set; }
}
public List<Command> commands { get; set; } = new List<Command>() { new Command() { name = "echo", shell = "echo" } };
public List<Command> checks { get; set; }
}

View File

@ -1,74 +1,45 @@
using Confluent.Kafka; using franz;
using Microsoft.Extensions.Configuration; using Newtonsoft.Json;
using System; using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Net.Http;
using System.Threading; using System.Threading;
using System.Threading.Tasks;
namespace directors_assistant namespace directors_assistant
{ {
class Program class Program
{ {
static void Main(string[] args) public static Config conf;
public static Telefranz telefranz;
static async Task Main(string[] args)
{ {
conf = JsonConvert.DeserializeObject<Config>(File.ReadAllText("appsettings.json"));
telefranz = new Telefranz(conf.name, conf.bootstrap_servers, conf.kafka_location,
new List<string>(){"proof_of_concept"});
telefranz.addHandler((silver_messages.directorial.execute_command ec) => {
Console.WriteLine("for FUCK'S sake");
Console.WriteLine(JsonConvert.SerializeObject(ec));
if(ec.command == "proof_of_concept")
{
Console.WriteLine("it's a POC");
var content = new StringContent("{\"username\": \"" + conf.name +"\", \"content\": \"hi\"}");
var client = new HttpClient();
client.DefaultRequestHeaders.Add("Content-Type", "application/json");
client.PostAsync("super secret url",
content);
Console.WriteLine("I'm P-O-C, I'm dyn-o-mite");
}
Console.WriteLine("it's not a proof of concept call");
});
Console.WriteLine("off we go");
telefranz.StartListening();
var config = new ConfigurationBuilder() await Task.Delay(20000);
.AddJsonFile("appsettings.json", true, true) Console.WriteLine("alright fuck it");
.Build();
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();
}
}
}
} }
} }
} }

View File

@ -1,7 +1,14 @@
{ {
"name": "guy who didn't configure", "name": "guy who didn't configure",
"bootstrap servers": "localhost:9092", "bootstrap servers": "localhost:9092",
"handling group": "foo", "handling group": "directors_assistant.foo",
"kafka location": "/usr/lib/librdkafka.so", "kafka location": "/usr/lib/librdkafka.so",
"topic": "SM" "topics": [
"silver_meddlists.global",
"silver_meddlists.directorial"
],
"commands": [
{"name":"echo", "shell": "echo"}
],
"checks": []
} }

View File

@ -4,6 +4,7 @@
<OutputType>Exe</OutputType> <OutputType>Exe</OutputType>
<TargetFramework>net5.0</TargetFramework> <TargetFramework>net5.0</TargetFramework>
<RootNamespace>directors_assistant</RootNamespace> <RootNamespace>directors_assistant</RootNamespace>
<RestoreSources>$(RestoreSources);../pkgs while i procastinate on a feed;https://api.nuget.org/v3/index.json</RestoreSources>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
@ -11,6 +12,8 @@
<PackageReference Include="Microsoft.Extensions.Configuration" Version="5.0.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.FileExtensions" Version="5.0.0" />
<PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="5.0.0" /> <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="5.0.0" />
<PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
<PackageReference Include="silvermeddlists.franz" Version="0.0.1" />
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>