support for most sample formats

.gitignore

@@ -1 +1,2 @@
 /target
+.DS_Store

client/Cargo.toml

@@ -13,3 +13,8 @@ tokio = { version = "1.53.1", features = ["full"] }
 rand = "0.10.2"
 clap = { version = "4.6.6", features = ["derive"] }
 ropus = "0.12.18"
+ratatui = "0.30.2"
+crossterm = "0.29.0"
+gag = "1.0.0"
+tokio-util = { version = "0.7.19", features = ["codec"]}
+futures = "0.3.34"

client/src/audio_streams.rs

@@ -0,0 +1,367 @@
+use cpal::{
+    Device, FromSample, Sample, SampleFormat, SizedSample, StreamConfig, SupportedStreamConfig,
+    traits::{DeviceTrait, StreamTrait},
+};
+use ringbuf::{
+    HeapRb, SharedRb,
+    storage::Heap,
+    traits::{Consumer, Observer, Producer, Split},
+    wrap::caching::Caching,
+};
+use ropus::Encoder;
+use shared::AudioStreamInfo;
+use std::sync::{Arc, RwLock};
+use tokio::{
+    net::UdpSocket,
+    sync::broadcast::{Receiver, Sender},
+};
+
+use crate::models::{
+    ClientEvent, ClientInfo, InputAudioStreamsConsumers, InputAudioStreamsEncoders,
+    InputAudioStreamsInfo, InputAudioStreamsProducers,
+};
+
+fn cpal_build_output_stream<T>(
+    output_device: &Device,
+    output_device_config: StreamConfig,
+    active_input_audio_streams_consumers: InputAudioStreamsConsumers,
+) -> cpal::Stream
+where
+    T: SizedSample,
+    T: FromSample<f32>,
+{
+    let output_data_fn = move |data: &mut [T], _: &cpal::OutputCallbackInfo| {
+        let mut streams = active_input_audio_streams_consumers.lock().unwrap();
+        for sample in data {
+            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.to_sample::<T>();
+        }
+    };
+    let err_fn = |err| eprintln!("stream error: {}", err);
+
+    return output_device
+        .build_output_stream(output_device_config, output_data_fn, err_fn, None)
+        .expect("building output stream failed");
+}
+
+pub async fn audio_stream_receiver(
+    output_device: Device,
+    output_device_config: SupportedStreamConfig,
+    active_input_audio_streams_producers: InputAudioStreamsProducers,
+    active_input_audio_streams_consumers: InputAudioStreamsConsumers,
+    active_input_audio_streams_encoders: InputAudioStreamsEncoders,
+    active_input_audio_streams_info: InputAudioStreamsInfo,
+    udp_socket: Arc<UdpSocket>,
+) {
+    let output_stream = match output_device_config.sample_format() {
+        SampleFormat::F64 => cpal_build_output_stream::<f64>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers,
+        ),
+        SampleFormat::I64 => cpal_build_output_stream::<i64>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers,
+        ),
+        SampleFormat::U64 => cpal_build_output_stream::<u64>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers,
+        ),
+        SampleFormat::F32 => cpal_build_output_stream::<f32>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers,
+        ),
+        SampleFormat::I32 => cpal_build_output_stream::<i32>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers,
+        ),
+        SampleFormat::U32 => cpal_build_output_stream::<u32>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers,
+        ),
+        SampleFormat::I16 => cpal_build_output_stream::<i16>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers,
+        ),
+        SampleFormat::U16 => cpal_build_output_stream::<u16>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers,
+        ),
+        SampleFormat::I8 => cpal_build_output_stream::<i8>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers,
+        ),
+        SampleFormat::U8 => cpal_build_output_stream::<u8>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers,
+        ),
+        _ => {
+            eprintln!("unsupported output stream format :(");
+            return;
+        }
+    };
+
+    output_stream.play().expect("failed to play output stream");
+
+    let mut receive_buffer = vec![0u8; 4096];
+
+    let mut decoded_samples = [0.0f32; 1024];
+    let mut decoded_samples_converted = [0.0f32; 2048];
+
+    loop {
+        match udp_socket.recv_from(&mut receive_buffer).await {
+            Ok((len, _)) => {
+                if len < 4 {
+                    continue;
+                }
+
+                let client_id_bytes: [u8; 4] = receive_buffer[0..4].try_into().unwrap();
+                let client_id = u32::from_be_bytes(client_id_bytes);
+
+                let stream_id_bytes = receive_buffer[4];
+                let stream_id = u8::from_be(stream_id_bytes);
+
+                let audio_packet = &receive_buffer[5..len];
+
+                let stream_info = {
+                    if let Some(&info) = active_input_audio_streams_info
+                        .read()
+                        .unwrap()
+                        .get(&(client_id, stream_id))
+                    {
+                        info
+                    } else {
+                        continue;
+                    }
+                };
+
+                let decoders_map = active_input_audio_streams_encoders.read().unwrap();
+
+                let decoder_mutex = decoders_map
+                    .get(&(client_id, stream_id))
+                    .expect("failed to get encoder mutex");
+
+                let mut decoder = decoder_mutex.lock().expect("failed to get encoder");
+
+                let decoded_len = decoder
+                    .decode_float(
+                        audio_packet,
+                        &mut decoded_samples,
+                        ropus::DecodeMode::Normal,
+                    )
+                    .expect("failed to decode opus");
+
+                let mut active_streams_map = active_input_audio_streams_producers.lock().unwrap();
+
+                if let Some(producer) = active_streams_map.get_mut(&(client_id, stream_id)) {
+                    if stream_info.channels == output_device_config.channels() {
+                        producer.push_slice(
+                            &decoded_samples[..decoded_len * stream_info.channels as usize],
+                        );
+                    } else if stream_info.channels > output_device_config.channels() {
+                        for (i, frame) in decoded_samples
+                            [..decoded_len * stream_info.channels as usize]
+                            .chunks_exact(2)
+                            .enumerate()
+                        {
+                            decoded_samples_converted[i] = (frame[0] + frame[1]) / 2.0;
+                        }
+
+                        producer.push_slice(
+                            &decoded_samples_converted
+                                [..decoded_len * stream_info.channels as usize],
+                        );
+                    } else {
+                        for (i, frame) in decoded_samples_converted
+                            [..decoded_len * output_device_config.channels() as usize]
+                            .chunks_exact_mut(2)
+                            .enumerate()
+                        {
+                            frame[0] = decoded_samples[i];
+                            frame[1] = decoded_samples[i];
+                        }
+
+                        producer.push_slice(
+                            &decoded_samples_converted
+                                [..decoded_len * output_device_config.channels() as usize],
+                        );
+                    }
+                }
+            }
+            Err(e) => eprintln!("failed to receive udp packet: {}", e),
+        }
+    }
+}
+
+fn cpal_build_input_stream<T>(
+    input_device: &Device,
+    input_device_config: StreamConfig,
+    mut producer: Caching<Arc<SharedRb<Heap<f32>>>, true, false>,
+) -> cpal::Stream
+where
+    T: SizedSample,
+    f32: FromSample<T>,
+{
+    let input_data_fn = move |data: &[T], _: &cpal::InputCallbackInfo| {
+        for &sample in data {
+            let f32_sample: f32 = sample.to_sample::<f32>();
+
+            if producer.try_push(f32_sample).is_err() {}
+        }
+    };
+    let err_fn = |err| eprintln!("stream error: {}", err);
+
+    return input_device
+        .build_input_stream(input_device_config, input_data_fn, err_fn, None)
+        .expect("building input stream failed");
+}
+
+pub async fn audio_stream_spawner(
+    input_device: Device,
+    input_device_config: SupportedStreamConfig,
+    global_event_tx: Sender<ClientEvent>,
+    mut global_event_rx: Receiver<ClientEvent>,
+    udp_socket: Arc<UdpSocket>,
+    client_info: Arc<RwLock<ClientInfo>>,
+) {
+    let current_stream_id: u8 = rand::random();
+
+    loop {
+        let (connected, client_id) = {
+            let info = client_info.read().unwrap();
+            (info.connected, info.client_id)
+        };
+        if connected {
+            let _ = global_event_tx.send(ClientEvent::NewAudioStreamRequest(AudioStreamInfo {
+                client_id: client_id,
+                stream_id: current_stream_id,
+                channels: input_device_config.channels(),
+                sample_rate: input_device_config.sample_rate(),
+            }));
+            break;
+        }
+    }
+
+    loop {
+        match global_event_rx.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 = 80.0;
+    let ring_buffer_frames = (ring_buffer_ms / 1000.0) * input_device_config.sample_rate() as f32;
+    let ring_buffer_samples = ring_buffer_frames as usize * input_device_config.channels() as usize;
+    let ring_buffer_size = ring_buffer_samples;
+    let ring_buffer = HeapRb::<f32>::new(ring_buffer_size);
+    let (producer, mut consumer) = ring_buffer.split();
+
+    let input_stream = match input_device_config.sample_format() {
+        SampleFormat::F64 => {
+            cpal_build_input_stream::<f64>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::I64 => {
+            cpal_build_input_stream::<i64>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::U64 => {
+            cpal_build_input_stream::<u64>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::F32 => {
+            cpal_build_input_stream::<f32>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::I32 => {
+            cpal_build_input_stream::<i32>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::U32 => {
+            cpal_build_input_stream::<u32>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::I16 => {
+            cpal_build_input_stream::<i16>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::U16 => {
+            cpal_build_input_stream::<u16>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::I8 => {
+            cpal_build_input_stream::<i8>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::U8 => {
+            cpal_build_input_stream::<u8>(&input_device, input_device_config.into(), producer)
+        }
+        _ => {
+            eprintln!("unsupported input stream format :(");
+            return;
+        }
+    };
+
+    input_stream.play().expect("failed to play input stream");
+
+    let frames_needed = ((20.0 / 1000.0) * input_device_config.sample_rate() as f32) as usize;
+    let mut frame_buffer = vec![0.0; frames_needed];
+
+    let channels: ropus::Channels = {
+        if input_device_config.channels() == 2 {
+            ropus::Channels::Stereo
+        } else {
+            ropus::Channels::Mono
+        }
+    };
+
+    let mut opus_encoder = Encoder::builder(
+        input_device_config.sample_rate(),
+        channels,
+        ropus::Application::Voip,
+    )
+    .build()
+    .unwrap();
+
+    let mut encoded_frame_buffer = [0u8; 1024];
+
+    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 = client_info.read().unwrap();
+                (info.server_address.clone(), info.client_id)
+            };
+
+            payload.extend(client_id.to_be_bytes());
+            payload.extend(current_stream_id.to_be_bytes());
+
+            let encoded_len = opus_encoder
+                .encode_float(&frame_buffer.as_slice(), &mut encoded_frame_buffer)
+                .expect("failed to encode frame");
+
+            payload.extend_from_slice(&encoded_frame_buffer[..encoded_len]);
+
+            let _ = udp_socket.send_to(&payload, server_address).await;
+        } else {
+            tokio::time::sleep(std::time::Duration::from_millis(2)).await;
+        }
+    }
+}

client/src/main.rs

@@ -1,65 +1,30 @@
+mod audio_streams;
+mod models;
+mod tcp_connection;
+mod tui;
+
 use clap::Parser;
-use cpal::{
-    Device,
-    traits::{DeviceTrait, HostTrait, StreamTrait},
-};
-use ringbuf::{
-    HeapRb,
-    traits::{Consumer, Observer, Producer, Split},
-    wrap::caching::Caching,
-};
-use ropus::{Decoder, Encoder};
-use shared::{AudioStreamInfo, ResponseStatus, TcpRequest, TcpResponse, TcpResponseData};
+use cpal::traits::{DeviceTrait, HostTrait};
 use std::{
     collections::HashMap,
     sync::{Arc, Mutex, RwLock},
 };
-use tokio::sync::broadcast::{self, Receiver, Sender};
-use tokio::{
-    io::{AsyncReadExt, AsyncWriteExt},
-    net::{TcpStream, UdpSocket},
+use tokio::net::UdpSocket;
+use tokio::sync::broadcast;
+
+use crate::{
+    audio_streams::{audio_stream_receiver, audio_stream_spawner},
+    models::{
+        AudioControl, CliArgs, ClientEvent, ClientInfo, InputAudioStreamsConsumers,
+        InputAudioStreamsEncoders, InputAudioStreamsInfo, InputAudioStreamsProducers,
+    },
+    tcp_connection::tcp_connection,
+    tui::Tui,
 };
 
-#[derive(Parser)]
-#[command(version, about)]
-struct Cli {
-    #[arg(short, long, required(true))]
-    server_ip: Option<String>,
-}
-
-#[derive(Clone)]
-enum ClientEvent {
-    NewConnection,
-    NewAudioStreamRequest(AudioStreamInfo),
-    NewAudioStreamAccepted(u8),
-}
-
-type AudioStreamKey = (u32, u8);
-
-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 InputAudioStreamsEncoders = Arc<RwLock<HashMap<AudioStreamKey, Mutex<Decoder>>>>;
-
-type InputAudioStreamsInfo = Arc<RwLock<HashMap<AudioStreamKey, AudioStreamInfo>>>;
-
-struct ClientInfo {
-    client_id: u32,
-    server_address: String,
-    connected: bool,
-}
-
 #[tokio::main]
 async fn main() {
-    let cli = Cli::parse();
-
-    let client_info: Arc<RwLock<ClientInfo>> = Arc::new(RwLock::new(ClientInfo {
-        client_id: 0,
-        server_address: cli.server_ip.unwrap(),
-        connected: false,
-    }));
+    let cli = CliArgs::parse();
 
     let (global_event_tx, _) = broadcast::channel::<ClientEvent>(16);
 
@@ -74,437 +39,97 @@ async fn main() {
     let active_input_audio_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_audio_streams_encoders_tcp = active_input_audio_streams_encoders.clone();
-    let active_input_audio_streams_info_tcp = active_input_audio_streams_info.clone();
+    let udp_socket = Arc::new(UdpSocket::bind("0.0.0.0:0").await.unwrap());
 
-    let server_address = {
-        let info = client_info.read().unwrap();
-        info.server_address.clone()
-    };
-
-    let mut tcp_stream = TcpStream::connect(server_address)
-        .await
-        .expect("failed to connect using tcp");
-    // tcp_stream.readable().await.expect("idk");
-    //
-    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) => {
-                            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 } => {
-                                        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 } => {
-                                        println!("new user connected: {}", user_id);
-                                    }
-                                    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 encoder_map = active_input_audio_streams_encoders_tcp.write().unwrap();
-                                        encoder_map.clear();
-
-                                        let mut info_map = active_input_audio_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,
-                                            );
-
-                                            let channels: ropus::Channels = {
-                                                if audio_stream.channels == 1 {
-                                                    ropus::Channels::Mono
-                                                } else {
-                                                    ropus::Channels::Stereo
-                                                }
-                                            };
-
-                                            let decoder = Decoder::new(audio_stream.sample_rate, channels).expect("failed to creat decoder");
-                                            encoder_map.insert(
-                                                (audio_stream.client_id, audio_stream.stream_id),
-                                                Mutex::new(decoder),
-                                            );
-
-                                            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 receive tcp packet: {}", e),
-                    }
-                }
-                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");
-                                }
-                                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),
-                    }
-                }
-            }
-        }
+    let audio_control: Arc<AudioControl> = Arc::new(AudioControl {
+        active_input_audio_streams_producers: active_input_audio_streams_producers.clone(),
+        active_input_audio_streams_consumers: active_input_audio_streams_consumers.clone(),
+        active_input_audio_streams_encoders: active_input_audio_streams_encoders.clone(),
+        active_input_audio_streams_info: active_input_audio_streams_info.clone(),
+        udp_socket: udp_socket.clone(),
     });
 
