diff --git a/pong/src/event.rs b/pong/src/event.rs index 93b5f53..3cf4d04 100644 --- a/pong/src/event.rs +++ b/pong/src/event.rs @@ -67,6 +67,7 @@ pub mod event { pub trait EventReaderImpl : Send + Sync { fn read(&mut self) -> Vec; + fn read_from_topic(&mut self, topic: &str, key: &str) -> Vec; } pub struct EventReader { @@ -83,5 +84,9 @@ pub mod event { pub fn read(&mut self) -> Vec { self.reader_impl.read() } + + pub fn read_from_topic(&mut self, topic: &str, key: &str) -> Vec { + self.reader_impl.read_from_topic(topic, key) + } } } diff --git a/server/Cargo.toml b/server/Cargo.toml index f165d96..27f2a4b 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -12,3 +12,6 @@ tokio = { version = "1", features = ["full"] } tokio-stream = {version = "0.1" } serde_json = { version = "1.0.79" } pong = { path = "../pong", version = "0.1.0" } + +[dev-dependencies] +rstest = "0.12.0" diff --git a/server/src/http.rs b/server/src/http.rs index 661d3b1..9b3032d 100644 --- a/server/src/http.rs +++ b/server/src/http.rs @@ -8,6 +8,7 @@ use kafka::producer::Producer; use tokio::sync::Mutex; use pong::event::event::{Event, EventReader, EventWriter}; use crate::kafka::{KafkaEventReaderImpl, KafkaEventWriterImpl}; +use crate::utils::http_utils::get_query_params; pub struct HttpServer { addr: [u8; 4], @@ -68,10 +69,19 @@ async fn handle_event_write(event_writer: &Arc>, req: Request } async fn handle_event_read(event_reader: &Arc>, req: Request) -> Result, Infallible> { - let mut locked = event_reader.lock().await; - let events = locked.read(); - println!("read {} events", events.len()); - Ok(Response::new(format!("{:?}", events).into())) + let query_params = get_query_params(&req); + println!("{:?}", query_params); + let topic = query_params.get("topic"); + let key = query_params.get("key"); + match (topic, key) { + (Some(topic), Some(key)) => { + let mut locked = event_reader.lock().await; + let events = locked.read(); + println!("read {} events", events.len()); + Ok(Response::new(format!("{:?}", events).into())) + }, + _ => Ok(Response::new(format!("provide topic and key as query params").into())), + } } diff --git a/server/src/kafka.rs b/server/src/kafka.rs index d127f64..deeabbc 100644 --- a/server/src/kafka.rs +++ b/server/src/kafka.rs @@ -57,16 +57,30 @@ impl KafkaEventReaderImpl { } impl EventReaderImpl for KafkaEventReaderImpl { fn read(&mut self) -> Vec { + self.consume(None, None) + } + + fn read_from_topic(&mut self, topic: &str, key: &str) -> Vec { + self.consume(Some(topic), Some(key)) + } +} + +impl KafkaEventReaderImpl { + fn consume(&mut self, topic: Option<&str>, key: Option<&str>) -> Vec { let polled = self.consumer.poll().unwrap(); let message_sets: Vec> = polled.iter().collect(); let mut events = vec![]; for ms in message_sets { + let topic = ms.topic(); + let partition = ms.partition(); + println!("topic={},partition={}", topic, partition); for m in ms.messages() { - let event = Event {topic: String::from(ms.topic()), key: std::str::from_utf8(m.key).unwrap().parse().unwrap(), msg: std::str::from_utf8(m.value).unwrap().parse().unwrap() }; + let event = Event {topic: String::from(topic), key: std::str::from_utf8(m.key).unwrap().parse().unwrap(), msg: std::str::from_utf8(m.value).unwrap().parse().unwrap() }; events.push(event); } self.consumer.consume_messageset(ms); } + self.consumer.commit_consumed().unwrap(); events } } diff --git a/server/src/main.rs b/server/src/main.rs index 6cfed44..7914fb0 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -2,6 +2,8 @@ use crate::http::HttpServer; mod http; mod kafka; +pub mod utils; + #[tokio::main] pub async fn main() { diff --git a/server/src/utils.rs b/server/src/utils.rs new file mode 100644 index 0000000..a1fab9a --- /dev/null +++ b/server/src/utils.rs @@ -0,0 +1,15 @@ +pub mod http_utils { + use std::collections::HashMap; + use hyper::{Body, Request}; + + pub fn get_query_params(req: &Request) -> HashMap<&str, &str> { + let uri = req.uri(); + let query = uri.query(); + match query { + None => HashMap::new(), + Some(query) => { + query.split("&").map(|s| s.split_at(query.find("=").unwrap())).collect() + } + } + } +} diff --git a/server/tests/utils_tests.rs b/server/tests/utils_tests.rs new file mode 100644 index 0000000..7cfbe82 --- /dev/null +++ b/server/tests/utils_tests.rs @@ -0,0 +1,14 @@ +mod test { + use std::collections::HashMap; + use rstest::rstest; + + #[rstest] + #[case( + "test=abc", + HashMap::from([("test", "abc")]) + )] + fn get_query_params_tests(#[case] query_str: &str, #[case] expected: HashMap<&str, &str>) { + let res = get_(query_str); + assert_eq!(res, expected) + } +}