kafka setup

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)
         }
     }

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 {

server/.env

@@ -0,0 +1,2 @@
+KAFKA_HOST=localhost
+KAFKA_PORT=9092

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

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"

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))
         }
     }
 }