+    if let Some(server_ip) = cli.connect_with_defaults {
+        let client_info: Arc<RwLock<ClientInfo>> = Arc::new(RwLock::new(ClientInfo {
+            client_id: 0,
+            server_address: server_ip,
+            connected: false,
+        }));
+
+        tokio::spawn(tcp_connection(
+            client_info.clone(),
+            global_event_tx.clone(),
+            global_event_tx.subscribe(),
+            audio_control.clone(),
+        ));
+
         let _ = global_event_tx.send(ClientEvent::NewConnection);
 
-    let socket = Arc::new(UdpSocket::bind("0.0.0.0:0").await.unwrap());
-
-    let output_socket = socket.clone();
-    let input_socket = socket.clone();
-
         let host = cpal::default_host();
 
         let input_device = host
             .default_input_device()
             .expect("no input device available");
+        let input_device_config = input_device
+            .default_input_config()
+            .expect("failed to get default input config");
+
         let output_device = host
             .default_output_device()
             .expect("no output device available");
+        let output_device_config = output_device
+            .default_output_config()
+            .expect("failed to get default output config");
 
         println!("input device: {}", input_device);
         println!("output device: {}", output_device);
 
-    let speaker_config: cpal::StreamConfig = output_device
-        .default_output_config()
-        .expect("stream config init failed")
-        .into();
-
-    println!("speaker config: {:?}", speaker_config);
-
-    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 {
-            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 output_stream = output_device
-        .build_output_stream(speaker_config, output_data_fn, err_fn, None)
-        .expect("building output stream failed");
-
-    output_stream.play().expect("failed to play output stream");
-
-    let active_input_audio_streams_producers_udp = active_input_audio_streams_producers.clone();
-    let active_input_audio_streams_info_udp = active_input_audio_streams_info.clone();
-    let speaker_config_clone = speaker_config.clone();
-    let active_input_audio_streams_encoders_clone = active_input_audio_streams_encoders.clone();
-
-    tokio::spawn(async move {
-        let mut receive_buffer = vec![0u8; 4096];
-
-        let mut decoded_samples = [0.0f32; 1024];
-        let mut decoded_samples_converted = [0.0f32; 2048];
-
-        loop {
-            match input_socket.recv_from(&mut receive_buffer).await {
-                Ok((len, _)) => {
-                    if len < 4 {
-                        continue;
-                    }
-
-                    let client_id_bytes: [u8; 4] = receive_buffer[0..4].try_into().unwrap();
-                    let client_id = u32::from_be_bytes(client_id_bytes);
-
-                    let stream_id_bytes = receive_buffer[4];
-                    let stream_id = u8::from_be(stream_id_bytes);
-
-                    let audio_packet = &receive_buffer[5..len];
-
-                    let stream_info = {
-                        if let Some(&info) = active_input_audio_streams_info_udp
-                            .read()
-                            .unwrap()
-                            .get(&(client_id, stream_id))
-                        {
-                            info
-                        } else {
-                            continue;
-                        }
-                    };
-
-                    let decoders_map = active_input_audio_streams_encoders_clone.read().unwrap();
-
-                    let decoder_mutex = decoders_map
-                        .get(&(client_id, stream_id))
-                        .expect("failed to get encoder mutex");
-
-                    let mut decoder = decoder_mutex.lock().expect("failed to get encoder");
-
-                    let decoded_len = decoder
-                        .decode_float(
-                            audio_packet,
-                            &mut decoded_samples,
-                            ropus::DecodeMode::Normal,
-                        )
-                        .expect("failed to decode opus");
-
-                    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)) {
-                        if stream_info.channels == speaker_config_clone.channels {
-                            producer.push_slice(
-                                &decoded_samples[..decoded_len * stream_info.channels as usize],
-                            );
-                        } else if stream_info.channels > speaker_config_clone.channels {
-                            for (i, frame) in decoded_samples
-                                [..decoded_len * stream_info.channels as usize]
-                                .chunks_exact(2)
-                                .enumerate()
-                            {
-                                decoded_samples_converted[i] = (frame[0] + frame[1]) / 2.0;
-                            }
-
-                            producer.push_slice(
-                                &decoded_samples_converted
-                                    [..decoded_len * stream_info.channels as usize],
-                            );
-                        } else {
-                            for (i, frame) in decoded_samples_converted
-                                [..decoded_len * speaker_config_clone.channels as usize]
-                                .chunks_exact_mut(2)
-                                .enumerate()
-                            {
-                                frame[0] = decoded_samples[i];
-                                frame[1] = decoded_samples[i];
-                            }
-
-                            producer.push_slice(
-                                &decoded_samples_converted
-                                    [..decoded_len * speaker_config_clone.channels as usize],
-                            );
-                        }
-                    }
-                }
-                Err(e) => eprintln!("failed to receive udp packet: {}", e),
-            }
-        }
-    });
-
-    let broadcast_transmitter = global_event_tx.clone();
-    let broadcast_receiver = global_event_tx.subscribe();
-
-    let client_info_audio_spawner = client_info.clone();
+        tokio::spawn(audio_stream_receiver(
+            output_device,
+            output_device_config.into(),
+            active_input_audio_streams_producers.clone(),
+            active_input_audio_streams_consumers.clone(),
+            active_input_audio_streams_encoders.clone(),
+            active_input_audio_streams_info.clone(),
+            udp_socket.clone(),
+        ));
 
         tokio::spawn(audio_stream_spawner(
             input_device,
-        broadcast_transmitter,
-        broadcast_receiver,
-        output_socket,
-        client_info_audio_spawner,
+            input_device_config.into(),
+            global_event_tx.clone(),
+            global_event_tx.subscribe(),
+            udp_socket.clone(),
+            client_info.clone(),
         ));
 
         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,
