Kafka implementation

Makefile

@@ -4,6 +4,8 @@ run-benchmarking:
 	cd backend/benchmarking && cargo run
 run-spectator-client:
 	cd spectator-client && npm run dev
+dev-env:
+	cd backend/dev && docker-compose up -d
 deploy-websockets:
 	cd backend && flyctl deploy --config websockets/fly.toml --app websockets
 deploy-spectator-client:

backend/dev/docker-compose.yml

@@ -0,0 +1,32 @@
+version: '3'
+services:
+  kafka:
+    container_name: kafka
+    image: 'bitnami/kafka:3.3.1'
+    ports:
+      - '9092:9092'
+    tmpfs:
+      - /bitnami/:uid=1001,gid=1
+    command: [
+      sh, -c, 
+      # Kraft specific initialise
+      '/opt/bitnami/scripts/kafka/setup.sh && kafka-storage.sh format --config "$${KAFKA_CONF_FILE}" --cluster-id "lkorDA4qT6W1K_dk0LHvtg" --ignore-formatted && /opt/bitnami/scripts/kafka/run.sh' 
+    ]
+    healthcheck:
+      test: "exit 0"
+    environment:
+      - ALLOW_PLAINTEXT_LISTENER=yes
+
+      # Start Kraft Setup (Kafka as Controller - no Zookeeper)
+      - KAFKA_CFG_NODE_ID=1
+      - KAFKA_CFG_BROKER_ID=1
+      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@127.0.0.1:9093
+      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
+      - KAFKA_CFG_LISTENERS=PLAINTEXT://kafka:29092,PLAINTEXT_HOST://:9092,CONTROLLER://:9093
+      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT
+      - KAFKA_CFG_PROCESS_ROLES=broker,controller
+      - KAFKA_CFG_LOG_DIRS=/tmp/logs
+      - KAFKA_AUTO_CREATE_TOPICS_ENABLE=true
+      # End Kraft Specific Setup
+
+      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092

backend/shared/src/model.rs

@@ -4,6 +4,32 @@ use serde::Deserialize;
 pub struct Settings {
     #[serde(default = "default_log")]
     pub log: Log,
+    #[serde(default = "default_kafka")]
+    pub kafka: Kafka,
+}
+
+fn default_kafka() -> Kafka {
+    Kafka {
+        host: "localhost:9092".into(),
+        auto_commit: true,
+        security_protocol: "PLAINTEXT".into(),
+        topics: Topics {
+            location_update: "location_update".into(),
+        },
+    }
+}
+
+#[derive(Deserialize)]
+pub struct Kafka {
+    pub host: String,
+    pub auto_commit: bool,
+    pub security_protocol: String,
+    pub topics: Topics,
+}
+
+#[derive(Deserialize, Clone)]
+pub struct Topics {
+    pub location_update: String,
 }
 
 #[derive(Deserialize)]

backend/websockets/Cargo.toml

@@ -17,3 +17,5 @@ serde_json = "1.0.70"
 prometheus = "0.13"
 lazy_static = "1.4"
 chrono = "0.4.26"
+rdkafka = { version = "0.33.2", features = ["cmake-build"] }
+async-stream = "0.3.5"

backend/websockets/config/default.toml

@@ -1,3 +1,10 @@
 [log]
-level = "debug"
+level = "info"
 pretty_print = true
+
+[kafka]
+host = "localhost:9092"
+auto_commit = true
+security_protocol = "PLAINTEXT"
+[kafka.topics]
+location_update = "location_updates"

backend/websockets/src/actors/kafka_consumer.rs

