github/wasm-pong
1
0
Fork 0
mirror of https://github.com/thilo-behnke/wasm-pong.git synced 2026-07-11 02:29:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

kafka consumer

Browse Source
This commit is contained in:
Thilo Behnke
2022-05-13 16:25:46 +02:00
parent 6c3c75d42f
commit 0cfc19fa57
3 changed files with 63 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
pong/src/event.rs
View File

@@ -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<Event>;
}
pub struct EventReader {
reader_impl: Box<dyn EventReaderImpl>
}
impl EventReader {
pub fn new(reader_impl: Box<dyn EventReaderImpl>) -> EventReader {
EventReader {
reader_impl
}
}
pub fn read(&mut self) -> Event<Event> {
self.reader_impl.read()
}
}
}

6
server/src/http.rs
View File

@@ -43,8 +43,10 @@ impl HttpServer {
async fn handle_request(event_writer: &Arc<Mutex<EventWriter>>, req: Request<Body>) -> Result<Response<Body>, 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()))
}

39
server/src/kafka.rs
View File

@@ -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<Event> {
let polled = self.consumer.poll().unwrap();
let message_sets: Vec<MessageSet<'_>> = 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
}
}