issue with sending event writer between threads

2026-08-19 05:06:14 +00:00 · 2022-05-12 22:55:29 +02:00
parent b999669efe
commit c0d72a631c
7 changed files with 65 additions and 14 deletions
@@ -10,12 +10,12 @@ pub mod event {
    }

    pub trait EventWriterImpl {
-        fn write(&self, event: Event) -> Result<(), ()>;
+        fn write(&mut self, event: Event) -> Result<(), ()>;
    }

    pub struct FileEventWriterImpl {}
    impl EventWriterImpl for FileEventWriterImpl {
-        fn write(&self, event: Event) -> Result<(), ()> {
+        fn write(&mut self, event: Event) -> Result<(), ()> {
            let options = OpenOptions::new().read(true).create(true).write(true).open("events.log");
            if let Err(_) = options {
                return Err(());
@@ -30,7 +30,7 @@ pub mod event {

    pub struct NoopEventWriterImpl {}
    impl EventWriterImpl for NoopEventWriterImpl {
-        fn write(&self, event: Event) -> Result<(), ()> {
+        fn write(&mut self, event: Event) -> Result<(), ()> {
            todo!()
        }
    }
@@ -40,7 +40,7 @@ pub mod event {
    }

    impl EventWriter {
-        fn new(writer_impl: Box<dyn EventWriterImpl>) -> EventWriter {
+        pub fn new(writer_impl: Box<dyn EventWriterImpl>) -> EventWriter {
            EventWriter {
                writer_impl
            }
@@ -57,9 +57,8 @@ pub mod event {
                writer_impl: Box::new(FileEventWriterImpl {})
            }
        }
-        // TODO: Kafka

-        pub fn write(&self, event: Event) -> Result<(), ()> {
+        pub fn write(&mut self, event: Event) -> Result<(), ()> {
           self.writer_impl.write(event)
        }
    }
@@ -115,7 +115,7 @@ impl Field {
        self.objs.push(Rc::new(RefCell::new(ball)));
    }

-    pub fn tick(&self, inputs: Vec<Input>) {
+    pub fn tick(&mut self, inputs: Vec<Input>) {
        for obj in self.objs.iter() {
            let mut obj_mut = RefCell::borrow_mut(obj);
            if obj_mut.obj_type() != "ball" {
@@ -83,7 +83,7 @@ pub mod pong_events {
    }

    pub trait PongEventWriter {
-        fn write(&self, event: PongEventType) -> Result<(), ()>;
+        fn write(&mut self, event: PongEventType) -> Result<(), ()>;
    }

    pub struct DefaultPongEventWriter {
@@ -91,7 +91,7 @@ pub mod pong_events {
    }

    impl PongEventWriter for DefaultPongEventWriter {
-        fn write(&self, event: PongEventType) -> Result<(), ()> {
+        fn write(&mut self, event: PongEventType) -> Result<(), ()> {
            let out_event = match event {
                PongEventType::GameObjUpdate(ref update) => {
                    Event {
@@ -10,3 +10,4 @@ kafka = { version = "0.8.0" }
 hyper = {version = "0.14.18", features = ["full"]}
 tokio = { version = "1", features = ["full"] }
 tokio-stream = {version = "0.1" }
+pong = { path = "../pong", version = "0.1.0" }
@@ -1,20 +1,33 @@
 use std::convert::Infallible;
+use std::sync::Arc;
 use hyper::{Body, Request, Response, Server};
 use hyper::server::conn::AddrStream;
 use hyper::service::{make_service_fn, service_fn};
+use kafka::producer::Producer;
+use tokio::sync::Mutex;
+use pong::event::event::{Event, EventWriter};
+use crate::kafka::KafkaEventWriterImpl;

 pub struct HttpServer {
    addr: [u8; 4],
-    port: u16
+    port: u16,
+    event_writer: Arc<Mutex<EventWriter>>
 }
 impl HttpServer {
    pub fn new(addr: [u8; 4], port: u16) -> HttpServer {
-        HttpServer {addr, port}
+        let event_writer = Arc::new(Mutex::new(EventWriter::new(Box::new(KafkaEventWriterImpl::default()))));
+        HttpServer {addr, port, event_writer}
    }

    pub async fn run(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>>  {
+        let mut event_writer = Arc::clone(&self.event_writer);
        let make_svc = make_service_fn(|socket: &AddrStream| async {
-            Ok::<_, Infallible>(service_fn(handle_request))
+            Ok::<_, Infallible>(service_fn(move |req: Request<Body>| {
+                async move {
+                    let mut event_writer = Arc::clone(&event_writer);
+                    handle_request(&event_writer, req).await
+                }
+            }))
        });

        let host = (self.addr, self.port).into();
@@ -26,10 +39,15 @@ impl HttpServer {
    }
 }

-async fn handle_request(req: Request<Body>) -> Result<Response<Body>, Infallible> {
-    Ok(Response::new("hello".into()))
+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);
+    }
+    Ok(Response::new("response".into()))
 }

+
 async fn shutdown_signal() {
    // Wait for the CTRL+C signal
    tokio::signal::ctrl_c()
@@ -0,0 +1,32 @@
+use std::time::Duration;
+use kafka::producer::{Producer, Record, RequiredAcks};
+use pong::event::event::{Event, EventWriter, EventWriterImpl};
+
+pub struct KafkaEventWriterImpl {
+    producer: Producer
+}
+impl KafkaEventWriterImpl {
+    pub fn default() -> KafkaEventWriterImpl {
+        KafkaEventWriterImpl::new("localhost:9092")
+    }
+
+    pub fn new(host: &str) -> KafkaEventWriterImpl {
+        let mut producer = Producer::from_hosts(vec![host.to_owned()])
+            .with_ack_timeout(Duration::from_secs(1))
+            .with_required_acks(RequiredAcks::One)
+            .create()
+            .unwrap();
+        KafkaEventWriterImpl {
+            producer
+        }
+    }
+}
+impl EventWriterImpl for KafkaEventWriterImpl {
+    fn write(&mut self, event: Event) -> Result<(), ()> {
+        let record = Record::from_key_value(event.topic.as_str(), event.key.as_str(), event.msg.as_str());
+        match self.producer.send(&record) {
+            Ok(()) => Ok(()),
+            Err(_) => Err(())
+        }
+    }
+}
@@ -1,6 +1,7 @@
 use crate::http::HttpServer;

 mod http;
+mod kafka;

 #[tokio::main]
 pub async fn main() {