+    if let Some(server_ip) = cli.server_ip {
+        let client_info: Arc<RwLock<ClientInfo>> = Arc::new(RwLock::new(ClientInfo {
+            client_id: 0,
+            server_address: server_ip,
+            connected: false,
         }));
-            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),
-        }
-    }
+        tokio::spawn(tcp_connection(
+            client_info.clone(),
+            global_event_tx.clone(),
+            global_event_tx.subscribe(),
+            audio_control.clone(),
+        ));
 
-    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 _ = global_event_tx.send(ClientEvent::NewConnection);
 
-    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];
-
-    let channels: ropus::Channels = {
-        if config.channels == 1 {
-            ropus::Channels::Mono
-        } else {
-            ropus::Channels::Stereo
-        }
-    };
-
-    let mut opus_encoder = Encoder::builder(config.sample_rate, channels, ropus::Application::Voip)
-        .build()
-        .unwrap();
-
-    let mut encoded_frame_buffer = [0u8; 1024];
-
-    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.clone(), info.client_id)
-            };
-
-            payload.extend(client_id.to_be_bytes());
-            payload.extend(current_stream_id.to_be_bytes());
-
-            let encoded_len = opus_encoder
-                .encode_float(&frame_buffer.as_slice(), &mut encoded_frame_buffer)
-                .expect("failed to encode frame");
-
-            payload.extend_from_slice(&encoded_frame_buffer[..encoded_len]);
-
-            let _ = udp_socket.send_to(&payload, server_address).await;
-        } else {
-            tokio::time::sleep(std::time::Duration::from_millis(2)).await;
-        }
+        ratatui::run(|terminal| {
+            Tui::new().run(
+                terminal,
+                audio_control.clone(),
+                global_event_tx.clone(),
+                client_info.clone(),
+            )
+        });
     }
 }

