initial commit

.gitignore

@@ -0,0 +1,4 @@
+/config.py
+__pycache__
+result
+result-*

config_template.py

@@ -0,0 +1,9 @@
+CONFIG = {
+    'sasl_mechanism': 'PLAIN',
+    'security_protocol': 'SASL_SSL',
+    'sasl_plain_username': '$NAME',
+    'sasl_plain_password': password,
+    'bootstrap_servers': 'kafka.ytrizja.de:9092'
+}
+
+GROUP_ID = 'chat-$NAME'

consume.py

@@ -0,0 +1,9 @@
+from kafka import KafkaConsumer
+from config import CONFIG, GROUP_ID
+
+consumer = KafkaConsumer('chat', group_id=GROUP_ID, auto_offset_reset='earliest', enable_auto_commit=True, **CONFIG)
+try:
+    for message in consumer:
+        print("%d:%d: %s := %s" % (message.partition, message.offset, message.key.decode(), message.value.decode()))
+except KeyboardInterrupt:
+    pass

produce.py

@@ -0,0 +1,15 @@
+import sys
+from kafka import KafkaProducer
+from config import CONFIG
+
+producer = KafkaProducer(compression_type='zstd', **CONFIG)
+topic = 'chat'
+
+key = input('Key: ').encode()
+
+for line in sys.stdin:
+    line = line.rstrip()
+    if '.' == line:
+        break
+
+    producer.send(topic, key=key, value=line.encode())

shell.nix

@@ -0,0 +1,13 @@
+{ pkgs ? import <nixpkgs> {} }:
+let
+  my-python = pkgs.python3;
+  python-with-my-packages = my-python.withPackages (p: with p; [
+    kafka-python
+    zstandard
+  ]);
+in
+pkgs.mkShell {
+  packages = [
+    python-with-my-packages
+  ];
+}