preserving tcp connection

client/src/main.rs

@@ -2,57 +2,110 @@ use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
 use ringbuf::{
     HeapRb,
     traits::{Consumer, Observer, Producer, Split},
+    wrap::caching::Caching,
 };
 use shared::{ResponseStatus, TcpRequest, TcpResponse, TcpResponseData};
-use std::sync::Arc;
+use std::{collections::HashMap, sync::Arc};
+use tokio::sync::{Mutex, broadcast};
 use tokio::{
-    io::AsyncWriteExt,
+    io::{AsyncReadExt, AsyncWriteExt},
     net::{TcpStream, UdpSocket},
 };
 
+#[derive(Clone, Debug)]
+enum ClientEvent {
+    NewConnection,
+}
+
+type AudioStreamKey = (u32, u8);
+
+type AudioMixerStreams =
+    Arc<Mutex<HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, false, true>>>>;
+
 #[tokio::main]
 async fn main() {
     let server_address = "127.0.0.1:8080";
 
+    let (global_event_tx, _) = broadcast::channel::<ClientEvent>(16);
+
+    let active_streams: AudioMixerStreams = Arc::new(Mutex::new(HashMap::new()));
+    let active_streams_cpal = active_streams.clone();
+    let active_streams_socket = active_streams.clone();
+
     let mut tcp_stream = TcpStream::connect(server_address)
         .await
         .expect("failed to connect using tcp");
-
+    // tcp_stream.readable().await.expect("idk");
-    let _ = tcp_stream
-        .write_all(
-            serde_json::to_string(&TcpRequest {
-                client_id: 0,
-                data: shared::TcpRequestData::NewConnection,
-            })
-            .expect("failed to parse initial request")
-            .as_bytes(),
-        )
-        .await
-        .expect("failed to send initial message");
-
-    tcp_stream.readable().await.expect("idk");
-    let mut tcp_buf = [0; 4096];
 
     let mut session_client_id: u32 = 0;
+    let mut session_microphone_stream_id: u8 = 0;
 
-    match tcp_stream.try_read(&mut tcp_buf) {
+    let mut global_event_rx = global_event_tx.subscribe();
-        Ok(n) => match serde_json::from_slice::<TcpResponse>(&tcp_buf[0..n]) {
+
-            Ok(response) => {
+    tokio::spawn(async move {
-                println!("connection Ok: {}", response.status == ResponseStatus::Ok);
+        let mut tcp_buf = [0u8; 4096];
-                match response.data {
+        loop {
-                    TcpResponseData::NewConnection { client_id } => {
+            tokio::select! {
-                        session_client_id = client_id;
+                socket_result = tcp_stream.read(&mut tcp_buf) => {
-                        println!("new client id: {}", client_id)
+                    match socket_result {
+                        Ok(0) => {
+                            println!("server shut down");
+                            break;
+                        }
+                        Ok(n) => match serde_json::from_slice::<TcpResponse>(&tcp_buf[0..n]) {
+                            Ok(response) => {
+                                println!("connection Ok: {}", response.status == ResponseStatus::Ok);
+                                match response.data {
+                                    TcpResponseData::NewConnection { client_id } => {
+                                        session_client_id = client_id;
+                                        println!("fetched client id: {}", client_id);
+                                    }
+                                    TcpResponseData::UserConnected { user_id } => {
+                                        println!("new user connected: {}", user_id);
+                                    }
+                                    TcpResponseData::UserDisconnected { user_id } => {
+                                        println!("user disconnected: {}", user_id);
+                                    }
+                                    _ => {
+                                        println!("wrong response")
+                                    }
+                                }
+                            }
+                            Err(e) => eprintln!("failed to parse tcp packet: {}", e),
+                        },
+                        Err(e) => eprintln!("failed to receive tcp packet: {}", e),
                     }
-                    _ => {
+                }
-                        println!("wrong response")
+                event_result = global_event_rx.recv() => {
+                    match event_result {
+                        Ok(event) => {
+                            match event {
+                                ClientEvent::NewConnection => {
+                                    let _ = tcp_stream
+                                        .write_all(
+                                            serde_json::to_string(&TcpRequest {
+                                                client_id: 0,
+                                                data: shared::TcpRequestData::NewConnection,
+                                            })
+                                            .expect("failed to parse initial request")
+                                            .as_bytes(),
+                                        )
+                                        .await
+                                        .expect("failed to send initial message");
+                                }
+                                _ => {
+                                    println!("aaaa")
+                                }
+                            }
+                        }
+                        Err(e) => eprintln!("error: {}", e),
                     }
                 }
             }
-            Err(e) => eprintln!("failed to parse tcp packet: {}", e),
+        }
-        },
+    });
-        Err(e) => eprintln!("failed to receive tcp packet: {}", e),
+
-    }
+    let _ = global_event_tx.send(ClientEvent::NewConnection);
 
     let socket = Arc::new(UdpSocket::bind("0.0.0.0:0").await.unwrap());
 
@@ -128,6 +181,7 @@ async fn main() {
                 let mut payload = Vec::with_capacity(frames_needed * 4);
 
                 payload.extend(session_client_id.to_be_bytes());
+                payload.extend(session_microphone_stream_id.to_be_bytes());
 
                 for sample in &frame_buffer {
                     payload.extend_from_slice(&sample.to_le_bytes());
@@ -150,10 +204,13 @@ async fn main() {
                         continue;
                     }
 
-                    let id_bytes: [u8; 4] = receive_buffer[0..4].try_into().unwrap();
+                    let client_id_bytes: [u8; 4] = receive_buffer[0..4].try_into().unwrap();
-                    let stream_id = u32::from_be_bytes(id_bytes);
+                    let client_id = u32::from_be_bytes(client_id_bytes);
 
-                    let audio_packet = &receive_buffer[4..len];
+                    let stream_id_bytes = receive_buffer[4];
+                    let stream_id = u8::from_be(stream_id_bytes);
+
+                    let audio_packet = &receive_buffer[5..len];
 
                     let mut decoded_samples = Vec::with_capacity(audio_packet.len() / 4);
 

server/src/main.rs

@@ -1,10 +1,10 @@
 use shared::{AudioStreamInfo, ResponseStatus, TcpRequest, TcpRequestData, TcpResponse};
 use std::sync::Arc;
 use std::{collections::HashMap, net::SocketAddr};
-use tokio::io::AsyncWriteExt;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::{
     net::{TcpListener, UdpSocket},
-    sync::RwLock,
+    sync::{RwLock, broadcast},
 };
 
 struct Client {
@@ -13,15 +13,24 @@ struct Client {
     audio_streams: Option<Vec<AudioStreamInfo>>,
 }
 
+#[derive(Clone, Debug)]
+enum ServerEvent {
+    NewStream(u32),
+    UserConnected(u32),
+    UserDisconnected(u32),
+    ClosedStream(u32),
+}
+
 type State = Arc<RwLock<HashMap<u32, Client>>>;
 
 #[tokio::main]
 async fn main() {
     let listen_address = "0.0.0.0:8080";
 
+    let (global_event_tx, _) = broadcast::channel::<ServerEvent>(16);
+
     let state: State = Arc::new(RwLock::new(HashMap::new()));
 
-    let tcp_state = state.clone();
     let udp_state = state.clone();
 
     let tcp_listener = TcpListener::bind(listen_address)
@@ -67,42 +76,101 @@ async fn main() {
         let (mut tcp_stream, tcp_address) =
             tcp_listener.accept().await.expect("failed to accept tcp");
 
-        tcp_stream.readable().await.expect("idk");
+        let tcp_state = state.clone();
-        let mut tcp_buf = [0; 4096];
 
-        match tcp_stream.try_read(&mut tcp_buf) {
+        let global_event_tx_clone = global_event_tx.clone();
-            Ok(n) => match serde_json::from_slice::<TcpRequest>(&tcp_buf[0..n]) {
+        let mut global_event_rx = global_event_tx.subscribe();
-                Ok(request) => match request.data {
+
-                    TcpRequestData::NewConnection => {
+        tokio::spawn(async move {
-                        let client_id = next_client_id;
+            let mut current_client_id: u32 = 0;
-                        next_client_id += 1;
+            let mut tcp_buf = [0; 4096];
-                        tcp_state.write().await.insert(
+
-                            client_id,
+            loop {
-                            Client {
+                tokio::select! {
-                                tcp_addr: tcp_address,
+                    socket_result = tcp_stream.read(&mut tcp_buf) => {
-                                udp_addr: None,
+                        match socket_result {
-                                audio_streams: None,
+                            Ok(0) => {
+                                let _ = global_event_tx_clone.send(ServerEvent::UserDisconnected(current_client_id));
+                                println!("client disconnected: {}", tcp_address);
+                                break;
+                            }
+                            Ok(n) => match serde_json::from_slice::<TcpRequest>(&tcp_buf[0..n]) {
+                                Ok(request) => match request.data {
+                                    TcpRequestData::NewConnection => {
+                                        current_client_id = next_client_id;
+                                        next_client_id += 1;
+                                        tcp_state.write().await.insert(
+                                            current_client_id,
+                                            Client {
+                                                tcp_addr: tcp_address,
+                                                udp_addr: None,
+                                                audio_streams: None,
+                                            },
+                                        );
+                                        let _ = global_event_tx_clone.send(ServerEvent::UserConnected(current_client_id));
+                                        println!("new client: {}, from: {}", current_client_id, tcp_address);
+                                        let _ = tcp_stream
+                                            .write_all(
+                                                serde_json::to_string(&TcpResponse {
+                                                    status: ResponseStatus::Ok,
+                                                    data: shared::TcpResponseData::NewConnection {
+                                                        client_id: current_client_id,
+                                                    },
+                                                })
+                                                .expect("failed to parse response json TcpRequestData::NewConnection")
+                                                .as_bytes(),
+                                            )
+                                            .await;
+                                    }
+                                    _ => eprintln!("i dont support other requests yet"),
+                                },
+                                Err(e) => eprintln!("failed to parse tcp packet: {}", e),
                             },
-                        );
+                            Err(e) => eprintln!("failed to receive tcp packet: {}", e),
-                        println!("new client: {}, from: {}", client_id, tcp_address);
+                        }
-                        let _ = tcp_stream
-                            .write_all(
-                                serde_json::to_string(&TcpResponse {
-                                    status: ResponseStatus::Ok,
-                                    data: shared::TcpResponseData::NewConnection {
-                                        client_id: client_id,
-                                    },
-                                })
-                                .expect("failed to parse response json")
-                                .as_bytes(),
-                            )
-                            .await;
                     }
-                    _ => eprintln!("i dont support other requests yet"),
+                    event_result = global_event_rx.recv() => {
-                },
+                        match event_result {
-                Err(e) => eprintln!("failed to parse tcp packet: {}", e),
+                            Ok(event) => {
-            },
+                                match event {
-            Err(e) => eprintln!("failed to receive tcp packet: {}", e),
+                                    ServerEvent::UserConnected(new_user_id) => {
-        }
+                                        let _ = tcp_stream
+                                            .write_all(
+                                                serde_json::to_string(&TcpResponse {
+                                                    status: ResponseStatus::Ok,
+                                                    data: shared::TcpResponseData::UserConnected {
+                                                        user_id: new_user_id
+                                                    },
+                                                })
+                                                .expect("failed to parse response json ServerEvent::UserConnected")
+                                                .as_bytes(),
+                                            )
+                                            .await;
+                                    }
+                                    ServerEvent::UserDisconnected(user_id) => {
+                                        let _ = tcp_stream
+                                            .write_all(
+                                                serde_json::to_string(&TcpResponse {
+                                                    status: ResponseStatus::Ok,
+                                                    data: shared::TcpResponseData::UserDisconnected {
+                                                        user_id: user_id
+                                                    },
+                                                })
+                                                .expect("failed to parse response json ServerEvent::UserConnected")
+                                                .as_bytes(),
+                                            )
+                                            .await;
+                                    }
+                                    _ => {
+                                        println!("lkdahjsf");
+                                    }
+                                }
+                            }
+                            Err(e) => eprintln!("error: {}", e),
+                        }
+                    }
+                }
+            }
+        });
     }
 }

shared/src/lib.rs

@@ -8,7 +8,7 @@ pub enum ResponseStatus {
 
 #[derive(Serialize, Deserialize)]
 pub struct AudioStreamInfo {
-    pub stream_id: u32,
+    pub stream_id: u8,
     pub channels: u8,
     pub sample_rate: u32,
 }
@@ -25,7 +25,7 @@ pub enum TcpRequestData {
     NewConnection,
     DropConnection,
     NewAudioStream { audio_stream_info: AudioStreamInfo },
-    DropAudioStream { audio_stream_id: u32 },
+    DropAudioStream { audio_stream_id: u8 },
 }
 
 #[derive(Serialize, Deserialize)]
@@ -44,9 +44,15 @@ pub enum TcpResponseData {
     UserJoined {
         user_id: u32,
     },
+    UserConnected {
+        user_id: u32,
+    },
     UserDisconnected {
         user_id: u32,
     },
+    NewAudioStream {
+        audio_stream_id: u8,
+    },
     AudioStreamsUpdate {
         audio_stream_info: Vec<AudioStreamInfo>,
     },