feat: ✨ websocket split

server/src/signal.rs

@@ -1,13 +1,27 @@
+use std::{collections::VecDeque, sync::LazyLock};
+
 use axum::{Router, http::StatusCode, response::IntoResponse, routing::get};
 use fastwebsockets::{
-    OpCode,
+    FragmentCollectorRead, OpCode, WebSocketError,
     upgrade::{IncomingUpgrade, UpgradeFut},
 };
-use tokio::net::TcpListener;
+use protocol::{Signal, SignalType};
+use tokio::{
+    net::TcpListener,
+    sync::{RwLock, broadcast},
+};
 use tower_http::cors::CorsLayer;
 
 const SERVER_ADDRESS: &str = "192.168.1.3:4546";
 
+static USER_MESSAGES: LazyLock<RwLock<VecDeque<UserMessages>>> =
+    LazyLock::new(|| VecDeque::new().into());
+
+struct UserMessages {
+    user: String,
+    message_receiver: broadcast::Receiver<Signal>,
+}
+
 pub async fn start_signalling() {
     let router = Router::new()
         .route("/", get(alive))
@@ -32,15 +46,33 @@ async fn signal(websocket: IncomingUpgrade) -> impl IntoResponse {
 
 async fn websocket_handler(websocket: UpgradeFut) {
     let mut websocket = websocket.await.unwrap();
-
     websocket.set_auto_pong(true);
     websocket.set_writev(false);
     websocket.set_auto_close(true);
 
-    while let Ok(received_frame) = websocket.read_frame().await {
+    let (mut websocket_receiver, websocker_sender) = websocket.split(tokio::io::split);
+    let mut user = String::default();
+    let (message_sender, message_receiver) = broadcast::channel(100);
+
+    if let Ok(received_frame) = websocket_receiver
+        .read_frame::<_, WebSocketError>(&mut move |_| async { unreachable!() })
+        .await
+    {
         if let OpCode::Text = received_frame.opcode {
-            let received_payload = received_frame.payload;
-            println!("Sent:\n{:#?}", received_payload);
+            let signal =
+                serde_json::from_slice::<Signal>(&received_frame.payload.to_vec()).unwrap();
+
+            if signal.get_signal_type() == SignalType::Auth && user == String::default() {
+                let new_user = UserMessages {
+                    user: signal.get_data(),
+                    message_receiver,
+                };
+                user = new_user.user.to_owned();
+                USER_MESSAGES.write().await.push_back(new_user);
+            } else {
+                return;
+            }
+            println!("{:#?}", signal);
         }
     }
 }