client/src/models.rs

@@ -0,0 +1,58 @@
+use clap::Parser;
+use ringbuf::{HeapRb, wrap::caching::Caching};
+use ropus::Decoder;
+use shared::AudioStreamInfo;
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex, RwLock},
+};
+use tokio::net::UdpSocket;
+
+#[derive(Parser)]
+#[command(version, about)]
+pub struct CliArgs {
+    #[arg(short, long, required(false))]
+    pub connect_with_defaults: Option<String>,
+
+    #[arg(short, long, required(false))]
+    pub server_ip: Option<String>,
+}
+
+#[derive(Debug, Clone, Copy)]
+pub enum TuiScreen {
+    DeviceSelect,
+    Blank,
+    Exit,
+}
+
+#[derive(Clone)]
+pub enum ClientEvent {
+    NewConnection,
+    NewAudioStreamRequest(AudioStreamInfo),
+    NewAudioStreamAccepted(u8),
+}
+
+pub struct ClientInfo {
+    pub client_id: u32,
+    pub server_address: String,
+    pub connected: bool,
+}
+
+type AudioStreamKey = (u32, u8);
+
+pub type InputAudioStreamsProducers =
+    Arc<Mutex<HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, true, false>>>>;
+pub type InputAudioStreamsConsumers =
+    Arc<Mutex<HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, false, true>>>>;
+
+pub type InputAudioStreamsEncoders = Arc<RwLock<HashMap<AudioStreamKey, Mutex<Decoder>>>>;
+
+pub type InputAudioStreamsInfo = Arc<RwLock<HashMap<AudioStreamKey, AudioStreamInfo>>>;
+
+pub struct AudioControl {
+    pub active_input_audio_streams_producers: InputAudioStreamsProducers,
+    pub active_input_audio_streams_consumers: InputAudioStreamsConsumers,
+    pub active_input_audio_streams_encoders: InputAudioStreamsEncoders,
+    pub active_input_audio_streams_info: InputAudioStreamsInfo,
+    pub udp_socket: Arc<UdpSocket>,
+}

