Merge pull request #36 from Tahinli/dev

Dev

back/src/main.rs

@@ -19,7 +19,10 @@ async fn main() {
             .parse::<SocketAddr>()
             .unwrap(),
     );
-    println!("\n\n\tOn Air -> http://{}\n\n", relay_config.axum_address.clone());
+    println!(
+        "\n\n\tOn Air -> http://{}\n\n",
+        relay_config.axum_address.clone()
+    );
     tokio::spawn(streaming::start(relay_config));
     axum_server::bind_rustls(addr, rustls_config)
         .serve(app.into_make_service())

back/src/streaming.rs

@@ -11,27 +11,20 @@ use rustls_pemfile::{certs, pkcs8_private_keys};
 use tokio::{
     net::TcpListener,
     sync::broadcast::{channel, Receiver, Sender},
+    task::JoinHandle,
     time::Instant,
 };
 use tokio_rustls::{
     rustls::pki_types::{CertificateDer, PrivateKeyDer},
     TlsAcceptor,
 };
-use tokio_tungstenite::tungstenite::{Error, Message};
+use tokio_tungstenite::tungstenite::{util::NonBlockingResult, Error, Message};
 
 use crate::{Config, Listener, Streamer};
 
 const BUFFER_LENGTH: usize = 1000000;
 const MAX_TOLERATED_MESSAGE_COUNT: usize = 10;
 pub async fn start(relay_configs: Config) {
-    //need to move them for multi streamer
-    let listener_socket = TcpListener::bind(relay_configs.listener_address)
-        .await
-        .unwrap();
-    let (record_producer, record_consumer) = channel(BUFFER_LENGTH);
-    let streamer_socket = TcpListener::bind(relay_configs.streamer_address)
-        .await
-        .unwrap();
     let timer = Instant::now();
 
     let fullchain: io::Result<Vec<CertificateDer<'static>>> = certs(&mut BufReader::new(
@@ -52,41 +45,72 @@ pub async fn start(relay_configs: Config) {
         .with_single_cert(fullchain, privkey)
         .unwrap();
     let acceptor = TlsAcceptor::from(Arc::new(server_tls_config));
+    loop {
+        //need to move them for multi streamer
+        let listener_socket = TcpListener::bind(relay_configs.listener_address.clone())
+            .await
+            .unwrap();
+        let (record_producer, record_consumer) = channel(BUFFER_LENGTH);
+        let streamer_socket = TcpListener::bind(relay_configs.streamer_address.clone())
+            .await
+            .unwrap();
+
+        let (streamer_alive_producer, streamer_alive_receiver) = tokio::sync::oneshot::channel();
+        let message_organizer_task: Option<JoinHandle<()>>;
+        let buffer_layer_task: Option<JoinHandle<()>>;
+        let (listener_stream_tasks_producer, listener_stream_tasks_receiver) =
+            tokio::sync::mpsc::channel(BUFFER_LENGTH);
+        let mut new_streamer = Streamer {
+            ip: "127.0.0.1".to_string().parse().unwrap(),
+            port: 0000,
+        };
+        let mut is_streaming = false;
         loop {
             match streamer_socket.accept().await {
                 Ok((streamer_tcp, streamer_info)) => {
-                let new_streamer = Streamer {
+                    new_streamer.ip = streamer_info.ip();
-                    ip: streamer_info.ip(),
+                    new_streamer.port = streamer_info.port();
-                    port: streamer_info.port(),
-                };
                     println!(
                         "New Streamer: {:#?} | {:#?}",
                         streamer_info,
                         timer.elapsed()
                     );
                     if relay_configs.tls {
-                    let streamer_tcp_tls = acceptor.accept(streamer_tcp).await.unwrap();
+                        match acceptor.accept(streamer_tcp).await {
+                            Ok(streamer_tcp_tls) => {
                                 match tokio_tungstenite::accept_async(streamer_tcp_tls).await {
                                     Ok(ws_stream) => {
                                         tokio::spawn(streamer_stream(
-                                new_streamer,
+                                            new_streamer.clone(),
                                             record_producer,
                                             ws_stream,
                                             timer,
+                                            streamer_alive_producer,
                                         ));
+                                        is_streaming = true;
+                                        break;
+                                    }
+                                    Err(err_val) => {
+                                        eprintln!("Error: TCP to WS Transform | {}", err_val)
+                                    }
+                                }
+                            }
+                            Err(err_val) => {
+                                eprintln!("Error: TCP TLS Streamer| {}", err_val);
                                 break;
                             }
-                        Err(err_val) => eprintln!("Error: TCP to WS Transform | {}", err_val),
                         }
                     } else {
                         match tokio_tungstenite::accept_async(streamer_tcp).await {
                             Ok(ws_stream) => {
                                 tokio::spawn(streamer_stream(
-                                new_streamer,
+                                    new_streamer.clone(),
                                     record_producer,
                                     ws_stream,
                                     timer,
+                                    streamer_alive_producer,
                                 ));
+                                is_streaming = true;
                                 break;
                             }
                             Err(err_val) => eprintln!("Error: TCP to WS Transform | {}", err_val),
@@ -96,51 +120,174 @@ pub async fn start(relay_configs: Config) {
                 Err(err_val) => eprintln!("Error: TCP Accept Connection | {}", err_val),
             }
         }
+        if is_streaming {
             let (message_producer, message_consumer) = channel(BUFFER_LENGTH);
             let (buffered_producer, _) = channel(BUFFER_LENGTH);
-    tokio::spawn(message_organizer(
+            message_organizer_task = tokio::spawn(message_organizer(
                 message_producer.clone(),
                 record_consumer,
                 relay_configs.latency,
-    ));
+            ))
-    tokio::spawn(buffer_layer(
+            .into();
+            buffer_layer_task = tokio::spawn(buffer_layer(
                 message_consumer,
                 buffered_producer.clone(),
                 relay_configs.latency,
+            ))
+            .into();
+            let (listener_socket_killer_producer, listener_socket_killer_receiver) =
+                tokio::sync::oneshot::channel();
+            let listener_handler_task = tokio::spawn(listener_handler(
+                listener_socket,
+                acceptor.clone(),
+                relay_configs.tls,
+                buffered_producer.clone(),
+                listener_stream_tasks_producer,
+                timer,
+                listener_socket_killer_receiver,
             ));
-    tokio::spawn(status_checker(buffered_producer.clone(), timer));
+            status_checker(
-    while let Ok((tcp_stream, listener_info)) = listener_socket.accept().await {
+                buffered_producer.clone(),
+                timer,
+                new_streamer,
+                streamer_alive_receiver,
+                message_organizer_task,
+                buffer_layer_task,
+                listener_stream_tasks_receiver,
+                listener_handler_task,
+                listener_socket_killer_producer,
+                relay_configs.listener_address.clone(),
+            )
+            .await;
+            drop(streamer_socket);
+        }
+    }
+}
+async fn listener_handler(
+    listener_socket: TcpListener,
+    acceptor: TlsAcceptor,
+    is_tls: bool,
+    buffered_producer: Sender<Message>,
+    listener_stream_tasks_producer: tokio::sync::mpsc::Sender<JoinHandle<()>>,
+    timer: Instant,
+    mut listener_socket_killer_receiver: tokio::sync::oneshot::Receiver<bool>,
+) {
+    while let Err(_) = listener_socket_killer_receiver.try_recv() {
+        match listener_socket.accept().await.no_block() {
+            Ok(accepted_request) => match accepted_request {
+                Some((tcp_stream, listener_info)) => {
                     let new_listener = Listener {
                         ip: listener_info.ip(),
                         port: listener_info.port(),
                     };
-        if relay_configs.tls {
+                    if is_tls {
-            let streamer_tcp_tls = acceptor.accept(tcp_stream).await.unwrap();
+                        match acceptor.accept(tcp_stream).await {
-            let wss_stream = tokio_tungstenite::accept_async(streamer_tcp_tls)
+                            Ok(listener_tcp_tls) => {
-                .await
+                                match tokio_tungstenite::accept_async(listener_tcp_tls).await {
-                .unwrap();
+                                    Ok(wss_stream) => {
-            tokio::spawn(stream(
+                                        let listener_stream_task = tokio::spawn(stream(
                                             new_listener,
                                             wss_stream,
                                             buffered_producer.subscribe(),
                                         ));
+                                        let _ = listener_stream_tasks_producer
+                                            .send(listener_stream_task)
+                                            .await;
+                                    }
+                                    Err(err_val) => {
+                                        eprintln!("Error: TCP WSS Listener | {}", err_val);
+                                        return;
+                                    }
+                                }
+                            }
+                            Err(err_val) => {
+                                eprintln!("Error: TCP TLS Listener | {}", err_val);
+                                return;
+                            }
+                        }
                     } else {
-            let ws_stream = tokio_tungstenite::accept_async(tcp_stream).await.unwrap();
+                        match tokio_tungstenite::accept_async(tcp_stream).await {
-            tokio::spawn(stream(
+                            Ok(ws_stream) => {
+                                let listener_stream_task = tokio::spawn(stream(
                                     new_listener,
                                     ws_stream,
                                     buffered_producer.subscribe(),
                                 ));
+                                let _ = listener_stream_tasks_producer
+                                    .send(listener_stream_task)
+                                    .await;
+                            }
+                            Err(err_val) => {
+                                eprintln!("Error: TCP WS Listener | {}", err_val);
+                                return;
+                            }
+                        }
                     }
                     println!("New Listener: {} | {:#?}", listener_info, timer.elapsed());
                 }
+                None => {}
+            },
+            Err(_) => {}
+        }
+    }
+    drop(listener_socket);
 }
-async fn status_checker(buffered_producer: Sender<Message>, timer: Instant) {
+async fn status_checker(
+    buffered_producer: Sender<Message>,
+    timer: Instant,
+    streamer: Streamer,
+    mut streamer_alive_receiver: tokio::sync::oneshot::Receiver<bool>,
+    message_organizer_task: Option<JoinHandle<()>>,
+    buffer_layer_task: Option<JoinHandle<()>>,
+    mut listener_stream_tasks_receiver: tokio::sync::mpsc::Receiver<JoinHandle<()>>,
+    listener_handler_task: JoinHandle<()>,
+    listener_socket_killer_producer: tokio::sync::oneshot::Sender<bool>,
+    listener_address: String,
+) {
     let mut listener_counter = buffered_producer.receiver_count();
     let mut bottleneck_flag = false;
     //let mut buffer_len = buffered_producer.len();
     loop {
         tokio::time::sleep(Duration::from_secs(3)).await;
+        match streamer_alive_receiver.try_recv() {
+            Ok(_) => {
+                println!(
+                    "Cleaning: Streamer Disconnected | {}",
+                    format!("{}:{}", streamer.ip, streamer.port)
+                );
+                let cleaning_timer = Instant::now();
+                message_organizer_task.as_ref().unwrap().abort();
+                buffer_layer_task.as_ref().unwrap().abort();
+                listener_socket_killer_producer.send(true).unwrap();
+                let mut listener_task_counter = 0;
+                while listener_stream_tasks_receiver.len() > 0 {
+                    match listener_stream_tasks_receiver.recv().await {
+                        Some(listener_stream_task) => {
+                            listener_stream_task.abort();
+                            listener_task_counter += 1;
+                        }
+                        None => {}
+                    }
+                }
+                if !listener_handler_task.is_finished() {
+                    listener_handler_task.abort();
+                    println!("Cleaning: Listener Handler Killed");
+                }
+                while TcpListener::bind(listener_address.clone()).await.is_err() {
+                    tokio::time::sleep(Duration::from_millis(1)).await;
+                }
+                println!("Cleaning: Listener Socket Killed | {}", listener_address);
+                println!(
+                    "Cleaning Done: Streamer Disconnected | {} | Disconnected Listener(s) = {} | {:#?}",
+                    format!("{}:{}", streamer.ip, streamer.port),
+                    listener_task_counter,
+                    cleaning_timer.elapsed()
+                );
+                return;
+            }
+            Err(_) => {}
+        }
+
         if buffered_producer.receiver_count() != 0 {
             if buffered_producer.len() > 2 {
                 bottleneck_flag = true;
@@ -186,6 +333,7 @@ async fn streamer_stream<
     record_producer: Sender<Message>,
     mut ws_stream: T,
     timer: Instant,
+    streamer_alive_producer: tokio::sync::oneshot::Sender<bool>,
 ) {
     loop {
         match ws_stream.next().await {
@@ -207,6 +355,7 @@ async fn streamer_stream<
                     format!("{}:{}", streamer.ip, streamer.port),
                     timer.elapsed()
                 );
+                streamer_alive_producer.send(false).unwrap();
                 return;
             }
         }

front/Cargo.toml

@@ -12,11 +12,9 @@ cpal = { version = "0.15.3", features = ["wasm-bindgen"] }
 dioxus = { version = "0.5.1", features = ["web"] }
 futures-core = "0.3.30"
 futures-util = { version = "0.3.30", features = ["futures-sink", "sink"] }
-#getrandom = { version = "0.2.14", features = ["js"] }
 log = "0.4.21"
 reqwest = { version = "0.12.2", features = ["json"] }
 ringbuf = "0.3.3"
-#rustls-platform-verifier = "0.2.0"
 serde = { version = "1.0.197", features = ["derive"] }
 tokio-tungstenite-wasm = { version = "0.3.1", features = ["rustls-tls-webpki-roots"] }
 tokio_with_wasm = "0.4.3"

front/src/streaming.rs

@@ -16,7 +16,7 @@ pub async fn start_listening(
 ) {
     if is_listening() {
         log::info!("Trying Sir");
-        let connect_addr = "wss://tahinli.com.tr:2424";
+        let connect_addr = "ws://192.168.1.2:2424";
         
         let ws_stream: tokio_tungstenite_wasm::WebSocketStream;
         match tokio_tungstenite_wasm::connect(