@@ -0,0 +1,94 @@
+use async_stream::stream;
+use std::sync::{Arc, Mutex};
+
+use actix::{Actor, Addr, AsyncContext, Context, Handler, StreamHandler, WrapFuture};
+use rdkafka::{
+    consumer::{Consumer, StreamConsumer},
+    error::{KafkaError, KafkaResult},
+    ClientConfig, Message,
+};
+use tracing::{error, info};
+
+use super::{messages::LocationUpdateMessage, race::Race};
+
+#[derive(Debug)]
+enum KafkaConsumerError {
+    KafkaError(KafkaError),
+    SerdeError(serde_json::Error),
+}
+
+pub struct KafkaConsumer {
+    stream_consumer: StreamConsumer,
+}
+
+impl KafkaConsumer {
+    pub fn new(config: ClientConfig, topics: Vec<&str>) -> Result<Self, KafkaError> {
+        let stream_consumer: StreamConsumer = config.create()?;
+        stream_consumer.subscribe(&topics)?;
+
+        Ok(KafkaConsumer { stream_consumer })
+    }
+
+    async fn consume(&mut self) -> Result<LocationUpdateMessage, KafkaConsumerError> {
+        match self.stream_consumer.recv().await {
+            KafkaResult::Err(err) => {
+                error!("could not consume kafka message: {}", err);
+                Err(KafkaConsumerError::KafkaError(err))
+            }
+            KafkaResult::Ok(msg) => match serde_json::from_slice(msg.payload().unwrap()) {
+                Ok(location_update) => Ok(location_update),
+                Err(e) => {
+                    error!("failed to deserialize kafka message: {}", e);
+                    Err(KafkaConsumerError::SerdeError(e))
+                }
+            },
+        }
+    }
+}
+
+pub struct KafkaConsumerActor {
+    kafka_consumer: Arc<Mutex<KafkaConsumer>>,
+    race_addr: Addr<Race>,
+}
+
+impl KafkaConsumerActor {
+    pub fn new(kafka_consumer: KafkaConsumer, race_addr: Addr<Race>) -> Self {
+        Self {
+            kafka_consumer: Arc::new(Mutex::new(kafka_consumer)),
+            race_addr,
+        }
+    }
+}
+
+impl Actor for KafkaConsumerActor {
+    type Context = Context<Self>;
+
+    fn started(&mut self, ctx: &mut Self::Context) {
+        let consumer_clone = self.kafka_consumer.clone();
+        let race_addr = self.race_addr.clone();
+
+        // Create a separate async task that runs the loop
+        ctx.spawn(
+            async move {
+                let mut kafka_consumer = match consumer_clone.lock() {
+                    Ok(consumer) => consumer,
+                    Err(_) => {
+                        error!("Failed to acquire lock on Kafka consumer");
+                        return;
+                    }
+                };
+
+                loop {
+                    match kafka_consumer.consume().await {
+                        Ok(location_update) => {
+                            // Directly sending the location_update to race_addr
+                            race_addr.do_send(location_update);
+                        }
+                        Err(e) => error!("error consuming kafka message: {:?}", e),
+                    }
+                }
+            }
+            .into_actor(self),
+        );
+    }
+}

backend/websockets/src/actors/mod.rs

@@ -2,6 +2,7 @@ use chrono::Utc;
 
 use self::messages::LocationUpdateMessage;
 
+pub mod kafka_consumer;
 pub mod messages;
 pub mod race;
 pub mod ws;

backend/websockets/src/actors/race.rs

@@ -11,7 +11,7 @@ use crate::{
 };
 
 use super::messages::{Connect, Disconnect, LocationUpdateMessage, WsMessage};
-use tracing::{debug, error};
+use tracing::{debug, error, info};
 
 type Socket = Recipient<WsMessage>;
 

backend/websockets/src/actors/ws.rs

@@ -4,42 +4,55 @@ use actix::{
 };
 use actix::{AsyncContext, Message};
 use actix_web_actors::ws;
+use rdkafka::producer::{FutureProducer, FutureRecord};
+use rdkafka::util::Timeout;
+use shared::model::Topics;
 use std::time::{Duration, Instant};
-use tracing::{debug, error};
+use tracing::{debug, error, info};
 
 use super::messages::{Connect, Disconnect, LocationUpdateMessage, WsMessage};
 use super::race::Race;
 
 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
 const CLIENT_TIMEOUT: Duration = Duration::from_secs(10);
 
