From 0cfc19fa57e99e38674bef8c40263efa9bcdc7f7 Mon Sep 17 00:00:00 2001 From: Thilo Behnke Date: Fri, 13 May 2022 16:25:46 +0200 Subject: [PATCH] kafka consumer --- pong/src/event.rs | 21 +++++++++++++++++++++ server/src/http.rs | 6 ++++-- server/src/kafka.rs | 39 ++++++++++++++++++++++++++++++++++++++- 3 files changed, 63 insertions(+), 3 deletions(-) diff --git a/pong/src/event.rs b/pong/src/event.rs index 5209851..75f5fea 100644 --- a/pong/src/event.rs +++ b/pong/src/event.rs @@ -3,6 +3,7 @@ pub mod event { use std::fs::OpenOptions; use std::io::Write; + #[derive(Debug)] pub struct Event { pub topic: String, pub key: String, @@ -62,4 +63,24 @@ pub mod event { self.writer_impl.write(event) } } + + pub trait EventReaderImpl { + fn read(&mut self) -> Vec; + } + + pub struct EventReader { + reader_impl: Box + } + + impl EventReader { + pub fn new(reader_impl: Box) -> EventReader { + EventReader { + reader_impl + } + } + + pub fn read(&mut self) -> Event { + self.reader_impl.read() + } + } } diff --git a/server/src/http.rs b/server/src/http.rs index 20815a8..551ae5a 100644 --- a/server/src/http.rs +++ b/server/src/http.rs @@ -43,8 +43,10 @@ impl HttpServer { async fn handle_request(event_writer: &Arc>, req: Request) -> Result, Infallible> { let mut locked = event_writer.lock().await; - if let err = locked.write(Event {topic: "topic".into(), key: "key".into(), msg: "msg".into()}) { - println!("Failed to write to kafka! {:?}", err); + let event = Event {topic: "topic".into(), key: "key".into(), msg: "msg".into()}; + println!("Writing event to kafka: {:?}", event); + if let Err(e) = locked.write(event) { + println!("Failed to write to kafka! {:?}", e); } Ok(Response::new("response".into())) } diff --git a/server/src/kafka.rs b/server/src/kafka.rs index 2f81636..1534f60 100644 --- a/server/src/kafka.rs +++ b/server/src/kafka.rs @@ -1,6 +1,7 @@ use std::time::Duration; +use kafka::consumer::{Consumer, FetchOffset, GroupOffsetStorage, MessageSet}; use kafka::producer::{Producer, Record, RequiredAcks}; -use pong::event::event::{Event, EventWriter, EventWriterImpl}; +use pong::event::event::{Event, EventReaderImpl, EventWriter, EventWriterImpl}; pub struct KafkaEventWriterImpl { producer: Producer @@ -30,3 +31,39 @@ impl EventWriterImpl for KafkaEventWriterImpl { } } } + +pub struct KafkaEventReaderImpl { + consumer: Consumer +} +impl KafkaEventReaderImpl { + pub fn default() -> KafkaEventReaderImpl { + KafkaEventReaderImpl::new("localhost:9092") + } + + pub fn new(host: &str) -> KafkaEventReaderImpl { + let mut consumer = Consumer::from_hosts(vec![host.to_owned()]) + .with_topic("topic".to_owned()) + .with_fallback_offset(FetchOffset::Earliest) + .with_group("group".to_owned()) + .with_offset_storage(GroupOffsetStorage::Kafka) + .create() + .unwrap(); + KafkaEventReaderImpl { + consumer + } + } +} +impl EventReaderImpl for KafkaEventReaderImpl { + fn read(&mut self) -> Vec { + let polled = self.consumer.poll().unwrap(); + let message_sets: Vec> = polled.iter().collect(); + let mut events = vec![]; + for ms in message_sets { + 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() }; + events.push(event); + } + } + events + } +}