perf: ⚡ compression

feat: ⚗️ buffer layer ignored

back/src/streaming.rs

@@ -13,7 +13,6 @@ const BUFFER_LENGTH: usize = 1000000;
 const MAX_TOLERATED_MESSAGE_COUNT: usize = 10;
 pub async fn start() {
     let socket = TcpListener::bind("192.168.1.2:2424").await.unwrap();
-    println!("Dude Someone Triggered");
     let (record_producer, record_consumer) = channel(BUFFER_LENGTH);
     let streamer_socket = TcpListener::bind("192.168.1.2:2525").await.unwrap();
     if let Ok((streamer_tcp, streamer_info)) = streamer_socket.accept().await {
@@ -26,6 +25,7 @@ pub async fn start() {
     let (buffered_producer, _) = channel(BUFFER_LENGTH);
     tokio::spawn(message_organizer(message_producer.clone(), record_consumer));
     tokio::spawn(buffer_layer(message_consumer, buffered_producer.clone()));
+    tokio::spawn(status_checker(buffered_producer.clone()));
     while let Ok((tcp_stream, listener_info)) = socket.accept().await {
         let ws_stream = tokio_tungstenite::accept_async(tcp_stream).await.unwrap();
         println!("New Listener: {}", listener_info);
@@ -40,25 +40,34 @@ pub async fn start() {
         ));
     }
 }
+async fn status_checker(buffered_producer: Sender<Message>) {
+    let mut listener_counter = buffered_producer.receiver_count();
+    //let mut buffer_len = buffered_producer.len();
+    loop {
+        tokio::time::sleep(Duration::from_secs(3)).await;
+        if buffered_producer.receiver_count() != 0 {
+            if buffered_producer.len() > 2 {
+                println!("Bottleneck: {}", buffered_producer.len());
+            }
+            if listener_counter != buffered_producer.receiver_count() {
+                listener_counter = buffered_producer.receiver_count();
+                println!("Listener(s): {}", listener_counter);
+            }    
+        }
+        
+    }
+}
 async fn buffer_layer(mut message_consumer: Receiver<Message>, buffered_producer: Sender<Message>) {
     loop {
-        tokio::time::sleep(Duration::from_millis(200)).await;
-        let mut messages_buffered: Vec<Message> = vec![];
+        tokio::time::sleep(Duration::from_millis(50)).await;
         while message_consumer.len() > 0 {
             match message_consumer.recv().await {
-                Ok(msg) => {
-                    messages_buffered.push(msg);
-                }
-                Err(_) => {}
-            }
-        }
-        let mut message_buffered_concreted = String::new();
-        if messages_buffered.len() > 0 {
-            for message_buffered in messages_buffered {
-                message_buffered_concreted = format!("{}{}", message_buffered_concreted, message_buffered);
-            }
-            match buffered_producer.send(message_buffered_concreted.into()) {
-                Ok(_) => {}
+                Ok(message) => match buffered_producer.send(message) {
+                    Ok(_) => {
+                        
+                    }
+                    Err(_) => {}
+                },
                 Err(_) => {}
             }
         }
@@ -71,7 +80,7 @@ async fn streamer_stream(
     while let Some(message_with_question) = ws_stream.next().await {
         match message_with_question {
             Ok(message) => {
-                println!("{}", message.len());
+                //println!("{}", message.len());
                 match record_producer.send(message) {
                     Ok(_) => {}
                     Err(_) => {}
@@ -87,30 +96,14 @@ async fn message_organizer(
     mut record_consumer: Receiver<Message>,
 ) {
     loop {
-        let mut messages = String::new();
-        let mut iteration = record_consumer.len();
-        while iteration > 0 {
-            iteration -= 1;
-            match record_consumer.recv().await {
-                Ok(single_message) => {
-                    let single_message_packet = single_message.to_string();
-                    messages = format!("{}{}", messages, single_message_packet);
-                }
-                Err(_) => {}
-            }
-        }
-        if messages.len() > 0 {
-            match message_producer.send(messages.into()) {
+        match record_consumer.recv().await {
+            Ok(single_message) => match message_producer.send(single_message) {
                 Ok(_) => {}
                 Err(_) => {}
-            }
-            println!(
-                "Message Counter = {} | Receiver Count = {}",
-                message_producer.len(),
-                message_producer.receiver_count()
-            );
+            },
+            Err(_) => {}
         }
-        tokio::time::sleep(Duration::from_millis(10)).await;
+        tokio::time::sleep(Duration::from_millis(50)).await;
     }
 }
 async fn stream(
@@ -123,9 +116,10 @@ async fn stream(
             println!(
                 "{} Forced to Disconnect | Reason -> Slow Consumer",
                 format!("{}:{}", listener.ip, listener.port)
-            );
+            );      
             break;
         }
+
         match ws_stream.send(message).await {
             Ok(_) => {
                 if let Err(_) = ws_stream.flush().await {