subscribe consumers to specific partition-topic combinations

pong/src/event.rs

@@ -68,7 +68,6 @@ pub mod event {
 
     pub trait EventReaderImpl : Send + Sync {
         fn read(&mut self) -> Result<Vec<Event>, String>;
-        fn read_from_topic(&mut self, topic: &str, key: &str) -> Result<Vec<Event>, String>;
     }
 
     pub struct EventReader {
@@ -85,9 +84,5 @@ pub mod event {
         pub fn read(&mut self) -> Result<Vec<Event>, String> {
             self.reader_impl.read()
         }
-
-        pub fn read_from_topic(&mut self, topic: &str, key: &str) -> Result<Vec<Event>, String> {
-            self.reader_impl.read_from_topic(topic, key)
-        }
     }
 }

server/src/hash.rs

@@ -2,7 +2,7 @@ use std::ptr::hash;
 
 pub struct Hasher {}
 impl Hasher {
-    pub fn hash(n: u32) -> String {
+    pub fn hash(n: u16) -> String {
         let digest = md5::compute(format!("{}", n));
         format!("{:x}", digest)
     }

server/src/kafka.rs

@@ -118,19 +118,34 @@ impl KafkaEventReaderImpl {
             consumer
         }
     }
+
+    pub fn for_partitions(host: &str, partitions: &[i32], topics: &[&str]) -> KafkaEventReaderImpl {
+        println!("Connecting partition specific consumer to kafka host: {}", host);
+        let mut builder = Consumer::from_hosts(vec!(host.to_owned()));
+        for topic in topics.iter() {
+            builder = builder.with_topic_partitions(topic.parse().unwrap(), partitions);
+        }
+        builder = builder
+            .with_fallback_offset(FetchOffset::Earliest)
+            .with_group("group".to_owned())
+            .with_offset_storage(GroupOffsetStorage::Kafka);
+
+        let consumer = builder
+            .create()
+            .unwrap();
+        KafkaEventReaderImpl {
+            consumer
+        }
+    }
 }
 impl EventReaderImpl for KafkaEventReaderImpl {
     fn read(&mut self) -> Result<Vec<Event>, String> {
-        self.consume(None, None)
-    }
-
-    fn read_from_topic(&mut self, topic: &str, key: &str) -> Result<Vec<Event>, String> {
-        self.consume(Some(topic), Some(key))
+        self.consume()
     }
 }
 
 impl KafkaEventReaderImpl {
-    fn consume(&mut self, topic: Option<&str>, key: Option<&str>) -> Result<Vec<Event>, String> {
+    fn consume(&mut self) -> Result<Vec<Event>, String> {
         // TODO: How to best filter messages by key (= game session id?)
         // E.g. https://docs.rs/kafka/latest/kafka/producer/struct.DefaultPartitioner.html - is it possible to read from partition by retrieving the hash of the key?
         // Does it even make sense to hash the key if it already is a hash? Custom partitioner?
@@ -152,6 +167,26 @@ impl KafkaEventReaderImpl {
     }
 }
 
+pub struct KafkaSessionEventReaderImpl {
+    inner: KafkaEventReaderImpl
+}
+
+impl KafkaSessionEventReaderImpl {
+    pub fn new(host: &str, session: &Session, topics: &[&str]) -> KafkaSessionEventReaderImpl {
+        let partitions = [session.id as i32];
+        KafkaSessionEventReaderImpl {
+            inner: KafkaEventReaderImpl::for_partitions(host, &partitions, topics)
+        }
+    }
+}
+
+impl EventReaderImpl for KafkaSessionEventReaderImpl {
+    fn read(&mut self) -> Result<Vec<Event>, String> {
+        self.inner.read()
+    }
+}
+
+
 #[derive(Debug)]
 pub struct KafkaTopicManager {
     partition_management_endpoint: String
@@ -166,7 +201,7 @@ impl KafkaTopicManager {
         KafkaTopicManager {partition_management_endpoint: format!("http://{}/add_partition", topic_manager_host).to_owned()}
     }
 
-    pub async fn add_partition(&self) -> Result<u32, String> {
+    pub async fn add_partition(&self) -> Result<u16, String> {
         let mut client = Client::new();
         let request = Request::builder().method(Method::POST).uri(Uri::from_str(&self.partition_management_endpoint).unwrap()).body(Body::empty()).unwrap();
         let res = client.request(request).await;
@@ -208,7 +243,7 @@ impl KafkaTopicManager {
 
 #[derive(Deserialize)]
 struct PartitionApiDTO {
-    data: u32
+    data: u16
 }
 
 pub struct SessionPartitioner {}

server/src/session.rs

@@ -1,19 +1,22 @@
 use kafka::producer::Producer;
 use crate::hash::Hasher;
-use crate::kafka::{KafkaDefaultEventWriterImpl, KafkaSessionEventWriterImpl, KafkaTopicManager};
+use crate::kafka::{KafkaDefaultEventWriterImpl, KafkaEventReaderImpl, KafkaSessionEventReaderImpl, KafkaSessionEventWriterImpl, KafkaTopicManager};
 use serde::{Serialize, Deserialize};
 use serde_json::json;
 use pong::event::event::{Event, EventReader, EventWriter};
 
 pub struct SessionManager {
+    kafka_host: String,
     sessions: Vec<Session>,
     session_producer: EventWriter,
     topic_manager: KafkaTopicManager
 }
 
+// TODO: On startup read the session events from kafka to restore the session id <-> hash mappings.
 impl SessionManager {
     pub fn new(kafka_host: &str) -> SessionManager {
         SessionManager {
+            kafka_host: kafka_host.to_owned(),
             sessions: vec![],
             topic_manager: KafkaTopicManager::from("localhost:7243"),
             session_producer: EventWriter::new(Box::new(KafkaDefaultEventWriterImpl::new(kafka_host)))
@@ -41,12 +44,16 @@ impl SessionManager {
         self.sessions.push(session.clone());
         Ok(session)
     }
+
+    pub fn get_session_consumer(&self, session: Session) -> EventReader {
+        EventReader::new(Box::new(KafkaSessionEventReaderImpl::new(&self.kafka_host, &session, &["move", "status", "input"])))
+    }
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Session {
-    id: u32,
-    hash: String
+    pub id: u16,
+    pub hash: String
 }
 
 pub struct SessionWriter {
@@ -67,8 +74,8 @@ pub struct SessionReader {
 }
 
 impl SessionReader {
-    pub fn read_from_session(&mut self, topic: String) -> Result<Vec<Event>, String> {
-        self.reader.read_from_topic(topic.as_str(), &self.session.id.to_string())
+    pub fn read_from_session(&mut self) -> Result<Vec<Event>, String> {
+        self.reader.read()
     }
 }