diff --git a/Cargo.lock b/Cargo.lock index c265b6a..2f331af 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -79,6 +79,7 @@ name = "client" version = "0.1.0" dependencies = [ "cpal", + "rand", "ringbuf", "serde", "serde_json", diff --git a/client/Cargo.toml b/client/Cargo.toml index 1022602..b601ec7 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -10,3 +10,4 @@ ringbuf = "0.5.1" serde = { version = "1.0.229", features = ["derive"] } serde_json = "1.0.151" tokio = { version = "1.53.1", features = ["full"] } +rand = "0.10.2" diff --git a/client/src/main.rs b/client/src/main.rs index f2b232b..e5f65fb 100644 --- a/client/src/main.rs +++ b/client/src/main.rs @@ -1,63 +1,106 @@ -use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; +use cpal::{ + Device, + traits::{DeviceTrait, HostTrait, StreamTrait}, +}; use ringbuf::{ - HeapRb, + HeapRb, consumer, traits::{Consumer, Observer, Producer, Split}, wrap::caching::Caching, }; -use shared::{ResponseStatus, TcpRequest, TcpResponse, TcpResponseData}; -use std::{collections::HashMap, sync::Arc}; -use tokio::sync::{Mutex, broadcast}; +use shared::{AudioStreamInfo, ResponseStatus, TcpRequest, TcpResponse, TcpResponseData}; +use std::{ + collections::HashMap, + sync::{Arc, Mutex, RwLock}, +}; +use tokio::sync::broadcast::{self, Receiver, Sender}; use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, net::{TcpStream, UdpSocket}, }; -#[derive(Clone, Debug)] +#[derive(Clone)] enum ClientEvent { NewConnection, + NewAudioStreamRequest(AudioStreamInfo), + NewAudioStreamAccepted(u8), } type AudioStreamKey = (u32, u8); -type AudioMixerStreams = +type InputAudioStreamsProducers = + Arc>, true, false>>>>; +type InputAudioStreamsConsumers = Arc>, false, true>>>>; +type InputAudioStreamsInfo = Arc>>; + +struct ClientInfo { + client_id: u32, + server_address: &'static str, + connected: bool, +} + #[tokio::main] async fn main() { - let server_address = "127.0.0.1:8080"; + let client_info: Arc> = Arc::new(RwLock::new(ClientInfo { + client_id: 0, + server_address: "127.0.0.1:8080", + connected: false, + })); 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 active_input_audio_streams_producers: InputAudioStreamsProducers = + Arc::new(Mutex::new(HashMap::new())); + let active_input_audio_streams_consumers: InputAudioStreamsConsumers = + Arc::new(Mutex::new(HashMap::new())); + let active_input_sudio_streams_info: InputAudioStreamsInfo = + Arc::new(RwLock::new(HashMap::new())); + + let active_input_audio_streams_producers_tcp = active_input_audio_streams_producers.clone(); + let active_input_audio_streams_consumers_tcp = active_input_audio_streams_consumers.clone(); + let active_input_sudio_streams_info_tcp = active_input_sudio_streams_info.clone(); + + let server_address = { + let info = client_info.read().unwrap(); + info.server_address + }; let mut tcp_stream = TcpStream::connect(server_address) .await .expect("failed to connect using tcp"); // tcp_stream.readable().await.expect("idk"); - - let mut session_client_id: u32 = 0; - let mut session_microphone_stream_id: u8 = 0; + // + let client_info_tcp = client_info.clone(); let mut global_event_rx = global_event_tx.subscribe(); + let global_event_tx_clone = global_event_tx.clone(); + + // tcp connection 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(n) => { + let ssss = String::from_utf8(tcp_buf[0..n].to_vec()).unwrap(); + println!("json data: {}", ssss); + + 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; + let mut client_info_var = client_info_tcp.write().unwrap(); + client_info_var.client_id = client_id; + client_info_var.connected = true; println!("fetched client id: {}", client_id); } TcpResponseData::UserConnected { user_id } => { @@ -66,12 +109,48 @@ async fn main() { TcpResponseData::UserDisconnected { user_id } => { println!("user disconnected: {}", user_id); } + TcpResponseData::AudioStreamsFullUpdate { audio_stream_info } => { + let mut producer_map = active_input_audio_streams_producers_tcp.lock().unwrap(); + producer_map.clear(); + let mut consumer_map = active_input_audio_streams_consumers_tcp.lock().unwrap(); + consumer_map.clear(); + + let mut info_map = active_input_sudio_streams_info_tcp.write().unwrap(); + info_map.clear(); + + + for audio_stream in audio_stream_info { + let ring_buffer_ms = 150.0; + let ring_buffer_frames = (ring_buffer_ms / 1000.0) * audio_stream.sample_rate as f32; + let ring_buffer_samples = ring_buffer_frames as usize * audio_stream.channels as usize; + let ring_buffer_size = ring_buffer_samples; + let ring_buffer = HeapRb::::new(ring_buffer_size); + let (producer, consumer) = ring_buffer.split(); + + producer_map.insert( + (audio_stream.client_id, audio_stream.stream_id), + producer, + ); + consumer_map.insert( + (audio_stream.client_id, audio_stream.stream_id), + consumer, + ); + info_map.insert( + (audio_stream.client_id, audio_stream.stream_id), + audio_stream, + ); + } + println!("audio streams update"); + } + TcpResponseData::NewAudioStreamAccepted { audio_stream_id } => { + let _ = global_event_tx_clone.send(ClientEvent::NewAudioStreamAccepted(audio_stream_id)); + } _ => { println!("wrong response") } } } - Err(e) => eprintln!("failed to parse tcp packet: {}", e), + Err(e) => eprintln!("failed to parse tcp packet: {}", e),} }, Err(e) => eprintln!("failed to receive tcp packet: {}", e), } @@ -93,9 +172,24 @@ async fn main() { .await .expect("failed to send initial message"); } - _ => { - println!("aaaa") + ClientEvent::NewAudioStreamRequest(audio_stream_info) => { + println!("new audio stream request"); + let session_client_id = client_info_tcp.read().unwrap().client_id; + let _ = tcp_stream + .write_all( + serde_json::to_string(&TcpRequest { + client_id: session_client_id, + data: shared::TcpRequestData::NewAudioStream { + audio_stream_info + }, + }) + .expect("failed to parse initial request") + .as_bytes(), + ) + .await + .expect("failed to send initial message"); } + _ => {} } } Err(e) => eprintln!("error: {}", e), @@ -131,68 +225,32 @@ async fn main() { println!("stream config: {:?}", config); - let ring_buffer_ms = 150.0; - let ring_buffer_frames = (ring_buffer_ms / 1000.0) * config.sample_rate as f32; - let ring_buffer_samples = ring_buffer_frames as usize * config.channels as usize; - - let ring_buffer_size = ring_buffer_samples; - - let microphone_ring_buffer = HeapRb::::new(ring_buffer_size); - let (mut microphone_producer, mut microphone_consumer) = microphone_ring_buffer.split(); - - let speaker_ring_buffer = HeapRb::::new(ring_buffer_size); - let (mut speaker_producer, mut speaker_consumer) = speaker_ring_buffer.split(); - - let input_data_fn = move |data: &[f32], _: &cpal::InputCallbackInfo| { - for &sample in data { - if microphone_producer.try_push(sample).is_err() {} - } - }; + let active_input_audio_streams_consumer_cpal = active_input_audio_streams_consumers.clone(); let output_data_fn = move |data: &mut [f32], _: &cpal::OutputCallbackInfo| { + let mut streams = active_input_audio_streams_consumer_cpal.lock().unwrap(); for sample in data { - *sample = match speaker_consumer.try_pop() { - Some(s) => s, - None => 0.0, - }; + let mut mixed_value: f32 = 0.0; + + for (_key, consumer) in streams.iter_mut() { + mixed_value += match consumer.try_pop() { + Some(s) => s, + None => 0.0, + } + } + + *sample = mixed_value; } }; let err_fn = |err| eprintln!("stream error: {}", err); - - let input_stream = input_device - .build_input_stream(config, input_data_fn, err_fn, None) - .expect("building input stream failed"); let output_stream = output_device .build_output_stream(config, output_data_fn, err_fn, None) .expect("building output stream failed"); - input_stream.play().expect("failed to play input stream"); output_stream.play().expect("failed to play output stream"); - tokio::spawn(async move { - let frames_needed = ((20.0 / 1000.0) * config.sample_rate as f32) as usize; - let mut frame_buffer = vec![0.0; frames_needed]; - - loop { - if microphone_consumer.occupied_len() >= frames_needed { - microphone_consumer.pop_slice(&mut frame_buffer); - - 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()); - } - - let _ = output_socket.send_to(&payload, server_address).await; - } else { - tokio::time::sleep(std::time::Duration::from_millis(2)).await; - } - } - }); + let active_input_audio_streams_producers_udp = active_input_audio_streams_producers.clone(); tokio::spawn(async move { let mut receive_buffer = vec![0u8; 4096]; @@ -220,13 +278,137 @@ async fn main() { decoded_samples.push(f32::from_le_bytes(bytes)); } - speaker_producer.push_slice(&decoded_samples); + let mut active_streams_map = + active_input_audio_streams_producers_udp.lock().unwrap(); + + if let Some(producer) = active_streams_map.get_mut(&(client_id, stream_id)) { + producer.push_slice(&decoded_samples); + } + + // speaker_producer.push_slice(&decoded_samples); } Err(e) => eprintln!("failed to receive udp packet: {}", e), } } }); + tokio::time::sleep(std::time::Duration::from_secs(5)).await; + + let broadcast_transmitter = global_event_tx.clone(); + let broadcast_receiver = global_event_tx.subscribe(); + + let client_info_audio_spawner = client_info.clone(); + + // let _ = audio_stream_spawner( + // input_device, + // broadcast_transmitter, + // broadcast_receiver, + // output_socket, + // client_info_audio_spawner, + // ) + // .await; + + tokio::spawn(audio_stream_spawner( + input_device, + broadcast_transmitter, + broadcast_receiver, + output_socket, + client_info_audio_spawner, + )); + tokio::signal::ctrl_c().await.expect("failed to close"); println!("shutting down"); } + +async fn audio_stream_spawner( + input_device: Device, + broadcast_transmitter: Sender, + mut broadcast_receiver: Receiver, + udp_socket: Arc, + clinet_info: Arc>, +) { + let config: cpal::StreamConfig = input_device + .default_input_config() + .expect("input stream config init failed") + .into(); + + let current_stream_id: u8 = rand::random(); + + loop { + let (connected, client_id) = { + let info = clinet_info.read().unwrap(); + (info.connected, info.client_id) + }; + if connected { + let _ = + broadcast_transmitter.send(ClientEvent::NewAudioStreamRequest(AudioStreamInfo { + client_id: client_id, + stream_id: current_stream_id, + channels: config.channels, + sample_rate: config.sample_rate, + })); + break; + } + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + } + + loop { + match broadcast_receiver.recv().await { + Ok(ClientEvent::NewAudioStreamAccepted(stream_id)) + if stream_id == current_stream_id => + { + println!("new stream accepted: {}", stream_id); + break; + } + Ok(_) => {} + Err(e) => eprintln!("error: {}", e), + } + } + + let ring_buffer_ms = 150.0; + let ring_buffer_frames = (ring_buffer_ms / 1000.0) * config.sample_rate as f32; + let ring_buffer_samples = ring_buffer_frames as usize * config.channels as usize; + let ring_buffer_size = ring_buffer_samples; + let ring_buffer = HeapRb::::new(ring_buffer_size); + let (mut producer, mut consumer) = ring_buffer.split(); + + let input_data_fn = move |data: &[f32], _: &cpal::InputCallbackInfo| { + for &sample in data { + if producer.try_push(sample).is_err() {} + } + }; + let err_fn = |err| eprintln!("stream error: {}", err); + + let input_stream = input_device + .build_input_stream(config, input_data_fn, err_fn, None) + .expect("building input stream failed"); + + input_stream.play().expect("failed to play input stream"); + + let frames_needed = ((20.0 / 1000.0) * config.sample_rate as f32) as usize; + let mut frame_buffer = vec![0.0; frames_needed]; + + loop { + if consumer.occupied_len() >= frames_needed { + consumer.pop_slice(&mut frame_buffer); + + let mut payload = Vec::with_capacity(4 + 1 + frames_needed * 4); + + let (server_address, client_id) = { + let info = clinet_info.read().unwrap(); + (info.server_address, info.client_id) + }; + + payload.extend(client_id.to_be_bytes()); + payload.extend(current_stream_id.to_be_bytes()); + + for sample in &frame_buffer { + payload.extend_from_slice(&sample.to_le_bytes()); + } + + let _ = udp_socket.send_to(&payload, server_address).await; + } else { + tokio::time::sleep(std::time::Duration::from_millis(2)).await; + } + } +} diff --git a/server/src/main.rs b/server/src/main.rs index cca00a7..6a57636 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -12,16 +12,28 @@ struct Client { audio_streams: Option>, } -#[derive(Clone, Debug)] +#[derive(Clone)] enum ServerEvent { - NewStream(u32), + NewAudioStream(AudioStreamInfo), + SendAllStreams, + SendAllOtherStreams(u32), UserConnected(u32), UserDisconnected(u32), - ClosedStream(u32), } type State = Arc>>; +fn scrape_all_client_streams_info(clients: &HashMap) -> Vec { + let mut all_audio_streams_info: Vec = Vec::new(); + for (_, data) in clients { + if let Some(audio_streams_info) = &data.audio_streams { + all_audio_streams_info.extend(audio_streams_info.iter().cloned()); + } + } + + return all_audio_streams_info; +} + #[tokio::main] async fn main() { let listen_address = "0.0.0.0:8080"; @@ -101,7 +113,7 @@ async fn main() { 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( @@ -115,9 +127,47 @@ async fn main() { .as_bytes(), ) .await; + + let _ = global_event_tx_clone.send(ServerEvent::UserConnected(current_client_id)); } TcpRequestData::NewAudioStream { audio_stream_info } => { + if let Some(client) = tcp_state.write().await.get_mut(¤t_client_id) { + if let Some(audio_streams) = &mut client.audio_streams { + audio_streams.push(audio_stream_info); + } else { + client.audio_streams = Some(vec![audio_stream_info]); + } + } + let _ = tcp_stream + .write_all( + serde_json::to_string(&TcpResponse { + status: ResponseStatus::Ok, + data: shared::TcpResponseData::NewAudioStreamAccepted { + audio_stream_id: audio_stream_info.stream_id, + }, + }) + .expect("failed to parse response json TcpRequestData::NewConnection") + .as_bytes(), + ) + .await; + let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id)); + } + TcpRequestData::DropAudioStream { audio_stream_id } => { + if let Some(client) = tcp_state.write().await.get_mut(¤t_client_id) { + if let Some(audio_streams) = &mut client.audio_streams { + let mut maybe_idx: Option = None; + for (i, audio_stream) in audio_streams.iter_mut().enumerate() { + if audio_stream.stream_id == audio_stream_id { + maybe_idx = Some(i); + } + } + if let Some(idx) = maybe_idx { + audio_streams.remove(idx); + } + } + } + let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id)); } _ => eprintln!("i dont support other requests yet"), }, @@ -131,29 +181,71 @@ async fn main() { 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; + if new_user_id != current_client_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; + + let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id)); + } } ServerEvent::UserDisconnected(user_id) => { + if user_id != current_client_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; + + let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id)); + } + } + ServerEvent::SendAllOtherStreams(id) => { + if current_client_id != id { + let clients = tcp_state.read().await; + + let _ = tcp_stream + .write_all( + serde_json::to_string(&TcpResponse { + status: ResponseStatus::Ok, + data: shared::TcpResponseData::AudioStreamsFullUpdate { + audio_stream_info: scrape_all_client_streams_info(&clients), + }, + }) + .expect("failed to parse response json TcpRequestData::NewConnection") + .as_bytes(), + ) + .await; + } + } + ServerEvent::SendAllStreams => { + let clients = tcp_state.read().await; + let _ = tcp_stream .write_all( serde_json::to_string(&TcpResponse { status: ResponseStatus::Ok, - data: shared::TcpResponseData::UserDisconnected { - user_id: user_id + data: shared::TcpResponseData::AudioStreamsFullUpdate { + audio_stream_info: scrape_all_client_streams_info(&clients), }, }) - .expect("failed to parse response json ServerEvent::UserConnected") + .expect("failed to parse response json TcpRequestData::NewConnection") .as_bytes(), ) .await; diff --git a/shared/src/lib.rs b/shared/src/lib.rs index 5eb9b33..82319ad 100644 --- a/shared/src/lib.rs +++ b/shared/src/lib.rs @@ -6,10 +6,11 @@ pub enum ResponseStatus { Err, } -#[derive(Serialize, Deserialize)] +#[derive(Serialize, Deserialize, Clone, Copy)] pub struct AudioStreamInfo { + pub client_id: u32, pub stream_id: u8, - pub channels: u8, + pub channels: u16, pub sample_rate: u32, } @@ -45,10 +46,10 @@ pub enum TcpResponseData { UserDisconnected { user_id: u32, }, - NewAudioStream { + NewAudioStreamAccepted { audio_stream_id: u8, }, - AudioStreamsUpdate { + AudioStreamsFullUpdate { audio_stream_info: Vec, }, }