From 6c3c75d42fa97a263bb638cb7bb28e951e009552 Mon Sep 17 00:00:00 2001
From: Thilo Behnke <thilo.behnke@gmx.net>
Date: Fri, 13 May 2022 15:55:42 +0200
Subject: [PATCH] kafka setup

---
 pong/src/event.rs         | 14 +++++++-------
 pong/src/pong.rs          |  4 ++--
 server/.env               |  2 ++
 server/docker-compose.yml | 22 ++++++++++++++++++++++
 server/run.sh             |  7 +++++++
 server/src/kafka.rs       |  4 ++--
 6 files changed, 42 insertions(+), 11 deletions(-)
 create mode 100644 server/.env
 create mode 100644 server/docker-compose.yml
 create mode 100755 server/run.sh

diff --git a/pong/src/event.rs b/pong/src/event.rs
index 04677a9..5209851 100644
--- a/pong/src/event.rs
+++ b/pong/src/event.rs
@@ -10,27 +10,27 @@ pub mod event {
     }
 
     pub trait EventWriterImpl : Send + Sync {
-        fn write(&mut self, event: Event) -> Result<(), ()>;
+        fn write(&mut self, event: Event) -> Result<(), String>;
     }
 
     pub struct FileEventWriterImpl {}
     impl EventWriterImpl for FileEventWriterImpl {
-        fn write(&mut self, event: Event) -> Result<(), ()> {
+        fn write(&mut self, event: Event) -> Result<(), String> {
             let options = OpenOptions::new().read(true).create(true).write(true).open("events.log");
-            if let Err(_) = options {
-                return Err(());
+            if let Err(e) = options {
+                return Err(format!("{}", e));
             }
             let mut file = options.unwrap();
             match file.write(event.msg.as_bytes()) {
                 Ok(_) => Ok(()),
-                Err(e) => Err(())
+                Err(e) => Err(format!("{}", e))
             }
         }
     }
 
     pub struct NoopEventWriterImpl {}
     impl EventWriterImpl for NoopEventWriterImpl {
-        fn write(&mut self, event: Event) -> Result<(), ()> {
+        fn write(&mut self, event: Event) -> Result<(), String> {
             todo!()
         }
     }
@@ -58,7 +58,7 @@ pub mod event {
             }
         }
 
-        pub fn write(&mut self, event: Event) -> Result<(), ()> {
+        pub fn write(&mut self, event: Event) -> Result<(), String> {
            self.writer_impl.write(event)
         }
     }
diff --git a/pong/src/pong.rs b/pong/src/pong.rs
index 4b0b575..5325c3b 100644
--- a/pong/src/pong.rs
+++ b/pong/src/pong.rs
@@ -83,7 +83,7 @@ pub mod pong_events {
     }
 
     pub trait PongEventWriter {
-        fn write(&mut self, event: PongEventType) -> Result<(), ()>;
+        fn write(&mut self, event: PongEventType) -> Result<(), String>;
     }
 
     pub struct DefaultPongEventWriter {
@@ -91,7 +91,7 @@ pub mod pong_events {
     }
 
     impl PongEventWriter for DefaultPongEventWriter {
-        fn write(&mut self, event: PongEventType) -> Result<(), ()> {
+        fn write(&mut self, event: PongEventType) -> Result<(), String> {
             let out_event = match event {
                 PongEventType::GameObjUpdate(ref update) => {
                     Event {
diff --git a/server/.env b/server/.env
new file mode 100644
index 0000000..6d7b559
--- /dev/null
+++ b/server/.env
@@ -0,0 +1,2 @@
+KAFKA_HOST=localhost
+KAFKA_PORT=9092
diff --git a/server/docker-compose.yml b/server/docker-compose.yml
new file mode 100644
index 0000000..2717cac
--- /dev/null
+++ b/server/docker-compose.yml
@@ -0,0 +1,22 @@
+version: "3"
+services:
+  zookeeper:
+    container_name: pong_server_zookeeper
+    image: 'bitnami/zookeeper:latest'
+    ports:
+      - '2181:2181'
+    environment:
+      - ALLOW_ANONYMOUS_LOGIN=yes
+  kafka:
+    container_name: pong_server_kafka
+    image: 'bitnami/kafka:latest'
+    ports:
+      - '9092:9092'
+    environment:
+      - KAFKA_BROKER_ID=1
+      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092
+      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092
+      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
+      - ALLOW_PLAINTEXT_LISTENER=yes
+    depends_on:
+      - zookeeper
diff --git a/server/run.sh b/server/run.sh
new file mode 100755
index 0000000..9a69c01
--- /dev/null
+++ b/server/run.sh
@@ -0,0 +1,7 @@
+#!/usr/bin/env bash
+
+source .env
+
+docker-compose down
+docker-compose up -d --build --force-recreate
+docker exec pong_server_kafka /opt/bitnami/kafka/bin/kafka-topics.sh --create --topic topic --bootstrap-server "$KAFKA_HOST:$KAFKA_PORT"
diff --git a/server/src/kafka.rs b/server/src/kafka.rs
index 016c63d..2f81636 100644
--- a/server/src/kafka.rs
+++ b/server/src/kafka.rs
@@ -22,11 +22,11 @@ impl KafkaEventWriterImpl {
     }
 }
 impl EventWriterImpl for KafkaEventWriterImpl {
-    fn write(&mut self, event: Event) -> Result<(), ()> {
+    fn write(&mut self, event: Event) -> Result<(), String> {
         let record = Record::from_key_value(event.topic.as_str(), event.key.as_str(), event.msg.as_str());
         match self.producer.send(&record) {
             Ok(()) => Ok(()),
-            Err(_) => Err(())
+            Err(e) => Err(format!("{}", e))
         }
     }
 }