From bcea6c6f0dcba068a4ed2a73e7dcbd4b3755a4ea Mon Sep 17 00:00:00 2001
From: Thilo Behnke <thilo.behnke@gmx.net>
Date: Mon, 6 Jun 2022 17:30:53 +0200
Subject: [PATCH] severe isses with concurrency - why is the ws read task
 starving while the write thread can still loop?

---
 server/src/http.rs | 56 ++++++++++++++++++++++++----------------------
 www/index.js       |  2 +-
 2 files changed, 30 insertions(+), 28 deletions(-)

diff --git a/server/src/http.rs b/server/src/http.rs
index eb3adaa..675a31b 100644
--- a/server/src/http.rs
+++ b/server/src/http.rs
@@ -156,33 +156,35 @@ async fn serve_websocket(websocket_session: WebSocketSession, websocket: HyperWe
         }
         println!("!!!! Exit websocket receiver !!!!")
     });
-    // let websocket_session_write_copy = websocket_session.clone();
-    // tokio::spawn(async move {
-    //     println!("Ready to read messages from kafka: {:?}", websocket_session_write_copy);
-    //     loop {
-    //         println!("Reading messages from kafka.");
-    //         let messages = event_reader.read_from_session();
-    //         if let Err(_) = messages {
-    //             eprintln!("Failed to read messages from kafka for session: {:?}", websocket_session_write_copy);
-    //             continue;
-    //         }
-    //         // println!("Read messages for websocket_session {:?} from consumer: {:?}", websocket_session_write_copy, messages);
-    //         let messages = messages.unwrap();
-    //         if messages.len() == 0 {
-    //             println!("No new messages from kafka.");
-    //             continue;
-    //         }
-    //         println!("{} new messages from kafka.", messages.len());
-    //         let json = serde_json::to_string(&messages).unwrap();
-    //         let message = Message::from(json);
-    //         println!("Sending kafka messages through websocket.");
-    //         let send_res = websocket_writer.send(message).await;
-    //         if let Err(e) = send_res {
-    //             eprintln!("Failed to send message to websocket for session {:?}: {:?}", websocket_session_write_copy, e)
-    //         }
-    //         sleep(Duration::from_millis(2000));
-    //     }
-    // });
+    let websocket_session_write_copy = websocket_session.clone();
+    tokio::spawn(async move {
+        println!("Ready to read messages from kafka: {:?}", websocket_session_write_copy);
+        loop {
+            println!("Reading messages from kafka.");
+            let messages = event_reader.read_from_session();
+            if let Err(_) = messages {
+                eprintln!("Failed to read messages from kafka for session: {:?}", websocket_session_write_copy);
+                continue;
+            }
+            // println!("Read messages for websocket_session {:?} from consumer: {:?}", websocket_session_write_copy, messages);
+            let messages = messages.unwrap();
+            if messages.len() == 0 {
+                println!("No new messages from kafka.");
+                continue;
+            }
+            println!("{} new messages from kafka.", messages.len());
+            let json = serde_json::to_string(&messages).unwrap();
+            let message = Message::from(json);
+            println!("Sending kafka messages through websocket.");
+            let send_res = websocket_writer.send(message).await;
+            if let Err(e) = send_res {
+                eprintln!("Failed to send message to websocket for session {:?}: {:?}", websocket_session_write_copy, e)
+            }
+            // Avoid starvation of read thread (?)
+            // TODO: How to avoid this? This is very bad for performance.
+            sleep(Duration::from_millis(10)).await;
+        }
+    });
     Ok(())
 }
 
diff --git a/www/index.js b/www/index.js
index dd58841..2e0d278 100644
--- a/www/index.js
+++ b/www/index.js
@@ -102,7 +102,7 @@ const tick = () => {
         const latestMoveEvents = Object.entries(moveEventsByObj)
             .map(([_, moveEvents]) => moveEvents[moveEvents.length - 1]);
         objects = latestMoveEvents.map(({msg}) => msg);
-        // sendEvents(getInputEvents())
+        sendEvents(getInputEvents())
     }
     render(objects);
 }