diff --git a/client/src/audio_streams.rs b/client/src/audio_streams.rs index 5f743f3..96248ce 100644 --- a/client/src/audio_streams.rs +++ b/client/src/audio_streams.rs @@ -1,10 +1,12 @@ use cpal::{ - Device, StreamConfig, + Device, FromSample, SampleFormat, SizedSample, StreamConfig, SupportedStreamConfig, traits::{DeviceTrait, StreamTrait}, }; use ringbuf::{ - HeapRb, + HeapRb, SharedRb, + storage::Heap, traits::{Consumer, Observer, Producer, Split}, + wrap::caching::Caching, }; use ropus::Encoder; use shared::AudioStreamInfo; @@ -21,7 +23,7 @@ use crate::models::{ pub async fn audio_stream_receiver( output_device: Device, - output_device_config: StreamConfig, + output_device_config: SupportedStreamConfig, active_input_audio_streams_producers: InputAudioStreamsProducers, active_input_audio_streams_consumers: InputAudioStreamsConsumers, active_input_audio_streams_encoders: InputAudioStreamsEncoders, @@ -46,7 +48,7 @@ pub async fn audio_stream_receiver( let err_fn = |err| eprintln!("stream error: {}", err); let output_stream = output_device - .build_output_stream(output_device_config, output_data_fn, err_fn, None) + .build_output_stream(output_device_config.into(), output_data_fn, err_fn, None) .expect("building output stream failed"); output_stream.play().expect("failed to play output stream"); @@ -102,11 +104,11 @@ pub async fn audio_stream_receiver( 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 { + 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 { + } 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) @@ -121,7 +123,7 @@ pub async fn audio_stream_receiver( ); } else { for (i, frame) in decoded_samples_converted - [..decoded_len * output_device_config.channels as usize] + [..decoded_len * output_device_config.channels() as usize] .chunks_exact_mut(2) .enumerate() { @@ -131,7 +133,7 @@ pub async fn audio_stream_receiver( producer.push_slice( &decoded_samples_converted - [..decoded_len * output_device_config.channels as usize], + [..decoded_len * output_device_config.channels() as usize], ); } } @@ -141,27 +143,50 @@ pub async fn audio_stream_receiver( } } +fn cpal_build_input_stream( + input_device: &Device, + input_device_config: StreamConfig, + mut producer: Caching>>, true, false>, +) -> cpal::Stream +where + T: SizedSample, + f32: FromSample, +{ + let input_data_fn = move |data: &[T], _: &cpal::InputCallbackInfo| { + for &sample in data { + let f32_sample: f32 = sample.to_sample::(); + + 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: StreamConfig, + input_device_config: SupportedStreamConfig, global_event_tx: Sender, mut global_event_rx: Receiver, udp_socket: Arc, - clinet_info: Arc>, + client_info: Arc>, ) { let current_stream_id: u8 = rand::random(); loop { let (connected, client_id) = { - let info = clinet_info.read().unwrap(); + 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, + channels: input_device_config.channels(), + sample_rate: input_device_config.sample_rate(), })); break; } @@ -181,30 +206,38 @@ pub async fn audio_stream_spawner( } 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_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::::new(ring_buffer_size); - let (mut producer, mut consumer) = ring_buffer.split(); + let (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 input_stream = match input_device_config.sample_format() { + SampleFormat::F32 => { + cpal_build_input_stream::(&input_device, input_device_config.into(), producer) + } + SampleFormat::I32 => { + cpal_build_input_stream::(&input_device, input_device_config.into(), producer) + } + SampleFormat::I16 => { + cpal_build_input_stream::(&input_device, input_device_config.into(), producer) + } + SampleFormat::U16 => { + cpal_build_input_stream::(&input_device, input_device_config.into(), producer) + } + _ => { + eprintln!("unsupported stream format :("); + return; } }; - let err_fn = |err| eprintln!("stream error: {}", err); - - let input_stream = input_device - .build_input_stream(input_device_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) * input_device_config.sample_rate as f32) as usize; + 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 { + if input_device_config.channels() == 2 { ropus::Channels::Stereo } else { ropus::Channels::Mono @@ -212,7 +245,7 @@ pub async fn audio_stream_spawner( }; let mut opus_encoder = Encoder::builder( - input_device_config.sample_rate, + input_device_config.sample_rate(), channels, ropus::Application::Voip, ) @@ -228,7 +261,7 @@ pub async fn audio_stream_spawner( let mut payload = Vec::with_capacity(4 + 1 + frames_needed * 4); let (server_address, client_id) = { - let info = clinet_info.read().unwrap(); + let info = client_info.read().unwrap(); (info.server_address.clone(), info.client_id) }; diff --git a/client/src/tui.rs b/client/src/tui.rs index ff3cd11..f1ff322 100644 --- a/client/src/tui.rs +++ b/client/src/tui.rs @@ -1,5 +1,5 @@ use cpal::{ - Device, SAMPLE_RATE_48K, SampleFormat, StreamConfig, + Device, SAMPLE_RATE_48K, SampleFormat, SupportedStreamConfig, traits::{DeviceTrait, HostTrait}, }; use crossterm::event::{self, Event, KeyCode, KeyEvent, KeyEventKind}; @@ -100,8 +100,8 @@ struct DeviceSelectScreen { exit_screen: TuiScreen, input_device_state: TableState, output_device_state: TableState, - input_devices: Vec<(Device, StreamConfig)>, - output_devices: Vec<(Device, StreamConfig)>, + input_devices: Vec<(Device, SupportedStreamConfig)>, + output_devices: Vec<(Device, SupportedStreamConfig)>, selected_selection: u8, audio_control: Arc, @@ -119,25 +119,22 @@ impl DeviceSelectScreen { let host = cpal::default_host(); - let mut input_devices: Vec<(Device, StreamConfig)> = Vec::new(); + 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.sample_format() == SampleFormat::F32 && (default_config.channels() == 1) { - input_devices.push((default_input_device.clone(), default_config.into())); + 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.sample_format() == SampleFormat::F32 - && (input_config.channels() == 1) - { - input_devices.push((default_input_device, input_config.into())); + if input_config.channels() == 1 { + input_devices.push((default_input_device, input_config)); break; } } @@ -149,10 +146,9 @@ impl DeviceSelectScreen { 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.sample_format() == SampleFormat::F32 && (default_config.channels() == 1) { - input_devices.push((possible_input_device.clone(), default_config.into())); + input_devices.push((possible_input_device.clone(), default_config)); continue; } } @@ -161,10 +157,8 @@ impl DeviceSelectScreen { if let Some(input_config) = supported_config.try_with_sample_rate(SAMPLE_RATE_48K) { - if input_config.sample_format() == SampleFormat::F32 - && (input_config.channels() == 1) - { - input_devices.push((possible_input_device, input_config.into())); + if input_config.channels() == 1 { + input_devices.push((possible_input_device, input_config)); break; } } @@ -173,7 +167,7 @@ impl DeviceSelectScreen { } } - let mut output_devices: Vec<(Device, StreamConfig)> = Vec::new(); + 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() { @@ -276,7 +270,7 @@ impl DeviceSelectScreen { for (device, config) in self.input_devices.iter() { input_device_rows.push(Row::new(vec![ device.description().unwrap().name().to_string(), - config.channels.to_string(), + config.channels().to_string(), ])); } let input_device_table = Table::new(input_device_rows, widths) @@ -311,7 +305,7 @@ impl DeviceSelectScreen { for (device, config) in self.output_devices.iter() { output_device_rows.push(Row::new(vec![ device.description().unwrap().name().to_string(), - config.channels.to_string(), + config.channels().to_string(), ])); } let output_device_table = Table::new(output_device_rows, widths)