diff --git a/CHANGELOG.md b/CHANGELOG.md index 54e6def03..655c43a73 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `StreamTrait::stop` ends a stream gracefully, draining buffered audio before halting (blocking up to a caller-supplied timeout). Dropping a stream still halts immediately without draining. - `CallbackInfo::xrun()` reports buffer over/underruns via the data callback. +- **AudioWorklet**: Input streams are now supported. +- **WebAudio**: Input streams are now supported. ### Changed @@ -35,6 +37,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - **ALSA**: Fix a remaining timestamp segfault on 32-bit platforms with a 64-bit kernel `time_t`. +- **AudioWorklet**: Fix processor construction failures not being reported to `error_callback`. ## [Unreleased] (v0.18.2) diff --git a/Cargo.toml b/Cargo.toml index 5afca71b4..6e2520f13 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -46,6 +46,8 @@ audioworklet = [ "web-sys/AudioWorklet", "web-sys/AudioWorkletNode", "web-sys/AudioWorkletNodeOptions", + "web-sys/ChannelCountMode", + "web-sys/Event", ] # Support for user-defined custom hosts, devices, and streams @@ -87,6 +89,16 @@ wasm-bindgen = [ "dep:wasm-bindgen-futures", "dep:futures-channel", "dep:futures-util", + "web-sys/Navigator", + "web-sys/MediaDevices", + "web-sys/MediaStream", + "web-sys/MediaStreamConstraints", + "web-sys/MediaStreamTrack", + "web-sys/MediaStreamAudioSourceNode", + "web-sys/ScriptProcessorNode", + "web-sys/AudioProcessingEvent", + "web-sys/GainNode", + "web-sys/AudioParam", ] [dependencies] diff --git a/examples/audioworklet-beep/.cargo/config.toml b/examples/audioworklet-beep/.cargo/config.toml index c6df64da1..5eccee0f8 100644 --- a/examples/audioworklet-beep/.cargo/config.toml +++ b/examples/audioworklet-beep/.cargo/config.toml @@ -16,6 +16,8 @@ rustflags = [ "link-arg=--export=__tls_align", "-C", "link-arg=--export=__tls_base", + "-C", + "link-arg=--export=__heap_base", ] [unstable] diff --git a/examples/audioworklet-beep/Cargo.toml b/examples/audioworklet-beep/Cargo.toml index e7afd3039..ebf6bc2ec 100644 --- a/examples/audioworklet-beep/Cargo.toml +++ b/examples/audioworklet-beep/Cargo.toml @@ -27,6 +27,9 @@ wasm-bindgen = "0.2" # logging them with `console.error`. console_error_panic_hook = "0.1" +# The `ringbuf` crate provides a lock-free ring buffer for passing audio between streams. +ringbuf = "0.4" + # The `web-sys` crate allows you to interact with the various browser APIs, # like the DOM. [dependencies.web-sys] diff --git a/examples/audioworklet-beep/index.html b/examples/audioworklet-beep/index.html index 6a748f05b..6e3139fc6 100644 --- a/examples/audioworklet-beep/index.html +++ b/examples/audioworklet-beep/index.html @@ -9,6 +9,9 @@ + + +

Recording plays your microphone back live. Wear headphones to avoid feedback howl.