client/src/tcp_connection.rs

@@ -0,0 +1,166 @@
+use futures::{SinkExt, StreamExt};
+use ringbuf::{HeapRb, traits::Split};
+use ropus::Decoder;
+use shared::{ResponseStatus, TcpRequest, TcpResponse, TcpResponseData};
+use std::sync::{Arc, Mutex, RwLock};
+use tokio::{
+    net::TcpStream,
+    sync::broadcast::{Receiver, Sender},
+};
+use tokio_util::{
+    bytes::Bytes,
+    codec::{Framed, LengthDelimitedCodec},
+};
+
+use crate::models::{AudioControl, ClientEvent, ClientInfo};
+
+pub async fn tcp_connection(
+    client_info: Arc<RwLock<ClientInfo>>,
+    global_event_tx: Sender<ClientEvent>,
+    mut global_event_rx: Receiver<ClientEvent>,
+    audio_control: Arc<AudioControl>,
+) {
+    let server_address = {
+        let info = client_info.read().unwrap();
+        info.server_address.clone()
+    };
+
+    let tcp_stream = TcpStream::connect(server_address)
+        .await
+        .expect("failed to connect using tcp");
+
+    let mut framed_tcp_stream = Framed::new(tcp_stream, LengthDelimitedCodec::new());
+
+    loop {
+        tokio::select! {
+            frame_result = framed_tcp_stream.next() => {
+                match frame_result {
+                    None => {
+                        println!("server shut down");
+                        break;
+                    }
+                    Some(Ok(bytes)) => {
+                        match serde_json::from_slice::<TcpResponse>(&bytes) {
+                            Ok(response) => {
+                                println!("connection Ok: {}", response.status == ResponseStatus::Ok);
+                                match response.data {
+                                    TcpResponseData::NewConnection { client_id } => {
+                                        let mut client_info_var = client_info.write().unwrap();
+                                        client_info_var.client_id = client_id;
+                                        client_info_var.connected = true;
+                                        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);
+                                    }
+                                    TcpResponseData::AudioStreamsFullUpdate { audio_stream_info } => {
+                                        let mut producer_map = audio_control.active_input_audio_streams_producers.lock().unwrap();
+                                        producer_map.clear();
+                                        let mut consumer_map = audio_control.active_input_audio_streams_consumers.lock().unwrap();
+                                        consumer_map.clear();
+
+                                        let mut encoder_map = audio_control.active_input_audio_streams_encoders.write().unwrap();
+                                        encoder_map.clear();
+
+                                        let mut info_map = audio_control.active_input_audio_streams_info.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,
+                                            );
+
+                                            let channels: ropus::Channels = {
+                                                if audio_stream.channels == 1 {
+                                                    ropus::Channels::Mono
+                                                } else {
+                                                    ropus::Channels::Stereo
+                                                }
+                                            };
+
+                                            let decoder = Decoder::new(audio_stream.sample_rate, channels).expect("failed to creat decoder");
+                                            encoder_map.insert(
+                                                (audio_stream.client_id, audio_stream.stream_id),
+                                                Mutex::new(decoder),
+                                            );
+
+                                            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.send(ClientEvent::NewAudioStreamAccepted(audio_stream_id));
+                                    }
+                                }
+                            }
+                            Err(e) => {
+                                eprintln!("parsing error: {}", e);
+                                break;
+                            }
+                        }
+                    }
+                    Some(Err(e)) => {
+                        eprintln!("socket read error: {}", e);
+                            break;
+                    }
+                }
+            }
+            event_result = global_event_rx.recv() => {
+                match event_result {
+                    Ok(event) => {
+                        match event {
+                            ClientEvent::NewConnection => {
+                                let request_json = serde_json::to_string(&TcpRequest {
+                                    client_id: 0,
+                                    data: shared::TcpRequestData::NewConnection,
+                                })
+                                .expect("failed to parse initial request");
+
+                                let _ = framed_tcp_stream.send(Bytes::from(request_json))
+                                .await
+                                .expect("failed to send initial message");
+
+                            }
+                            ClientEvent::NewAudioStreamRequest(audio_stream_info) => {
+                                println!("new audio stream request");
+                                let session_client_id = client_info.read().unwrap().client_id;
+
+                                let request_json = serde_json::to_string(&TcpRequest {
+                                    client_id: session_client_id,
+                                    data: shared::TcpRequestData::NewAudioStream {
+                                        audio_stream_info
+                                    }
+                                }).expect("failed to parse initial request");
+
+                                let _ = framed_tcp_stream.send(Bytes::from(request_json))
+                                .await
+                                .expect("failed to send initial message");
+                            }
+                            _ => {}
+                        }
+                    }
+                    Err(e) => eprintln!("error: {}", e),
+                }
+            }
+        }
+    }
+}

client/src/tui.rs