-#[derive(Message, Debug)]
+#[derive(Message)]
 #[rtype(result = "()")]
 pub struct WsConnection {
     pub user_id: String,
     pub race_id: String,
     pub race_addr: Addr<Race>,
     pub heartbeat: Instant,
+    pub kafka_producer: FutureProducer,
+    pub kafka_topics: Topics,
 }
 
 impl WsConnection {
-    pub fn new(user_id: String, race_id: String, race: Addr<Race>) -> WsConnection {
+    pub fn new(
+        user_id: String,
+        race_id: String,
+        race: Addr<Race>,
+        kafka_producer: FutureProducer,
+        kafka_topics: Topics,
+    ) -> WsConnection {
         WsConnection {
             user_id,
             race_id,
             heartbeat: Instant::now(),
             race_addr: race,
+            kafka_producer,
+            kafka_topics,
         }
     }
 
     fn heartbeat(&self, ctx: &mut ws::WebsocketContext<WsConnection>) {
         ctx.run_interval(HEARTBEAT_INTERVAL, |act, ctx| {
-            debug!(message = "pinging client", action = "heartbeat", ?act);
+            debug!(message = "running heartbeat check", user_id = ?act.user_id);
             if Instant::now().duration_since(act.heartbeat) > CLIENT_TIMEOUT {
                 error!(
                     message = "client heartbeat failed, disconnecting",
                     action = "heartbeat",
-                    ?act
+                    user_id = ?act.user_id
                 );
                 act.race_addr.do_send(Disconnect {
                     user_id: act.user_id.clone(),
@@ -51,6 +64,43 @@ impl WsConnection {
             ctx.ping(b"PING");
         });
     }
+
+    fn process_message(&mut self, location_update_message: LocationUpdateMessage) {
+        let kafka_message = match serde_json::to_string(&location_update_message) {
+            Ok(message) => message,
+            Err(e) => {
+                error!(
+                    message = "failed to serialize message",
+                    action = "process_message",
+                    ?e
+                );
+                return;
+            }
+        };
+
+        let user_id = location_update_message.user_id;
+        let kafka_producer = self.kafka_producer.clone();
+        let kafka_topic = self.kafka_topics.location_update.clone();
+        actix_web::rt::spawn(async move {
+            let result = kafka_producer
+                .send(
+                    FutureRecord::to(&kafka_topic)
+                        .payload(&kafka_message)
+                        .key(&user_id),
+                    Timeout::Never,
+                )
+                .await;
+
+            match result {
+                Ok(_) => {
+                    debug!(message = "message sent", action = "process_message");
+                }
+                Err(e) => {
+                    error!(message = "message not sent", action = "process_message", ?e);
+                }
+            }
+        });
+    }
 }
 
 impl Handler<WsMessage> for WsConnection {
@@ -85,6 +135,14 @@ impl Actor for WsConnection {
             })
             .wait(ctx)
     }
+
+    fn stopping(&mut self, ctx: &mut Self::Context) -> actix::Running {
+        self.race_addr.do_send(Disconnect {
+            user_id: self.user_id.clone(),
+        });
+        ctx.stop();
+        actix::Running::Stop
+    }
 }
 
 impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsConnection {
@@ -100,13 +158,12 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsConnection {
                 self.heartbeat = Instant::now();
             }
             Ok(ws::Message::Binary(bin)) => {
-                //These messages comes from the app
+                //These messages come from the app
                 debug!(message = "binary message", action = "handle", ?bin);
-                let location_update_message: LocationUpdateMessage = match bin.try_into() {
-                    Ok(message) => message,
-                    Err(e) => return ctx.text(e.to_json().to_string()),
+                match bin.try_into() {
+                    Ok(location_update_message) => self.process_message(location_update_message),
+                    Err(e) => ctx.text(e.to_json().to_string()),
                 };
-                self.race_addr.do_send(location_update_message);
             }
             Ok(ws::Message::Close(reason)) => {
                 ctx.close(reason);
@@ -119,12 +176,10 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsConnection {
             Ok(ws::Message::Text(text)) => {
                 //These messages comes from other clients
                 debug!(message = "text message", action = "handle", ?text);
-                let location_update_message: LocationUpdateMessage =
-                    match text.as_bytes().to_owned().try_into() {
-                        Ok(message) => message,
-                        Err(e) => return ctx.text(e.to_json().to_string()),
-                    };
-                self.race_addr.do_send(location_update_message);
+                match text.as_bytes().to_owned().try_into() {
+                    Ok(location_update_message) => self.process_message(location_update_message),
+                    Err(e) => ctx.text(e.to_json().to_string()),
+                };
             }
             Err(e) => panic!("{}", e),
         }