diff --git a/client/src/main.rs b/client/src/main.rs index cdbd08d..f2b232b 100644 --- a/client/src/main.rs +++ b/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>, false, true>>>>; + #[tokio::main] async fn main() { let server_address = "127.0.0.1:8080"; + let (global_event_tx, _) = broadcast::channel::(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"); - - 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]; + // tcp_stream.readable().await.expect("idk"); let mut session_client_id: u32 = 0; + let mut session_microphone_stream_id: u8 = 0; - match tcp_stream.try_read(&mut tcp_buf) { - Ok(n) => match serde_json::from_slice::(&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!("new client id: {}", client_id) + let mut global_event_rx = global_event_tx.subscribe(); + + tokio::spawn(async move { + let mut tcp_buf = [0u8; 4096]; + loop { + tokio::select! { + socket_result = tcp_stream.read(&mut tcp_buf) => { + match socket_result { + Ok(0) => { + println!("server shut down"); + break; + } + Ok(n) => match serde_json::from_slice::(&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 stream_id = u32::from_be_bytes(id_bytes); + let client_id_bytes: [u8; 4] = receive_buffer[0..4].try_into().unwrap(); + 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); diff --git a/server/src/main.rs b/server/src/main.rs index b8cbafa..5f82d76 100644 --- a/server/src/main.rs +++ b/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>, } +#[derive(Clone, Debug)] +enum ServerEvent { + NewStream(u32), + UserConnected(u32), + UserDisconnected(u32), + ClosedStream(u32), +} + type State = Arc>>; #[tokio::main] async fn main() { let listen_address = "0.0.0.0:8080"; + let (global_event_tx, _) = broadcast::channel::(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 mut tcp_buf = [0; 4096]; + let tcp_state = state.clone(); - match tcp_stream.try_read(&mut tcp_buf) { - Ok(n) => match serde_json::from_slice::(&tcp_buf[0..n]) { - Ok(request) => match request.data { - TcpRequestData::NewConnection => { - let client_id = next_client_id; - next_client_id += 1; - tcp_state.write().await.insert( - client_id, - Client { - tcp_addr: tcp_address, - udp_addr: None, - audio_streams: None, + let global_event_tx_clone = global_event_tx.clone(); + let mut global_event_rx = global_event_tx.subscribe(); + + tokio::spawn(async move { + let mut current_client_id: u32 = 0; + let mut tcp_buf = [0; 4096]; + + loop { + tokio::select! { + socket_result = tcp_stream.read(&mut tcp_buf) => { + match socket_result { + 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::(&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), }, - ); - 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; + Err(e) => eprintln!("failed to receive tcp packet: {}", e), + } } - _ => 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), - } + event_result = global_event_rx.recv() => { + match event_result { + Ok(event) => { + match event { + 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), + } + } + } + } + }); } } diff --git a/shared/src/lib.rs b/shared/src/lib.rs index b35a7b3..718f412 100644 --- a/shared/src/lib.rs +++ b/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, },