@@ -0,0 +1,422 @@
+use cpal::{
+    Device, SAMPLE_RATE_48K, SampleFormat, SupportedStreamConfig,
+    traits::{DeviceTrait, HostTrait},
+};
+use crossterm::event::{self, Event, KeyCode, KeyEvent, KeyEventKind};
+use gag::Gag;
+use ratatui::{
+    DefaultTerminal, Frame,
+    layout::{Constraint, Direction, Layout},
+    style::Style,
+    widgets::{Block, Paragraph, Row, Table, TableState},
+};
+use std::{
+    sync::{Arc, RwLock},
+    vec,
+};
+use tokio::sync::broadcast::Sender;
+
+use crate::{
+    audio_streams::{audio_stream_receiver, audio_stream_spawner},
+    models::{AudioControl, ClientEvent, ClientInfo, TuiScreen},
+};
+
+pub struct Tui {
+    current_screen: TuiScreen,
+}
+
+impl Tui {
+    pub fn new() -> Self {
+        Self {
+            current_screen: TuiScreen::DeviceSelect,
+        }
+    }
+
+    pub fn run(
+        &mut self,
+        terminal: &mut DefaultTerminal,
+        audio_control: Arc<AudioControl>,
+        global_event_tx: Sender<ClientEvent>,
+        client_info: Arc<RwLock<ClientInfo>>,
+    ) {
+        loop {
+            match self.current_screen {
+                TuiScreen::Exit => {
+                    break;
+                }
+                TuiScreen::Blank => self.current_screen = BlankScreen::new().run(terminal),
+                TuiScreen::DeviceSelect => {
+                    self.current_screen = DeviceSelectScreen::new(
+                        audio_control.clone(),
+                        global_event_tx.clone(),
+                        client_info.clone(),
+                    )
+                    .run(terminal);
+                }
+            }
+        }
+    }
+}
+
+struct BlankScreen {
+    exit: bool,
+}
+
+impl BlankScreen {
+    pub fn new() -> Self {
+        Self { exit: false }
+    }
+
+    pub fn run(&mut self, terminal: &mut DefaultTerminal) -> TuiScreen {
+        while !self.exit {
+            terminal
+                .draw(|frame| self.draw(frame))
+                .expect("failed to draw");
+            self.handle_events();
+        }
+
+        return TuiScreen::Exit;
+    }
+
+    fn draw(&mut self, frame: &mut Frame) {
+        frame.render_widget(Paragraph::new("q - quit"), frame.area());
+    }
+
+    fn handle_events(&mut self) {
+        match event::read().expect("failed to read event") {
+            Event::Key(key_event) if key_event.kind == KeyEventKind::Press => {
+                match key_event.code {
+                    KeyCode::Char('q') => self.exit = true,
+                    _ => {}
+                }
+            }
+            _ => {}
+        }
+    }
+}
+
+struct DeviceSelectScreen {
+    exit: bool,
+    exit_screen: TuiScreen,
+    input_device_state: TableState,
+    output_device_state: TableState,
+    input_devices: Vec<(Device, SupportedStreamConfig)>,
+    output_devices: Vec<(Device, SupportedStreamConfig)>,
+    selected_selection: u8,
+
+    audio_control: Arc<AudioControl>,
+    global_event_tx: Sender<ClientEvent>,
+    client_info: Arc<RwLock<ClientInfo>>,
+}
+
+impl DeviceSelectScreen {
+    pub fn new(
+        audio_control: Arc<AudioControl>,
+        global_event_tx: Sender<ClientEvent>,
+        client_info: Arc<RwLock<ClientInfo>>,
+    ) -> Self {
+        let _print_gag = Gag::stderr().ok();
+
+        let host = cpal::default_host();
+
+        let mut input_devices: Vec<(Device, SupportedStreamConfig)> = Vec::new();
+
+        if let Some(default_input_device) = host.default_input_device() {
+            if let Ok(default_config) = default_input_device.default_input_config() {
+                if default_config.sample_rate() == SAMPLE_RATE_48K
+                    && (default_config.channels() == 1)
+                {
+                    input_devices.push((default_input_device.clone(), default_config));
+                }
+            } else if let Ok(supported_configs) = default_input_device.supported_input_configs() {
+                for supported_config in supported_configs {
+                    if let Some(input_config) =
+                        supported_config.try_with_sample_rate(SAMPLE_RATE_48K)
+                    {
+                        if input_config.channels() == 1 {
+                            input_devices.push((default_input_device, input_config));
+                            break;
+                        }
+                    }
+                }
+            }
+        }
+
+        if let Ok(possible_input_devices) = host.input_devices() {
+            for possible_input_device in possible_input_devices {
+                if let Ok(default_config) = possible_input_device.default_input_config() {
+                    if default_config.sample_rate() == SAMPLE_RATE_48K
+                        && (default_config.channels() == 1)
+                    {
+                        input_devices.push((possible_input_device.clone(), default_config));
+                        continue;
+                    }
+                }
+                if let Ok(supported_configs) = possible_input_device.supported_input_configs() {
+                    for supported_config in supported_configs {
+                        if let Some(input_config) =
+                            supported_config.try_with_sample_rate(SAMPLE_RATE_48K)
+                        {
+                            if input_config.channels() == 1 {
+                                input_devices.push((possible_input_device, input_config));
+                                break;
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
+        let mut output_devices: Vec<(Device, SupportedStreamConfig)> = Vec::new();
+
+        if let Some(default_output_device) = host.default_output_device() {
+            if let Ok(default_config) = default_output_device.default_output_config() {
+                if default_config.sample_rate() == SAMPLE_RATE_48K
+                    && (default_config.channels() == 1 || default_config.channels() == 2)
+                {
+                    output_devices.push((default_output_device.clone(), default_config.into()));
+                }
+            } else if let Ok(supported_configs) = default_output_device.supported_output_configs() {
+                for supported_config in supported_configs {
+                    if let Some(output_config) =
+                        supported_config.try_with_sample_rate(SAMPLE_RATE_48K)
+                    {
+                        if output_config.channels() == 1 || output_config.channels() == 2 {
+                            output_devices.push((default_output_device, output_config.into()));
+                            break;
+                        }
+                    }
+                }
+            }
+        }
+
+        if let Ok(possible_output_devices) = host.output_devices() {
+            for possible_output_device in possible_output_devices {
+                if let Ok(default_config) = possible_output_device.default_output_config() {
+                    if default_config.sample_rate() == SAMPLE_RATE_48K
+                        && (default_config.channels() == 1 || default_config.channels() == 2)
+                    {
+                        output_devices
+                            .push((possible_output_device.clone(), default_config.into()));
+                        continue;
+                    }
+                }
+                if let Ok(supported_configs) = possible_output_device.supported_output_configs() {
+                    for supported_config in supported_configs {
+                        if let Some(output_config) =
+                            supported_config.try_with_sample_rate(SAMPLE_RATE_48K)
+                        {
+                            if output_config.channels() == 1 || output_config.channels() == 2 {
+                                output_devices.push((possible_output_device, output_config.into()));
+                                break;
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
+        let mut input_state = TableState::default();
+        input_state.select(Some(0));
+
+        let mut output_state = TableState::default();
+        output_state.select(Some(0));
+
+        Self {
+            exit: false,
+            exit_screen: TuiScreen::Exit,
+            input_device_state: input_state,
+            output_device_state: output_state,
+            input_devices: input_devices,
+            output_devices: output_devices,
+            selected_selection: 0,
+
+            audio_control: audio_control,
+            global_event_tx: global_event_tx,
+            client_info: client_info,
+        }
+    }
+
+    pub fn run(&mut self, terminal: &mut DefaultTerminal) -> TuiScreen {
+        while !self.exit {
+            terminal
+                .draw(|frame| self.draw(frame))
+                .expect("failed to draw");
+            self.handle_events();
+        }
+
+        return self.exit_screen;
+    }
+
+    fn draw(&mut self, frame: &mut Frame) {
+        let outer_layout = Layout::default()
+            .direction(Direction::Horizontal)
+            .constraints(vec![Constraint::Percentage(50), Constraint::Percentage(50)])
+            .split(frame.area());
+
+        let block = Block::new();
+        let left_block = block.inner(outer_layout[0]);
+        let right_block = block.inner(outer_layout[1]);
+
+        let widths = [Constraint::Percentage(100), Constraint::Min(8)];
+
+        let mut input_device_rows: Vec<Row> = Vec::new();
+        for (device, config) in self.input_devices.iter() {
+            input_device_rows.push(Row::new(vec![
+                device.description().unwrap().name().to_string(),
+                config.sample_format().to_string(),
+            ]));
+        }
+        let input_device_table = Table::new(input_device_rows, widths)
+            .header(
+                Row::new(vec!["device name", "channels"])
+                    .style(Style::new().bold())
+                    .bottom_margin(1),
+            )
+            .block(
+                Block::bordered()
+                    .border_style({
+                        if self.selected_selection == 0 {
+                            Style::new().bold()
+                        } else {
+                            Style::default()
+                        }
+                    })
+                    .border_type({
+                        if self.selected_selection == 0 {
+                            ratatui::widgets::BorderType::Thick
+                        } else {
+                            ratatui::widgets::BorderType::Plain
+                        }
+                    })
+                    .title("select input device"),
+            )
+            .row_highlight_style(Style::new().reversed());
+
+        frame.render_stateful_widget(input_device_table, left_block, &mut self.input_device_state);
+
+        let mut output_device_rows: Vec<Row> = Vec::new();
+        for (device, config) in self.output_devices.iter() {
+            output_device_rows.push(Row::new(vec![
+                device.description().unwrap().name().to_string(),
+                config.sample_format().to_string(),
+            ]));
+        }
+        let output_device_table = Table::new(output_device_rows, widths)
+            .header(
+                Row::new(vec!["device name", "channels"])
+                    .style(Style::new().bold())
+                    .bottom_margin(1),
+            )
+            .block(
+                Block::bordered()
+                    .title("select output device")
+                    .border_style({
+                        if self.selected_selection == 1 {
+                            Style::new().bold()
+                        } else {
+                            Style::default()
+                        }
+                    })
+                    .border_type({
+                        if self.selected_selection == 1 {
+                            ratatui::widgets::BorderType::Thick
+                        } else {
+                            ratatui::widgets::BorderType::Plain
+                        }
+                    }),
+            )
+            .row_highlight_style(Style::new().reversed());
+
+        frame.render_stateful_widget(
+            output_device_table,
+            right_block,
+            &mut self.output_device_state,
+        );
+    }
+
+    fn handle_events(&mut self) {
+        match event::read().expect("failed to read event") {
+            Event::Key(key_event) if key_event.kind == KeyEventKind::Press => {
+                self.handle_key_events(key_event);
+            }
+            _ => {}
+        }
+    }
+
+    fn handle_key_events(&mut self, key_event: KeyEvent) {
+        match key_event.code {
+            KeyCode::Char('q') => self.exit(),
+            KeyCode::Down => self.increment_table(),
+            KeyCode::Up => self.decrement_table(),
+            KeyCode::Left => self.selected_selection = 0,
+            KeyCode::Right => self.selected_selection = 1,
+            KeyCode::Enter => self.connect_devices(),
+            _ => {}
+        }
+    }
+
+    fn exit(&mut self) {
+        self.exit = true;
+    }
+
+    fn increment_table(&mut self) {
+        if self.selected_selection == 0 {
+            self.input_device_state.scroll_down_by(1);
+        } else {
+            self.output_device_state.scroll_down_by(1);
+        }
+    }
+
+    fn decrement_table(&mut self) {
+        if self.selected_selection == 0 {
+            self.input_device_state.scroll_up_by(1);
+        } else {
+            self.output_device_state.scroll_up_by(1);
+        }
+    }
+
+    fn connect_devices(&mut self) {
+        if let (Some(input_device_row), Some(output_device_row)) = (
+            self.input_device_state.selected(),
+            self.output_device_state.selected(),
+        ) {
+            let (input_device, input_config) = self
+                .input_devices
+                .get(input_device_row)
+                .expect("failed to get input config");
+            let (output_device, output_config) = self
+                .output_devices
+                .get(output_device_row)
+                .expect("failed to get output config");
+
+            tokio::spawn(audio_stream_receiver(
+                output_device.clone(),
+                output_config.clone(),
+                self.audio_control
+                    .active_input_audio_streams_producers
+                    .clone(),
+                self.audio_control
+                    .active_input_audio_streams_consumers
+                    .clone(),
+                self.audio_control
+                    .active_input_audio_streams_encoders
+                    .clone(),
+                self.audio_control.active_input_audio_streams_info.clone(),
+                self.audio_control.udp_socket.clone(),
+            ));
+
+            tokio::spawn(audio_stream_spawner(
+                input_device.clone(),
+                input_config.clone(),
+                self.global_event_tx.clone(),
+                self.global_event_tx.subscribe(),
+                self.audio_control.udp_socket.clone(),
+                self.client_info.clone(),
+            ));
+
+            self.exit_screen = TuiScreen::Blank;
+            self.exit = true;
+        }
+    }
+}

server/Cargo.toml

@@ -10,3 +10,5 @@ serde_json = "1.0.151"
 tokio = { version = "1.53.1", features = ["full"] }
 rand = "0.10.2"
 clap = { version = "4.6.6", features = ["derive"] }
+tokio-util = { version = "0.7.19", features = ["codec"]}
+futures = "0.3.34"

server/src/main.rs

@@ -1,17 +1,19 @@
 use clap::Parser;
+use futures::{SinkExt, StreamExt};
 use shared::{AudioStreamInfo, ResponseStatus, TcpRequest, TcpRequestData, TcpResponse};
 use std::sync::Arc;
 use std::{collections::HashMap, net::SocketAddr};
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::{
     net::{TcpListener, UdpSocket},
     sync::{RwLock, broadcast},
 };
+use tokio_util::bytes::Bytes;
+use tokio_util::codec::{Framed, LengthDelimitedCodec};
 
 #[derive(Parser)]
 #[command(version, about)]
 struct Cli {
-    #[arg(short, long, default_value("0.0.0.0:8080"))]
+    #[arg(short, long, default_value("0.0.0.0:6767"))]
     server_ip: Option<String>,
 }
 
@@ -93,8 +95,9 @@ async fn main() {
     });
 
     loop {
-        let (mut tcp_stream, tcp_address) =
-            tcp_listener.accept().await.expect("failed to accept tcp");
+        let (tcp_stream, tcp_address) = tcp_listener.accept().await.expect("failed to accept tcp");
+
+        let mut framed_tcp_stream = Framed::new(tcp_stream, LengthDelimitedCodec::new());
 
         let tcp_state = state.clone();
 
@@ -103,18 +106,18 @@ async fn main() {
 
         tokio::spawn(async move {
             let current_client_id: u32 = rand::random();
-            let mut tcp_buf = [0; 4096];
 
             loop {
                 tokio::select! {
-                    socket_result = tcp_stream.read(&mut tcp_buf) => {
-                        match socket_result {
-                            Ok(0) => {
+                    frame_result = framed_tcp_stream.next() => {
+                        match frame_result {
+                            None => {
                                 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]) {
+                            Some(Ok(bytes)) => {
+                                match serde_json::from_slice::<TcpRequest>(&bytes) {
                                     Ok(request) => match request.data {
                                         TcpRequestData::NewConnection => {
                                             tcp_state.write().await.insert(
@@ -126,65 +129,58 @@ async fn main() {
                                             );
 
                                             println!("new client: {}, from: {}", current_client_id, tcp_address);
-                                        let _ = tcp_stream
-                                            .write_all(
-                                                serde_json::to_string(&TcpResponse {
+
+                                            let response_json = 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;
+                                            }).expect("failed to serialize TcpRequestData::NewAudioStreamAccepted");
+
+                                            let _ = framed_tcp_stream.send(Bytes::from(response_json))
+                                            .await
+                                            .expect("failed to send message");
 
                                             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);
+                                                    audio_streams.push(audio_stream_info.clone());
                                                 } else {
-                                                client.audio_streams = Some(vec![audio_stream_info]);
+                                                    client.audio_streams = Some(vec![audio_stream_info.clone()]);
                                                 }
                                             }
-                                        let _ = tcp_stream
-                                            .write_all(
-                                                serde_json::to_string(&TcpResponse {
+
+                                            let response_json = 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;
+                                            }).expect("failed to serialize TcpRequestData::NewAudioStreamAccepted");
+
+                                            let _ = framed_tcp_stream.send(Bytes::from(response_json)).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 {
+                                                    if let Some(idx) = audio_streams.iter().position(|s| s.stream_id == audio_stream_id) {
                                                         audio_streams.remove(idx);
                                                     }
                                                 }
                                             }
                                             let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id));
                                         }
-                                    _ => 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),
+                                }
+                            }
+                            Some(Err(e)) => {
+                                eprintln!("socket read error: {}", e);
+                                break;
+                            }
                         }
                     }
                     event_result = global_event_rx.recv() => {
@@ -193,36 +189,28 @@ async fn main() {
                                 match event {
                                     ServerEvent::UserConnected(new_user_id) => {
                                         if new_user_id != current_client_id {
-                                            let _ = tcp_stream
-                                                .write_all(
-                                                    serde_json::to_string(&TcpResponse {
+                                            let response_json = 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;
+                                            }).expect("failed to serialize ServerEvent::UserConnected");
+
+                                            let _ = framed_tcp_stream.send(Bytes::from(response_json)).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 {
+                                            let response_json = 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;
+                                            }).expect("failed to serialize ServerEvent::UserDisconnected");
+
+                                            let _ = framed_tcp_stream.send(Bytes::from(response_json)).await;
 
                                             let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id));
                                         }
@@ -231,38 +219,30 @@ async fn main() {
                                         if current_client_id != id {
                                             let clients = tcp_state.read().await;
 
-                                            let _ = tcp_stream
-                                                .write_all(
-                                                    serde_json::to_string(&TcpResponse {
+                                            let response_json = 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;
+                                            }).expect("failed to serialize ServerEvent::SendAllOtherStreams");
+
+                                            let _ = framed_tcp_stream.send(Bytes::from(response_json)).await;
                                         }
                                     }
                                     ServerEvent::SendAllStreams => {
                                         let clients = tcp_state.read().await;
 
-                                        let _ = tcp_stream
-                                            .write_all(
-                                                serde_json::to_string(&TcpResponse {
+                                        let response_json = 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;
+                                        }).expect("failed to serialize ServerEvent::SendAllStreams");
+
+                                        let _ = framed_tcp_stream.send(Bytes::from(response_json)).await;
                                     }
                                     _ => {
-                                        println!("lkdahjsf");
+                                        println!("unhandled event");
                                     }
                                 }
                             }