sending stream info

Cargo.lock

@@ -79,6 +79,7 @@ name = "client"
 version = "0.1.0"
 dependencies = [
  "cpal",
+ "rand",
  "ringbuf",
  "serde",
  "serde_json",

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"

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 shared::{AudioStreamInfo, ResponseStatus, TcpRequest, TcpResponse, TcpResponseData};
-use std::{collections::HashMap, sync::Arc};
+use std::{
-use tokio::sync::{Mutex, broadcast};
+    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<Mutex<HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, true, false>>>>;
+type InputAudioStreamsConsumers =
     Arc<Mutex<HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, false, true>>>>;
 
+type InputAudioStreamsInfo = Arc<RwLock<HashMap<AudioStreamKey, AudioStreamInfo>>>;
+
+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<RwLock<ClientInfo>> = Arc::new(RwLock::new(ClientInfo {
+        client_id: 0,
+        server_address: "127.0.0.1:8080",
+        connected: false,
+    }));
 
     let (global_event_tx, _) = broadcast::channel::<ClientEvent>(16);
 
-    let active_streams: AudioMixerStreams = Arc::new(Mutex::new(HashMap::new()));
+    let active_input_audio_streams_producers: InputAudioStreamsProducers =
-    let active_streams_cpal = active_streams.clone();
+        Arc::new(Mutex::new(HashMap::new()));
-    let active_streams_socket = active_streams.clone();
+    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 client_info_tcp = client_info.clone();
-    let mut session_microphone_stream_id: u8 = 0;
 
     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::<TcpResponse>(&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::<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;
+                                        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::<f32>::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");
                                 }
-                                _ => {
+                                ClientEvent::NewAudioStreamRequest(audio_stream_info) => {
-                                    println!("aaaa")
+                                    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 active_input_audio_streams_consumer_cpal = active_input_audio_streams_consumers.clone();
-    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::<f32>::new(ring_buffer_size);
-    let (mut microphone_producer, mut microphone_consumer) = microphone_ring_buffer.split();
-
-    let speaker_ring_buffer = HeapRb::<f32>::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 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() {
+            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 active_input_audio_streams_producers_udp = active_input_audio_streams_producers.clone();
-        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;
-            }
-        }
-    });
 
     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<ClientEvent>,
+    mut broadcast_receiver: Receiver<ClientEvent>,
+    udp_socket: Arc<UdpSocket>,
+    clinet_info: Arc<RwLock<ClientInfo>>,
+) {
+    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::<f32>::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;
+        }
+    }
+}

server/src/main.rs

@@ -12,16 +12,28 @@ struct Client {
     audio_streams: Option<Vec<AudioStreamInfo>>,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 enum ServerEvent {
-    NewStream(u32),
+    NewAudioStream(AudioStreamInfo),
+    SendAllStreams,
+    SendAllOtherStreams(u32),
     UserConnected(u32),
     UserDisconnected(u32),
-    ClosedStream(u32),
 }
 
 type State = Arc<RwLock<HashMap<u32, Client>>>;
 
+fn scrape_all_client_streams_info(clients: &HashMap<u32, Client>) -> Vec<AudioStreamInfo> {
+    let mut all_audio_streams_info: Vec<AudioStreamInfo> = 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(&current_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(&current_client_id) {
+                                            if let Some(audio_streams) = &mut client.audio_streams {
+                                                let mut maybe_idx: Option<usize> = 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,6 +181,7 @@ async fn main() {
                             Ok(event) => {
                                 match event {
                                     ServerEvent::UserConnected(new_user_id) => {
+                                        if new_user_id != current_client_id {
                                             let _ = tcp_stream
                                                 .write_all(
                                                     serde_json::to_string(&TcpResponse {
@@ -143,8 +194,12 @@ async fn main() {
                                                     .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 {
@@ -157,6 +212,43 @@ async fn main() {
                                                     .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::AudioStreamsFullUpdate {
+                                                        audio_stream_info: scrape_all_client_streams_info(&clients),
+                                                    },
+                                                })
+                                                .expect("failed to parse response json TcpRequestData::NewConnection")
+                                                .as_bytes(),
+                                            )
+                                            .await;
                                     }
                                     _ => {
                                         println!("lkdahjsf");

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<AudioStreamInfo>,
     },
 }