\ No newline at end of file diff --git a/examples/audioworklet-beep/src/lib.rs b/examples/audioworklet-beep/src/lib.rs index b4af1ce92..d39979ef0 100644 --- a/examples/audioworklet-beep/src/lib.rs +++ b/examples/audioworklet-beep/src/lib.rs @@ -1,9 +1,13 @@ use std::{cell::Cell, rc::Rc}; use cpal::{ - traits::{DeviceTrait, HostTrait, StreamTrait}, Device, Error, ErrorKind, FromSample, HostId, Sample, SampleFormat, SizedSample, Stream, StreamConfig, + traits::{DeviceTrait, HostTrait, StreamTrait}, +}; +use ringbuf::{ + HeapCons, HeapProd, HeapRb, + traits::{Consumer, Producer, Split}, }; use wasm_bindgen::prelude::*; use web_sys::console; @@ -19,6 +23,8 @@ pub fn main_js() -> Result<(), JsValue> { let document = gloo::utils::document(); let play_button = document.get_element_by_id("play").unwrap(); let stop_button = document.get_element_by_id("stop").unwrap(); + let record_button = document.get_element_by_id("record").unwrap(); + let stop_record_button = document.get_element_by_id("stop-record").unwrap(); // stream needs to be referenced from the "play" and "stop" closures let stream = Rc::new(Cell::new(None)); @@ -45,12 +51,36 @@ pub fn main_js() -> Result<(), JsValue> { closure.forget(); } + // input stream needs its own slot; recording and playback run independently + let record_stream = Rc::new(Cell::new(None)); + + // set up record button + { + let record_stream = record_stream.clone(); + let closure = Closure::::new(move |_event: web_sys::MouseEvent| { + record_stream.set(Some(record())); + }); + record_button + .add_event_listener_with_callback("mousedown", closure.as_ref().unchecked_ref())?; + closure.forget(); + } + + // set up stop-record button + { + let closure = Closure::::new(move |_event: web_sys::MouseEvent| { + // stop the stream by dropping it; releases the microphone + record_stream.take(); + }); + stop_record_button + .add_event_listener_with_callback("mousedown", closure.as_ref().unchecked_ref())?; + closure.forget(); + } + Ok(()) } fn beep() -> Stream { - let host = - cpal::host_from_id(HostId::AudioWorklet).expect("AudioWorklet host not available"); + let host = cpal::host_from_id(HostId::AudioWorklet).expect("AudioWorklet host not available"); let device = host .default_output_device() @@ -98,6 +128,98 @@ where stream } +/// Captures microphone input into a ring buffer and immediately plays it back, so you can hear +/// your own voice. Wear headphones: routing a live mic to speakers risks feedback howl. +fn record() -> (Stream, Stream) { + let host = cpal::host_from_id(HostId::AudioWorklet).expect("AudioWorklet host not available"); + + let input_device = host + .default_input_device() + .expect("failed to find a default input device"); + let output_device = host + .default_output_device() + .expect("failed to find a default output device"); + + let input_config = input_device.default_input_config().unwrap(); + let output_config = output_device.default_output_config().unwrap(); + + // Bound end-to-end latency; once full, the producer drops the newest samples instead of + // blocking. + let max_buffered_samples = + output_config.sample_rate() as usize * output_config.channels() as usize / 2; + let ring = HeapRb::::new(max_buffered_samples); + let (producer, consumer) = ring.split(); + + let input_stream = match input_config.sample_format() { + SampleFormat::F32 => build_input::(&input_device, input_config.into(), producer), + SampleFormat::I16 => build_input::(&input_device, input_config.into(), producer), + SampleFormat::U16 => build_input::(&input_device, input_config.into(), producer), + _ => panic!("unsupported sample format"), + }; + let output_stream = match output_config.sample_format() { + SampleFormat::F32 => build_output::(&output_device, output_config.into(), consumer), + SampleFormat::I16 => build_output::(&output_device, output_config.into(), consumer), + SampleFormat::U16 => build_output::(&output_device, output_config.into(), consumer), + _ => panic!("unsupported sample format"), + }; + + (input_stream, output_stream) +} + +fn build_input(device: &Device, config: StreamConfig, mut producer: HeapProd) -> Stream +where + T: Sample + SizedSample, + f32: FromSample, +{ + let err_fn = |err: Error| match err.kind() { + ErrorKind::DeviceChanged | ErrorKind::RealtimeDenied => { + console::log_1(&format!("{err}").into()) + } + _ => console::error_1(&format!("Stream error: {err}").into()), + }; + + let stream = device + .build_input_stream( + config, + move |data: &[T], _| { + producer.push_iter(data.iter().map(|&s| f32::from_sample(s))); + }, + err_fn, + None, + ) + .unwrap(); + stream.start().unwrap(); + stream +} + +fn build_output(device: &Device, config: StreamConfig, mut consumer: HeapCons) -> Stream +where + T: Sample + SizedSample + FromSample, +{ + let err_fn = |err: Error| match err.kind() { + ErrorKind::DeviceChanged | ErrorKind::RealtimeDenied => { + console::log_1(&format!("{err}").into()) + } + _ => console::error_1(&format!("Stream error: {err}").into()), + }; + + let stream = device + .build_output_stream( + config, + move |data: &mut [T], _| { + for sample in data.iter_mut() { + let value = consumer.try_pop().unwrap_or(f32::EQUILIBRIUM); + *sample = T::from_sample(value); + } + }, + err_fn, + None, + ) + .unwrap(); + stream.start().unwrap(); + stream +} + fn write_data(output: &mut [T], channels: usize, next_sample: &mut dyn FnMut() -> f32) where T: Sample + FromSample, diff --git a/examples/wasm-beep/Cargo.toml b/examples/wasm-beep/Cargo.toml index b906f8d76..eec5778f7 100644 --- a/examples/wasm-beep/Cargo.toml +++ b/examples/wasm-beep/Cargo.toml @@ -27,6 +27,9 @@ wasm-bindgen = "0.2" # logging them with `console.error`. console_error_panic_hook = "0.1" +# The `ringbuf` crate provides a lock-free ring buffer for passing audio between streams. +ringbuf = "0.4" + # The `web-sys` crate allows you to interact with the various browser APIs, # like the DOM. [dependencies.web-sys] diff --git a/examples/wasm-beep/index.html b/examples/wasm-beep/index.html index 5cbecc110..344eecdf3 100644 --- a/examples/wasm-beep/index.html +++ b/examples/wasm-beep/index.html @@ -7,5 +7,8 @@ + + +

Recording plays your microphone back live. Wear headphones to avoid feedback howl.

diff --git a/examples/wasm-beep/src/lib.rs b/examples/wasm-beep/src/lib.rs index f4281753d..1e9c8f0b7 100644 --- a/examples/wasm-beep/src/lib.rs +++ b/examples/wasm-beep/src/lib.rs @@ -1,8 +1,12 @@ use std::{cell::Cell, rc::Rc}; use cpal::{ - traits::{DeviceTrait, HostTrait, StreamTrait}, Device, Error, ErrorKind, FromSample, Sample, SampleFormat, SizedSample, Stream, StreamConfig, + traits::{DeviceTrait, HostTrait, StreamTrait}, +}; +use ringbuf::{ + HeapCons, HeapProd, HeapRb, + traits::{Consumer, Producer, Split}, }; use wasm_bindgen::prelude::*; use web_sys::console; @@ -18,6 +22,8 @@ pub fn main_js() -> Result<(), JsValue> { let document = gloo::utils::document(); let play_button = document.get_element_by_id("play").unwrap(); let stop_button = document.get_element_by_id("stop").unwrap(); + let record_button = document.get_element_by_id("record").unwrap(); + let stop_record_button = document.get_element_by_id("stop-record").unwrap(); // stream needs to be referenced from the "play" and "stop" closures let stream = Rc::new(Cell::new(None)); @@ -44,6 +50,31 @@ pub fn main_js() -> Result<(), JsValue> { closure.forget(); } + // input stream needs its own slot; recording and playback run independently + let record_stream = Rc::new(Cell::new(None)); + + // set up record button + { + let record_stream = record_stream.clone(); + let closure = Closure::::new(move |_event: web_sys::MouseEvent| { + record_stream.set(Some(record())); + }); + record_button + .add_event_listener_with_callback("mousedown", closure.as_ref().unchecked_ref())?; + closure.forget(); + } + + // set up stop-record button + { + let closure = Closure::::new(move |_event: web_sys::MouseEvent| { + // stop the stream by dropping it; releases the microphone + record_stream.take(); + }); + stop_record_button + .add_event_listener_with_callback("mousedown", closure.as_ref().unchecked_ref())?; + closure.forget(); + } + Ok(()) } @@ -95,6 +126,97 @@ where stream } +/// Captures microphone input into a ring buffer and immediately plays it back, so you can hear +/// your own voice. Wear headphones: routing a live mic to speakers risks feedback howl. +fn record() -> (Stream, Stream) { + let host = cpal::default_host(); + let input_device = host + .default_input_device() + .expect("failed to find a default input device"); + let output_device = host + .default_output_device() + .expect("failed to find a default output device"); + + let input_config = input_device.default_input_config().unwrap(); + let output_config = output_device.default_output_config().unwrap(); + + // Bound end-to-end latency; once full, the producer drops the newest samples instead of + // blocking. + let max_buffered_samples = + output_config.sample_rate() as usize * output_config.channels() as usize / 2; + let ring = HeapRb::::new(max_buffered_samples); + let (producer, consumer) = ring.split(); + + let input_stream = match input_config.sample_format() { + SampleFormat::F32 => build_input::(&input_device, input_config.into(), producer), + SampleFormat::I16 => build_input::(&input_device, input_config.into(), producer), + SampleFormat::U16 => build_input::(&input_device, input_config.into(), producer), + _ => panic!("unsupported sample format"), + }; + let output_stream = match output_config.sample_format() { + SampleFormat::F32 => build_output::(&output_device, output_config.into(), consumer), + SampleFormat::I16 => build_output::(&output_device, output_config.into(), consumer), + SampleFormat::U16 => build_output::(&output_device, output_config.into(), consumer), + _ => panic!("unsupported sample format"), + }; + + (input_stream, output_stream) +} + +fn build_input(device: &Device, config: StreamConfig, mut producer: HeapProd) -> Stream +where + T: Sample + SizedSample, + f32: FromSample, +{ + let err_fn = |err: Error| match err.kind() { + ErrorKind::DeviceChanged | ErrorKind::RealtimeDenied => { + console::log_1(&format!("{err}").into()) + } + _ => console::error_1(&format!("Stream error: {err}").into()), + }; + + let stream = device + .build_input_stream( + config, + move |data: &[T], _| { + producer.push_iter(data.iter().map(|&s| f32::from_sample(s))); + }, + err_fn, + None, + ) + .unwrap(); + stream.start().unwrap(); + stream +} + +fn build_output(device: &Device, config: StreamConfig, mut consumer: HeapCons) -> Stream +where + T: Sample + SizedSample + FromSample, +{ + let err_fn = |err: Error| match err.kind() { + ErrorKind::DeviceChanged | ErrorKind::RealtimeDenied => { + console::log_1(&format!("{err}").into()) + } + _ => console::error_1(&format!("Stream error: {err}").into()), + }; + + let stream = device + .build_output_stream( + config, + move |data: &mut [T], _| { + for sample in data.iter_mut() { + let value = consumer.try_pop().unwrap_or(f32::EQUILIBRIUM); + *sample = T::from_sample(value); + } + }, + err_fn, + None, + ) + .unwrap(); + stream.start().unwrap(); + stream +} + fn write_data(output: &mut [T], channels: usize, next_sample: &mut dyn FnMut() -> f32) where T: Sample + FromSample, diff --git a/src/host/audioworklet/mod.rs b/src/host/audioworklet/mod.rs index 5d7647836..aeb077b17 100644 --- a/src/host/audioworklet/mod.rs +++ b/src/host/audioworklet/mod.rs @@ -4,7 +4,9 @@ //! See the `audioworklet-beep` example for setup instructions. use std::{ + cell::RefCell, fmt, + rc::Rc, sync::{ Arc, atomic::{AtomicU64, Ordering}, @@ -72,6 +74,10 @@ const SUPPORTED_SAMPLE_FORMAT: SampleFormat = SampleFormat::F32; // https://webaudio.github.io/web-audio-api/#render-quantum-size const DEFAULT_RENDER_SIZE: u64 = 128; +// Must match the names passed to `registerProcessor()` in worklet.js. +const OUTPUT_PROCESSOR_NAME: &str = "CpalProcessor"; +const CAPTURE_PROCESSOR_NAME: &str = "CpalCaptureProcessor"; + fn render_quantum_size_supported() -> bool { (|| -> Option { let global = js_sys::global(); @@ -97,6 +103,85 @@ fn supported_render_quantum_range(sample_rate: SampleRate) -> SupportedBufferSiz } } +/// Checks shared by both `build_input_stream_raw` and `build_output_stream_raw`. +fn validate_config(config: &StreamConfig, sample_format: SampleFormat) -> Result<(), Error> { + crate::validate_stream_config(config)?; + if config.channels > MAX_CHANNELS { + return Err(Error::with_message( + ErrorKind::UnsupportedConfig, + format!( + "Channel count {} exceeds the maximum of {MAX_CHANNELS}", + config.channels + ), + )); + } + if sample_format != SUPPORTED_SAMPLE_FORMAT { + return Err(Error::with_message( + ErrorKind::UnsupportedConfig, + format!( + "Sample format {sample_format} is not supported; required format is {SUPPORTED_SAMPLE_FORMAT}" + ), + )); + } + if !(MIN_SAMPLE_RATE..=MAX_SAMPLE_RATE).contains(&config.sample_rate) { + return Err(Error::with_message( + ErrorKind::UnsupportedConfig, + format!( + "Sample rate {} Hz is not in the supported range {MIN_SAMPLE_RATE}..={MAX_SAMPLE_RATE} Hz", + config.sample_rate + ), + )); + } + if let BufferSize::Fixed(n) = config.buffer_size { + if let SupportedBufferSize::Range { min, max } = + supported_render_quantum_range(config.sample_rate) + { + if !(min..=max).contains(&n) { + return Err(Error::with_message( + ErrorKind::UnsupportedConfig, + format!( + "Buffer size {n} is not in the supported render quantum range {min}..={max}" + ), + )); + } + } + } + Ok(()) +} + +/// The full matrix of channel counts x sample rates; identical for input and output, since +/// neither is known ahead of time (see the callers' doc comments). +fn supported_configs() -> Vec { + (MIN_CHANNELS..=MAX_CHANNELS) + .flat_map(|channels| { + crate::COMMON_SAMPLE_RATES + .iter() + .copied() + .filter(|&r| (MIN_SAMPLE_RATE..=MAX_SAMPLE_RATE).contains(&r)) + .map(move |rate| SupportedStreamConfigRange { + channels, + min_sample_rate: rate, + max_sample_rate: rate, + buffer_size: supported_render_quantum_range(rate), + sample_format: SUPPORTED_SAMPLE_FORMAT, + }) + }) + .collect() +} + +/// Picks the best default from a supported-config iterator via the standard heuristics. +fn default_config( + configs: impl Iterator, + none_supported_message: &'static str, +) -> Result { + let range = configs + .max_by(|a, b| a.cmp_default_heuristics(b)) + .ok_or_else(|| Error::with_message(ErrorKind::UnsupportedConfig, none_supported_message))?; + Ok(range + .try_with_standard_sample_rate() + .unwrap_or_else(|| range.with_max_sample_rate())) +} + enum Command { Play, Pause, @@ -141,8 +226,11 @@ impl HostTrait for Host { } fn default_input_device(&self) -> Option { - // TODO - None + if Self::is_available() && crate::host::is_get_user_media_available() { + Some(Device) + } else { + None + } } fn default_output_device(&self) -> Option { @@ -166,8 +254,13 @@ impl DeviceTrait for Device { type Stream = Stream; fn description(&self) -> Result { + let direction = if crate::host::is_get_user_media_available() { + DeviceDirection::Duplex + } else { + DeviceDirection::Output + }; Ok(DeviceDescriptionBuilder::new("Default Device") - .direction(DeviceDirection::Output) + .direction(direction) .build()) } @@ -179,72 +272,260 @@ impl DeviceTrait for Device { } fn supported_input_configs(&self) -> Result { - // TODO - Ok(Vec::new().into_iter()) + // The actual channel count and sample rate depend on the microphone getUserMedia() + // grants access to, which isn't known ahead of time; WebAudio resamples and up/downmixes + // whatever it gets to match, so this reports the same broad matrix as output. + Ok(supported_configs().into_iter()) } fn supported_output_configs(&self) -> Result { // In actuality the number of supported channels cannot be fully known until - // the browser attempts to initialized the AudioWorklet. - - let configs: Vec<_> = (MIN_CHANNELS..=MAX_CHANNELS) - .flat_map(|channels| { - crate::COMMON_SAMPLE_RATES - .iter() - .copied() - .filter(|&r| (MIN_SAMPLE_RATE..=MAX_SAMPLE_RATE).contains(&r)) - .map(move |rate| SupportedStreamConfigRange { - channels, - min_sample_rate: rate, - max_sample_rate: rate, - buffer_size: supported_render_quantum_range(rate), - sample_format: SUPPORTED_SAMPLE_FORMAT, - }) - }) - .collect(); - Ok(configs.into_iter()) + // the browser attempts to initialize the AudioWorklet. + Ok(supported_configs().into_iter()) } fn default_input_config(&self) -> Result { - Err(Error::with_message( - ErrorKind::UnsupportedOperation, - "Device does not support input", - )) + default_config( + self.supported_input_configs()?, + "No supported input configuration", + ) } fn default_output_config(&self) -> Result { - let range = self - .supported_output_configs()? - .max_by(|a, b| a.cmp_default_heuristics(b)) - .ok_or_else(|| { - Error::with_message( - ErrorKind::UnsupportedConfig, - "No supported output configuration", - ) - })?; - let config = range - .try_with_standard_sample_rate() - .unwrap_or_else(|| range.with_max_sample_rate()); - - Ok(config) + default_config( + self.supported_output_configs()?, + "No supported output configuration", + ) } + /// Create an input stream capturing microphone audio via `getUserMedia()`. + /// + /// # Async completion + /// + /// This function returns `Ok` synchronously once the [`AudioContext`] is created, before + /// microphone access has been granted or denied and before the AudioWorklet module has been + /// loaded. Both happen asynchronously via [`wasm_bindgen_futures::spawn_local`]; if the user + /// denies access, no microphone is present, or the worklet fails to initialize, the error is + /// delivered to `error_callback` after the caller already holds a [`Stream`]. There is no way + /// to surface such errors synchronously given the Web Audio API's design. + /// + /// [`start`](crate::traits::StreamTrait::start) and [`pause`](crate::traits::StreamTrait::pause) + /// calls made before initialization completes return `Ok` immediately and are queued. If + /// initialization succeeds, the queued commands take effect; if it fails they are discarded + /// and the error is delivered to `error_callback`. + /// + /// [`AudioContext`]: web_sys::AudioContext fn build_input_stream_raw( &self, - _config: StreamConfig, - _sample_format: SampleFormat, - _data_callback: D, - _error_callback: E, + config: StreamConfig, + sample_format: SampleFormat, + data_callback: D, + error_callback: E, _timeout: Option, ) -> Result where D: FnMut(&Data, &CallbackInfo) + Send + 'static, E: FnMut(Error) + Send + 'static, { - Err(Error::with_message( - ErrorKind::UnsupportedOperation, - "Device does not support input", - )) + validate_config(&config, sample_format)?; + let mut data_callback = crate::host::monotonic_input_callback(data_callback); + + let n_channels = config.channels as u32; + + let stream_opts = web_sys::AudioContextOptions::new(); + stream_opts.set_sample_rate(config.sample_rate as f32); + if let BufferSize::Fixed(n) = config.buffer_size { + let _ = js_sys::Reflect::set( + stream_opts.as_ref(), + &JsValue::from_str("renderSizeHint"), + &JsValue::from_f64(n as f64), + ); + } + + let audio_context = + web_sys::AudioContext::new_with_context_options(&stream_opts).map_err(|_| { + Error::with_message( + ErrorKind::UnsupportedConfig, + "Failed to create audio context", + ) + })?; + + // Chrome rounds renderSizeHint to a power of two; read back the actual quantum. + let actual_render_quantum = + js_sys::Reflect::get(audio_context.as_ref(), &JsValue::from("renderQuantumSize")) + .ok() + .and_then(|v| v.as_f64()) + .map(|v| v as u64); + + let initial_quantum = actual_render_quantum.unwrap_or(match config.buffer_size { + BufferSize::Fixed(n) => n as u64, + BufferSize::Default => DEFAULT_RENDER_SIZE, + }); + let buffer_size_frames = Arc::new(AtomicU64::new(initial_quantum)); + let buffer_size_frames_cb = buffer_size_frames.clone(); + + let current_time_bits = Arc::new(AtomicU64::new(audio_context.current_time().to_bits())); + let current_time_bits_cb = current_time_bits.clone(); + let current_time_bits_init = current_time_bits.clone(); + + let (command_tx, mut command_rx) = mpsc::unbounded::(); + + let ctx = audio_context.clone(); + wasm_bindgen_futures::spawn_local(async move { + let error_callback = Rc::new(RefCell::new(error_callback)); + + let media_stream = match crate::host::request_microphone().await { + Ok(stream) => stream, + Err(js_err) => { + (error_callback.borrow_mut())(crate::host::get_user_media_error(&js_err)); + let _ = audio_context.close(); + return; + } + }; + + let result: Result< + ( + web_sys::MediaStreamAudioSourceNode, + web_sys::AudioWorkletNode, + web_sys::GainNode, + ), + JsValue, + > = async { + let mod_url = dependent_module!("worklet.js")?; + wasm_bindgen_futures::JsFuture::from(ctx.audio_worklet()?.add_module(&mod_url)?) + .await?; + + let source = ctx.create_media_stream_source(&media_stream)?; + + let options = web_sys::AudioWorkletNodeOptions::new(); + options.set_number_of_inputs(1); + // A node with zero outputs and nothing downstream has no path to the + // destination, so the graph never pulls it for processing (the same class of + // bug as ScriptProcessorNode requiring a real output in some browsers). Give it + // one silent output, muted below, purely to keep it in the render graph. + options.set_number_of_outputs(1); + options.set_output_channel_count(&js_sys::Array::of1(&JsValue::from_f64(1.0))); + // `config.channels` is a promise to the caller: every callback delivers exactly + // that many interleaved channels. Force WebAudio to up/downmix the microphone + // track to match, regardless of how many channels it actually carries. + options.set_channel_count(n_channels); + options.set_channel_count_mode(web_sys::ChannelCountMode::Explicit); + + options.set_processor_options(Some(&js_sys::Array::of3( + &wasm_bindgen::module(), + &wasm_bindgen::memory(), + &WasmAudioCaptureProcessor::new(Box::new( + move |interleaved_data, frame_size, sample_rate, now| { + buffer_size_frames_cb.store(frame_size as u64, Ordering::Relaxed); + current_time_bits_cb.store(now.to_bits(), Ordering::Relaxed); + let data = interleaved_data.as_ptr() as *mut (); + let data = unsafe { + Data::from_parts(data, interleaved_data.len(), sample_format) + }; + + let callback = StreamInstant::from_secs_f64(now); + let buffer_duration = + frames_to_duration(frame_size as FrameCount, sample_rate); + let device = callback.checked_sub(buffer_duration).unwrap_or(callback); + let timestamp = StreamTimestamp { callback, device }; + let info = CallbackInfo { + timestamp, + xrun: false, + }; + (data_callback)(&data, &info); + }, + )) + .pack() + .into(), + ))); + let audio_worklet_node = web_sys::AudioWorkletNode::new_with_options( + &ctx, + CAPTURE_PROCESSOR_NAME, + &options, + )?; + + let error_callback_setup = error_callback.clone(); + let on_processor_error = + Closure::::new(move |_event: web_sys::Event| { + (error_callback_setup.borrow_mut())(Error::with_message( + ErrorKind::BackendError, + "AudioWorklet capture processor failed to initialize or crashed", + )); + }); + audio_worklet_node + .set_onprocessorerror(Some(on_processor_error.as_ref().unchecked_ref())); + on_processor_error.forget(); + + source.connect_with_audio_node(&audio_worklet_node)?; + + // Route the node's silent dummy output through a muted gain to the + // destination, keeping it in the render graph without producing audible sound. + let mute_gain = web_sys::GainNode::new(&ctx)?; + mute_gain.gain().set_value(0.0); + audio_worklet_node.connect_with_audio_node(&mute_gain)?; + mute_gain.connect_with_audio_node(&ctx.destination())?; + + Ok((source, audio_worklet_node, mute_gain)) + } + .await; + + // Unlike AudioWorkletNode/GainNode, a MediaStreamAudioSourceNode isn't an + // AudioScheduledSourceNode with a spec-guaranteed lifetime tied to the render graph: + // browsers may garbage-collect it once its last JS reference is dropped, silently + // killing the mic connection a few render quanta later. Keep all three alive for as + // long as the command loop below runs, i.e. until the Stream is dropped. + let (_source, _audio_worklet_node, _mute_gain) = match result { + Ok(nodes) => nodes, + Err(err) => { + let message = err + .as_string() + .unwrap_or_else(|| "Failed to initialize audio worklet".to_string()); + (error_callback.borrow_mut())(Error::with_message( + ErrorKind::HostUnavailable, + message, + )); + + crate::host::stop_tracks(&media_stream); + let _ = audio_context.close(); + return; + } + }; + + current_time_bits_init.store(audio_context.current_time().to_bits(), Ordering::Relaxed); + + // Process play/pause commands from any thread until Stream is dropped. + // Dropping Stream closes command_tx, which terminates this loop. + while let Some(cmd) = command_rx.next().await { + match cmd { + Command::Play => { + if audio_context.resume().is_err() { + (error_callback.borrow_mut())(Error::with_message( + ErrorKind::DeviceNotAvailable, + "Failed to resume audio context", + )); + } + } + Command::Pause => { + if audio_context.suspend().is_err() { + (error_callback.borrow_mut())(Error::with_message( + ErrorKind::DeviceNotAvailable, + "Failed to suspend audio context", + )); + } + } + } + } + + // Stream dropped: release the microphone and close the AudioContext. + crate::host::stop_tracks(&media_stream); + let _ = audio_context.close(); + }); + + Ok(Self::Stream { + command_tx, + current_time_bits, + buffer_size_frames, + }) } /// Create an output stream. @@ -276,58 +557,17 @@ impl DeviceTrait for Device { config: StreamConfig, sample_format: SampleFormat, data_callback: D, - mut error_callback: E, + error_callback: E, _timeout: Option, ) -> Result where D: FnMut(&mut Data, &CallbackInfo) + Send + 'static, E: FnMut(Error) + Send + 'static, { - crate::validate_stream_config(&config)?; + validate_config(&config, sample_format)?; // Keep `device` monotonic: the polled outputLatency can drop when the // page calls `setSinkId()` to switch output devices, pulling `device` backward. let mut data_callback = crate::host::monotonic_output_callback(data_callback); - if config.channels > MAX_CHANNELS { - return Err(Error::with_message( - ErrorKind::UnsupportedConfig, - format!( - "Channel count {} exceeds the maximum of {MAX_CHANNELS}", - config.channels - ), - )); - } - if sample_format != SUPPORTED_SAMPLE_FORMAT { - return Err(Error::with_message( - ErrorKind::UnsupportedConfig, - format!( - "Sample format {sample_format} is not supported; required format is {SUPPORTED_SAMPLE_FORMAT}" - ), - )); - } - if !(MIN_SAMPLE_RATE..=MAX_SAMPLE_RATE).contains(&config.sample_rate) { - return Err(Error::with_message( - ErrorKind::UnsupportedConfig, - format!( - "Sample rate {} Hz is not in the supported range {MIN_SAMPLE_RATE}..={MAX_SAMPLE_RATE} Hz", - config.sample_rate - ), - )); - } - - if let BufferSize::Fixed(n) = config.buffer_size { - if let SupportedBufferSize::Range { min, max } = - supported_render_quantum_range(config.sample_rate) - { - if !(min..=max).contains(&n) { - return Err(Error::with_message( - ErrorKind::UnsupportedConfig, - format!( - "Buffer size {n} is not in the supported render quantum range {min}..={max}" - ), - )); - } - } - } let stream_opts = web_sys::AudioContextOptions::new(); stream_opts.set_sample_rate(config.sample_rate as f32); @@ -386,6 +626,9 @@ impl DeviceTrait for Device { let ctx = audio_context.clone(); wasm_bindgen_futures::spawn_local(async move { + let error_callback = Rc::new(RefCell::new(error_callback)); + let error_callback_setup = error_callback.clone(); + let result: Result<(), JsValue> = async move { let mod_url = dependent_module!("worklet.js")?; wasm_bindgen_futures::JsFuture::from(ctx.audio_worklet()?.add_module(&mod_url)?) @@ -428,9 +671,22 @@ impl DeviceTrait for Device { .pack() .into(), ))); - // This name 'CpalProcessor' must match the name registered in worklet.js - let audio_worklet_node = - web_sys::AudioWorkletNode::new_with_options(&ctx, "CpalProcessor", &options)?; + let audio_worklet_node = web_sys::AudioWorkletNode::new_with_options( + &ctx, + OUTPUT_PROCESSOR_NAME, + &options, + )?; + + let on_processor_error = + Closure::::new(move |_event: web_sys::Event| { + (error_callback_setup.borrow_mut())(Error::with_message( + ErrorKind::BackendError, + "AudioWorklet processor failed to initialize or crashed", + )); + }); + audio_worklet_node + .set_onprocessorerror(Some(on_processor_error.as_ref().unchecked_ref())); + on_processor_error.forget(); audio_worklet_node.connect_with_audio_node(&destination)?; Ok(()) @@ -441,7 +697,10 @@ impl DeviceTrait for Device { let message = err .as_string() .unwrap_or_else(|| "Failed to initialize audio worklet".to_string()); - error_callback(Error::with_message(ErrorKind::HostUnavailable, message)); + (error_callback.borrow_mut())(Error::with_message( + ErrorKind::HostUnavailable, + message, + )); // Close AudioContext and exit; dropping command_rx closes the channel, // so subsequent play()/pause() calls return HostUnavailable. @@ -479,7 +738,7 @@ impl DeviceTrait for Device { match cmd { Command::Play => { if audio_context.resume().is_err() { - error_callback(Error::with_message( + (error_callback.borrow_mut())(Error::with_message( ErrorKind::DeviceNotAvailable, "Failed to resume audio context", )); @@ -487,7 +746,7 @@ impl DeviceTrait for Device { } Command::Pause => { if audio_context.suspend().is_err() { - error_callback(Error::with_message( + (error_callback.borrow_mut())(Error::with_message( ErrorKind::DeviceNotAvailable, "Failed to suspend audio context", )); @@ -631,6 +890,72 @@ impl WasmAudioProcessor { } } +type AudioCaptureCallback = Box; + +/// WasmAudioCaptureProcessor provides an interface for the JavaScript code running in the +/// AudioWorklet to hand captured microphone audio to Rust. The mirror image of +/// [`WasmAudioProcessor`]: JS interleaves the input channels into a Rust-owned buffer instead of +/// Rust filling a buffer for JS to deinterleave out. +#[wasm_bindgen] +pub struct WasmAudioCaptureProcessor { + interleaved_buffer: Vec, + // Receives the interleaved captured buffer, frame size, sample rate, and current time. + callback: AudioCaptureCallback, +} + +impl WasmAudioCaptureProcessor { + pub fn new(callback: AudioCaptureCallback) -> Self { + Self { + interleaved_buffer: Vec::new(), + callback, + } + } +} + +#[wasm_bindgen] +impl WasmAudioCaptureProcessor { + /// Ensures the capture buffer can hold `channels * frame_size` samples and returns a pointer + /// for JS to interleave captured audio into. + pub fn capture_buffer_ptr(&mut self, channels: u32, frame_size: u32) -> u32 { + let interleaved_buffer_size = channels as usize * frame_size as usize; + self.interleaved_buffer.resize( + interleaved_buffer_size.max(self.interleaved_buffer.len()), + f32::EQUILIBRIUM, + ); + self.interleaved_buffer.as_mut_ptr() as _ + } + + /// Invokes the Rust callback with the interleaved audio JS wrote via the pointer returned by + /// [`capture_buffer_ptr`](Self::capture_buffer_ptr). + pub fn process_captured( + &mut self, + channels: u32, + frame_size: u32, + sample_rate: u32, + current_time: f64, + ) { + let interleaved_buffer_size = channels as usize * frame_size as usize; + (self.callback)( + &self.interleaved_buffer[..interleaved_buffer_size], + frame_size, + sample_rate, + current_time, + ); + } + + pub fn pack(self) -> usize { + Box::into_raw(Box::new(self)) as usize + } + /// # Safety + /// + /// The `val` parameter must be a value previously returned by `Self::pack`. + /// It must not have already been unpacked or deallocated, and must not be used after this call. + /// Using an invalid or already-consumed pointer will result in undefined behavior. + pub unsafe fn unpack(val: usize) -> Self { + unsafe { *Box::from_raw(val as *mut _) } + } +} + /// Drives a `setInterval` that refreshes the shared output-latency value. struct LatencyPoller { window: web_sys::Window, diff --git a/src/host/audioworklet/worklet.js b/src/host/audioworklet/worklet.js index f395dd191..583181e6e 100644 --- a/src/host/audioworklet/worklet.js +++ b/src/host/audioworklet/worklet.js @@ -24,7 +24,7 @@ registerProcessor("CpalProcessor", class WasmProcessor extends AudioWorkletProce currentTime ); - const interleaved_start = interleaved_ptr / 4; // Convert byte offset to f32 index + const interleaved_start = interleaved_ptr / Float32Array.BYTES_PER_ELEMENT; const interleaved = this.wasm_memory; const total_samples = frame_size * channels_count; @@ -46,4 +46,54 @@ registerProcessor("CpalProcessor", class WasmProcessor extends AudioWorkletProce return true; } -}); \ No newline at end of file +}); + +registerProcessor("CpalCaptureProcessor", class WasmCaptureProcessor extends AudioWorkletProcessor { + constructor(options) { + super(); + let [module, memory, handle] = options.processorOptions; + bindgen.initSync({ module, memory }); + this.processor = bindgen.WasmAudioCaptureProcessor.unpack(handle); + this.memory = memory; + this.wasm_memory = new Float32Array(memory.buffer); + } + + process(inputs) { + const channels = inputs[0]; + const channels_count = channels.length; + if (channels_count === 0) { + // No source connected to the node yet. + return true; + } + const frame_size = channels[0].length; + + const interleaved_ptr = this.processor.capture_buffer_ptr(channels_count, frame_size); + + // capture_buffer_ptr() may have grown Wasm memory; refresh the view before writing. + if (this.wasm_memory.buffer !== this.memory.buffer) { + this.wasm_memory = new Float32Array(this.memory.buffer); + } + + const interleaved_start = interleaved_ptr / Float32Array.BYTES_PER_ELEMENT; + const total_samples = frame_size * channels_count; + if (interleaved_start + total_samples > this.wasm_memory.length) { + console.error("CpalCaptureProcessor: Audio buffer out of bounds! Ptr:", interleaved_ptr, "Len:", total_samples); + return false; // Safely stop the node + } + + // Interleave: read sequential from the input channels, write strided into Wasm + for (let ch = 0; ch < channels_count; ch++) { + const channel = channels[ch]; + let write_pos = interleaved_start + ch; + + for (let i = 0; i < frame_size; i++) { + this.wasm_memory[write_pos] = channel[i]; + write_pos += channels_count; + } + } + + this.processor.process_captured(channels_count, frame_size, sampleRate, currentTime); + + return true; + } +}); diff --git a/src/host/mod.rs b/src/host/mod.rs index 2d901d8d8..b109acb59 100644 --- a/src/host/mod.rs +++ b/src/host/mod.rs @@ -306,3 +306,87 @@ where data_callback(data, &info); } } + +/// Maps a rejected `getUserMedia()` promise to a [`crate::Error`], based on the DOMException +/// `name` the browser rejects with. +/// +/// +#[cfg(all( + target_arch = "wasm32", + target_os = "unknown", + feature = "wasm-bindgen" +))] +pub(crate) fn get_user_media_error(js_err: &wasm_bindgen::JsValue) -> crate::Error { + use crate::{Error, ErrorKind}; + + let name = js_sys::Reflect::get(js_err, &wasm_bindgen::JsValue::from_str("name")) + .ok() + .and_then(|v| v.as_string()); + let message = js_sys::Reflect::get(js_err, &wasm_bindgen::JsValue::from_str("message")) + .ok() + .and_then(|v| v.as_string()) + .filter(|s| !s.is_empty()) + .or_else(|| name.clone()) + .unwrap_or_else(|| "unknown error".to_string()); + + let kind = match name.as_deref() { + Some("NotAllowedError" | "SecurityError") => ErrorKind::PermissionDenied, + Some("NotFoundError") => ErrorKind::DeviceNotAvailable, + Some("OverconstrainedError") => ErrorKind::UnsupportedConfig, + Some("NotReadableError") => ErrorKind::DeviceBusy, + Some("TypeError") => ErrorKind::InvalidInput, + _ => ErrorKind::BackendError, + }; + + Error::with_message(kind, format!("getUserMedia() failed: {message}")) +} + +/// Requests microphone access via `getUserMedia()`. +/// +/// +#[cfg(all( + target_arch = "wasm32", + target_os = "unknown", + feature = "wasm-bindgen" +))] +pub(crate) async fn request_microphone() -> Result { + use wasm_bindgen::JsCast; + + let constraints = web_sys::MediaStreamConstraints::new(); + constraints.set_audio_bool(true); + + let window = + web_sys::window().ok_or_else(|| wasm_bindgen::JsValue::from_str("No window available"))?; + let media_devices = window.navigator().media_devices()?; + let promise = media_devices.get_user_media_with_constraints(&constraints)?; + let media_stream = wasm_bindgen_futures::JsFuture::from(promise).await?; + Ok(media_stream.unchecked_into::()) +} + +/// Whether the current context can request microphone access. There is no way to know if a +/// microphone is actually present without asking for permission first via getUserMedia(). +#[cfg(all( + target_arch = "wasm32", + target_os = "unknown", + feature = "wasm-bindgen" +))] +pub(crate) fn is_get_user_media_available() -> bool { + web_sys::window().is_some_and(|w| w.navigator().media_devices().is_ok()) +} + +/// Stops every audio track of `media_stream`, releasing the microphone and turning off the +/// browser's capture indicator. Dropping a WebAudio graph alone does not do this. +#[cfg(all( + target_arch = "wasm32", + target_os = "unknown", + feature = "wasm-bindgen" +))] +pub(crate) fn stop_tracks(media_stream: &web_sys::MediaStream) { + use wasm_bindgen::JsCast; + + for track in media_stream.get_audio_tracks().iter() { + if let Ok(track) = track.dyn_into::() { + track.stop(); + } + } +} diff --git a/src/host/webaudio/mod.rs b/src/host/webaudio/mod.rs index db9da96bf..6b2f3a299 100644 --- a/src/host/webaudio/mod.rs +++ b/src/host/webaudio/mod.rs @@ -16,7 +16,7 @@ use std::{ }; #[cfg(not(target_feature = "atomics"))] -use std::sync::atomic::AtomicBool; +use std::{cell::Cell, rc::Rc, sync::atomic::AtomicBool}; #[cfg(target_feature = "atomics")] use futures_channel::mpsc; @@ -24,10 +24,25 @@ use futures_channel::mpsc; use futures_util::StreamExt as _; type OutputDataCallbackArc = Arc>; +type InputDataCallbackArc = Arc>; +type CaptureGraphResult = Result< + ( + MediaStreamAudioSourceNode, + ScriptProcessorNode, + GainNode, + Closure, + ), + JsValue, +>; +#[cfg(not(target_feature = "atomics"))] +use self::web_sys::MediaStream; use self::{ wasm_bindgen::{JsCast, prelude::*}, - web_sys::{AudioContext, AudioContextOptions}, + web_sys::{ + AudioContext, AudioContextOptions, AudioProcessingEvent, GainNode, + MediaStreamAudioSourceNode, ScriptProcessorNode, + }, }; use crate::{ BufferSize, CallbackInfo, ChannelCount, Data, DeviceDescription, DeviceDescriptionBuilder, @@ -74,6 +89,10 @@ pub struct Stream { on_ended_closures: Vec, #[cfg(not(target_feature = "atomics"))] is_started: Arc, + // Populated asynchronously once getUserMedia() resolves; empty for output streams. Never + // read: held only so Drop-ing the Stream drops CaptureHandles, which stops the mic tracks. + #[cfg(not(target_feature = "atomics"))] + _capture: Rc>>, // Multi-threaded WASM (+atomics): all fields are Send+Sync; JS types are owned by a // spawn_local future on the local thread and are never stored here. @@ -90,6 +109,27 @@ unsafe impl Send for Stream {} unsafe impl Sync for Stream {} // With atomics, all Stream fields auto-derive Send+Sync. +/// Keeps the getUserMedia() capture graph alive for the lifetime of an input [`Stream`]. +/// Stopping the underlying tracks on drop releases the microphone and turns off the browser's +/// capture indicator; merely disconnecting the WebAudio graph does not. +#[cfg(not(target_feature = "atomics"))] +struct CaptureHandles { + media_stream: MediaStream, + _source: MediaStreamAudioSourceNode, + _processor: ScriptProcessorNode, + _mute_gain: GainNode, + _on_audio_process: Closure, +} + +#[cfg(not(target_feature = "atomics"))] +impl Drop for CaptureHandles { + fn drop(&mut self) { + stop_tracks(&self.media_stream); + } +} + +use crate::host::{is_get_user_media_available, request_microphone, stop_tracks}; + pub use crate::iter::{SupportedInputConfigs, SupportedOutputConfigs}; // https://webaudio.github.io/web-audio-api/#dom-baseaudiocontext-createbuffer @@ -109,6 +149,12 @@ const DEFAULT_BUFFER_SIZE: usize = 2048; // https://html.spec.whatwg.org/multipage/timers-and-user-prompts.html#timers const INITIAL_TIMEOUT_MS: i32 = 4; +// https://webaudio.github.io/web-audio-api/#dom-baseaudiocontext-createscriptprocessor +// createScriptProcessor() only accepts 0 (browser picks) or a power of two in this range. +const SCRIPT_PROCESSOR_VALID_BUFFER_SIZES: [usize; 7] = [256, 512, 1024, 2048, 4096, 8192, 16384]; +const SCRIPT_PROCESSOR_MIN_BUFFER_SIZE: usize = 256; +const SCRIPT_PROCESSOR_MAX_BUFFER_SIZE: usize = 16384; + impl Host { pub fn new() -> Result { if Self::is_available() { @@ -151,8 +197,13 @@ impl Devices { impl Device { fn description(&self) -> Result { + let direction = if is_get_user_media_available() { + DeviceDirection::Duplex + } else { + DeviceDirection::Output + }; Ok(DeviceDescriptionBuilder::new("Default Device") - .direction(DeviceDirection::Output) + .direction(direction) .build()) } @@ -161,8 +212,15 @@ impl Device { } fn supported_input_configs(&self) -> Result { - // TODO - Ok(Vec::new().into_iter()) + // The actual channel count and sample rate depend on the microphone getUserMedia() + // grants access to, which isn't known ahead of time; WebAudio resamples and up/downmixes + // whatever it gets to match, so this reports the same broad matrix as output, just with + // ScriptProcessorNode's narrower buffer-size range. + let buffer_size = SupportedBufferSize::Range { + min: SCRIPT_PROCESSOR_MIN_BUFFER_SIZE as FrameCount, + max: SCRIPT_PROCESSOR_MAX_BUFFER_SIZE as FrameCount, + }; + Ok(supported_configs(buffer_size).into_iter()) } fn supported_output_configs(&self) -> Result { @@ -170,46 +228,21 @@ impl Device { min: 1, max: FrameCount::MAX, }; - let configs: Vec<_> = (MIN_CHANNELS..=MAX_CHANNELS) - .flat_map(|channels| { - crate::COMMON_SAMPLE_RATES - .iter() - .copied() - .filter(|&r| (MIN_SAMPLE_RATE..=MAX_SAMPLE_RATE).contains(&r)) - .map(move |rate| SupportedStreamConfigRange { - channels, - min_sample_rate: rate, - max_sample_rate: rate, - buffer_size, - sample_format: SUPPORTED_SAMPLE_FORMAT, - }) - }) - .collect(); - Ok(configs.into_iter()) + Ok(supported_configs(buffer_size).into_iter()) } fn default_input_config(&self) -> Result { - Err(Error::with_message( - ErrorKind::UnsupportedOperation, - "Device does not support input", - )) + default_config( + self.supported_input_configs()?, + "No supported input configuration", + ) } fn default_output_config(&self) -> Result { - let range = self - .supported_output_configs()? - .max_by(|a, b| a.cmp_default_heuristics(b)) - .ok_or_else(|| { - Error::with_message( - ErrorKind::UnsupportedConfig, - "No supported output configuration", - ) - })?; - let config = range - .try_with_standard_sample_rate() - .unwrap_or_else(|| range.with_max_sample_rate()); - - Ok(config) + default_config( + self.supported_output_configs()?, + "No supported output configuration", + ) } } @@ -242,22 +275,280 @@ impl DeviceTrait for Device { Self::default_output_config(self) } + /// Create an input stream capturing microphone audio via `getUserMedia()`. + /// + /// # Async completion + /// + /// This function returns `Ok` synchronously once the [`AudioContext`] is created, before the + /// browser has granted or denied microphone access. Permission is requested asynchronously via + /// [`wasm_bindgen_futures::spawn_local`]; if the user denies access, no microphone is present, + /// or the browser rejects the capture graph, the error is delivered to `error_callback` after + /// the caller already holds a [`Stream`]. There is no way to surface such errors synchronously + /// given `getUserMedia()`'s async, permission-gated design. + /// + /// [`start`](crate::traits::StreamTrait::start) and [`pause`](crate::traits::StreamTrait::pause) + /// calls made before permission is granted return `Ok` immediately: they resume/suspend the + /// already-created [`AudioContext`], and the capture graph starts delivering callbacks as soon + /// as it connects, in step with whatever run state the context is already in. fn build_input_stream_raw( &self, - _config: StreamConfig, - _sample_format: SampleFormat, - _data_callback: D, - _error_callback: E, + config: StreamConfig, + sample_format: SampleFormat, + data_callback: D, + error_callback: E, _timeout: Option, ) -> Result where D: FnMut(&Data, &CallbackInfo) + Send + 'static, E: FnMut(Error) + Send + 'static, { - Err(Error::with_message( - ErrorKind::UnsupportedOperation, - "Device does not support input", - )) + validate_config(&config, sample_format)?; + + let n_channels = config.channels as usize; + + let buffer_size_frames = match config.buffer_size { + BufferSize::Fixed(v) => v as usize, + BufferSize::Default => DEFAULT_BUFFER_SIZE, + }; + if !SCRIPT_PROCESSOR_VALID_BUFFER_SIZES.contains(&buffer_size_frames) { + return Err(Error::with_message( + ErrorKind::UnsupportedConfig, + format!( + "Buffer size {buffer_size_frames} is not supported; must be one of {SCRIPT_PROCESSOR_VALID_BUFFER_SIZES:?}" + ), + )); + } + let buffer_duration_secs = buffer_time_step_secs(buffer_size_frames, config.sample_rate); + + let data_callback = crate::host::monotonic_input_callback(data_callback); + let data_callback: InputDataCallbackArc = Arc::new(Mutex::new(data_callback)); + let error_callback: ErrorCallbackArc = Arc::new(Mutex::new(error_callback)); + + let stream_opts = AudioContextOptions::new(); + stream_opts.set_sample_rate(config.sample_rate as f32); + let ctx = AudioContext::new_with_context_options(&stream_opts).map_err(|_| { + Error::with_message( + ErrorKind::UnsupportedConfig, + "Failed to create audio context", + ) + })?; + + // SAFETY: see the SAFETY note in `build_output_stream_raw` above; the same single-thread + // (or task-confined) reasoning applies here. + #[allow(clippy::arc_with_non_send_sync)] + let ctx = Arc::new(ctx); + let ctx_task = ctx.clone(); + + #[cfg(not(target_feature = "atomics"))] + let is_started = Arc::new(AtomicBool::new(false)); + #[cfg(not(target_feature = "atomics"))] + let capture: Rc>> = Rc::new(Cell::new(None)); + #[cfg(not(target_feature = "atomics"))] + let capture_weak = Rc::downgrade(&capture); + + #[cfg(target_feature = "atomics")] + let current_time_bits = Arc::new(AtomicU64::new(ctx.current_time().to_bits())); + #[cfg(target_feature = "atomics")] + let current_time_bits_stream = current_time_bits.clone(); + #[cfg(target_feature = "atomics")] + let (command_tx, mut command_rx) = mpsc::unbounded::(); + + wasm_bindgen_futures::spawn_local(async move { + let media_stream = match request_microphone().await { + Ok(stream) => stream, + Err(js_err) => { + error_callback.lock().unwrap_or_else(|e| e.into_inner())( + crate::host::get_user_media_error(&js_err), + ); + return; + } + }; + + #[cfg(not(target_feature = "atomics"))] + let capture = match capture_weak.upgrade() { + Some(capture) => capture, + None => { + // The Stream was dropped while permission was pending; release the + // microphone instead of leaving a live capture session behind. + stop_tracks(&media_stream); + return; + } + }; + + let build_graph = || -> CaptureGraphResult { + let source = ctx_task.create_media_stream_source(&media_stream)?; + let processor = ctx_task + .create_script_processor_with_buffer_size_and_number_of_input_channels_and_number_of_output_channels( + buffer_size_frames as u32, + n_channels as u32, + // A processor with 0 output channels never fires `onaudioprocess` in + // some browsers, so it is given a single silent output instead. + 1, + )?; + let mute_gain = GainNode::new(&ctx_task)?; + mute_gain.gain().set_value(0.0); + source.connect_with_audio_node(&processor)?; + processor.connect_with_audio_node(&mute_gain)?; + mute_gain.connect_with_audio_node(&ctx_task.destination())?; + + let mut temporary_buffer = vec![f32::EQUILIBRIUM; n_channels * buffer_size_frames]; + let mut temporary_channel_buffer = vec![f32::EQUILIBRIUM; buffer_size_frames]; + let ctx_cb = ctx_task.clone(); + let data_callback_cb = data_callback.clone(); + let error_callback_cb = error_callback.clone(); + #[cfg(target_feature = "atomics")] + let current_time_bits_cb = current_time_bits.clone(); + + let on_audio_process = Closure::wrap(Box::new(move |event: AudioProcessingEvent| { + let now = ctx_cb.current_time(); + #[cfg(target_feature = "atomics")] + current_time_bits_cb.store(now.to_bits(), Ordering::Relaxed); + + let input_buffer = match event.input_buffer() { + Ok(b) => b, + Err(_) => { + (error_callback_cb.lock().unwrap_or_else(|e| e.into_inner()))( + Error::with_message( + ErrorKind::StreamInvalidated, + "Failed to read captured audio", + ), + ); + return; + } + }; + + // Deinterleave from the browser's per-channel buffers into our interleaved + // scratch buffer; the mirror image of the deinterleave loop in + // `build_output_stream_raw`. + for channel in 0..n_channels { + if input_buffer + .copy_from_channel(&mut temporary_channel_buffer, channel as i32) + .is_err() + { + (error_callback_cb.lock().unwrap_or_else(|e| e.into_inner()))( + Error::with_message( + ErrorKind::StreamInvalidated, + "Failed to copy captured audio", + ), + ); + return; + } + for i in 0..buffer_size_frames { + temporary_buffer[n_channels * i + channel] = + temporary_channel_buffer[i]; + } + } + + let data = temporary_buffer.as_mut_ptr() as *mut (); + let data = + unsafe { Data::from_parts(data, temporary_buffer.len(), sample_format) }; + + let callback = StreamInstant::from_secs_f64(now); + let device = + StreamInstant::from_secs_f64((now - buffer_duration_secs).max(0.0)); + let info = CallbackInfo { + timestamp: StreamTimestamp { callback, device }, + xrun: false, + }; + + match data_callback_cb.lock() { + Ok(mut data_callback) => (data_callback.deref_mut())(&data, &info), + Err(_) => (error_callback_cb.lock().unwrap_or_else(|e| e.into_inner()))( + Error::with_message( + ErrorKind::StreamInvalidated, + "Stream lock poisoned", + ), + ), + } + }) + as Box); + + processor.set_onaudioprocess(Some(on_audio_process.as_ref().unchecked_ref())); + + Ok((source, processor, mute_gain, on_audio_process)) + }; + + let (source, processor, mute_gain, on_audio_process) = match build_graph() { + Ok(handles) => handles, + Err(js_err) => { + stop_tracks(&media_stream); + error_callback.lock().unwrap_or_else(|e| e.into_inner())(Error::with_message( + ErrorKind::UnsupportedConfig, + format!("Failed to initialize capture graph: {js_err:?}"), + )); + return; + } + }; + + #[cfg(not(target_feature = "atomics"))] + { + capture.set(Some(CaptureHandles { + media_stream, + _source: source, + _processor: processor, + _mute_gain: mute_gain, + _on_audio_process: on_audio_process, + })); + } + + #[cfg(target_feature = "atomics")] + { + let _source = source; + let _processor = processor; + let _mute_gain = mute_gain; + let _on_audio_process = on_audio_process; + // Process play/pause commands from any thread until Stream is dropped. + // Dropping Stream closes command_tx, which terminates this loop. + while let Some(cmd) = command_rx.next().await { + match cmd { + Command::Play => { + if ctx_task.resume().is_err() { + error_callback.lock().unwrap_or_else(|e| e.into_inner())( + Error::with_message( + ErrorKind::DeviceNotAvailable, + "Failed to resume audio context", + ), + ); + } + } + Command::Pause => { + if ctx_task.suspend().is_err() { + error_callback.lock().unwrap_or_else(|e| e.into_inner())( + Error::with_message( + ErrorKind::DeviceNotAvailable, + "Failed to suspend audio context", + ), + ); + } + } + } + } + // Stream dropped: release the microphone and close the AudioContext. + stop_tracks(&media_stream); + let _ = ctx_task.close(); + } + }); + + #[cfg(not(target_feature = "atomics"))] + { + Ok(Self::Stream { + ctx, + on_ended_closures: Vec::new(), + config, + buffer_size_frames, + is_started, + _capture: capture, + }) + } + + #[cfg(target_feature = "atomics")] + { + Ok(Self::Stream { + command_tx, + current_time_bits: current_time_bits_stream, + buffer_size_frames, + }) + } } /// Create an output stream. @@ -273,33 +564,7 @@ impl DeviceTrait for Device { D: FnMut(&mut Data, &CallbackInfo) + Send + 'static, E: FnMut(Error) + Send + 'static, { - crate::validate_stream_config(&config)?; - if config.channels > MAX_CHANNELS { - return Err(Error::with_message( - ErrorKind::UnsupportedConfig, - format!( - "Channel count {} exceeds the maximum of {MAX_CHANNELS}", - config.channels - ), - )); - } - if sample_format != SUPPORTED_SAMPLE_FORMAT { - return Err(Error::with_message( - ErrorKind::UnsupportedConfig, - format!( - "Sample format {sample_format} is not supported; required format is {SUPPORTED_SAMPLE_FORMAT}" - ), - )); - } - if !(MIN_SAMPLE_RATE..=MAX_SAMPLE_RATE).contains(&config.sample_rate) { - return Err(Error::with_message( - ErrorKind::UnsupportedConfig, - format!( - "Sample rate {} Hz is not in the supported range {MIN_SAMPLE_RATE}..={MAX_SAMPLE_RATE} Hz", - config.sample_rate - ), - )); - } + validate_config(&config, sample_format)?; let n_channels = config.channels as usize; @@ -628,6 +893,7 @@ impl DeviceTrait for Device { config, buffer_size_frames, is_started, + _capture: Rc::new(Cell::new(None)), }) } @@ -790,9 +1056,78 @@ impl Iterator for Devices { } } +/// Checks shared by both `build_input_stream_raw` and `build_output_stream_raw`. Buffer-size +/// validation differs between the two (`ScriptProcessorNode`'s discrete sizes only apply to +/// input) and stays in each caller. +fn validate_config(config: &StreamConfig, sample_format: SampleFormat) -> Result<(), Error> { + crate::validate_stream_config(config)?; + if config.channels > MAX_CHANNELS { + return Err(Error::with_message( + ErrorKind::UnsupportedConfig, + format!( + "Channel count {} exceeds the maximum of {MAX_CHANNELS}", + config.channels + ), + )); + } + if sample_format != SUPPORTED_SAMPLE_FORMAT { + return Err(Error::with_message( + ErrorKind::UnsupportedConfig, + format!( + "Sample format {sample_format} is not supported; required format is {SUPPORTED_SAMPLE_FORMAT}" + ), + )); + } + if !(MIN_SAMPLE_RATE..=MAX_SAMPLE_RATE).contains(&config.sample_rate) { + return Err(Error::with_message( + ErrorKind::UnsupportedConfig, + format!( + "Sample rate {} Hz is not in the supported range {MIN_SAMPLE_RATE}..={MAX_SAMPLE_RATE} Hz", + config.sample_rate + ), + )); + } + Ok(()) +} + +/// The full matrix of channel counts x sample rates for a given buffer-size range. +fn supported_configs(buffer_size: SupportedBufferSize) -> Vec { + (MIN_CHANNELS..=MAX_CHANNELS) + .flat_map(|channels| { + crate::COMMON_SAMPLE_RATES + .iter() + .copied() + .filter(|&r| (MIN_SAMPLE_RATE..=MAX_SAMPLE_RATE).contains(&r)) + .map(move |rate| SupportedStreamConfigRange { + channels, + min_sample_rate: rate, + max_sample_rate: rate, + buffer_size, + sample_format: SUPPORTED_SAMPLE_FORMAT, + }) + }) + .collect() +} + +/// Picks the best default from a supported-config iterator via the standard heuristics. +fn default_config( + configs: impl Iterator, + none_supported_message: &'static str, +) -> Result { + let range = configs + .max_by(|a, b| a.cmp_default_heuristics(b)) + .ok_or_else(|| Error::with_message(ErrorKind::UnsupportedConfig, none_supported_message))?; + Ok(range + .try_with_standard_sample_rate() + .unwrap_or_else(|| range.with_max_sample_rate())) +} + fn default_input_device() -> Option { - // TODO - None + if Host::is_available() && is_get_user_media_available() { + Some(Device) + } else { + None + } } fn default_output_device() -> Option {