diff --git a/.github/workflows/platforms.yml b/.github/workflows/platforms.yml
index 8d904683c..1c8317ab0 100644
--- a/.github/workflows/platforms.yml
+++ b/.github/workflows/platforms.yml
@@ -476,8 +476,8 @@ jobs:
- name: Check all features (no atomics)
run: cargo +${{ env.MSRV_WASM }} check --workspace --all-features --verbose --target ${{ env.TARGET }}
- - name: Build wasm-beep example
- working-directory: ./examples/wasm-beep
+ - name: Build webaudio example
+ working-directory: ./examples/webaudio
run: trunk build
env:
RUSTUP_TOOLCHAIN: ${{ env.MSRV_WASM }}
@@ -516,8 +516,8 @@ jobs:
- name: Check all features
run: cargo +nightly check --workspace --all-features --verbose -Z build-std=std,panic_abort --target ${{ env.TARGET }}
- - name: Build audioworklet-beep example
- working-directory: ./examples/audioworklet-beep
+ - name: Build audioworklet example
+ working-directory: ./examples/audioworklet
run: trunk build
env:
RUSTUP_TOOLCHAIN: nightly
diff --git a/CHANGELOG.md b/CHANGELOG.md
index b163261e7..827f892fe 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,8 +11,9 @@ 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.
+- `DeviceTrait::build_duplex_stream()`, `build_duplex_stream_raw()`, and `supports_duplex()` for capture and playback from one device-level callback.
+- **AudioWorklet**: Input and duplex streams are now supported.
+- **WebAudio**: Input and duplex streams are now supported.
### Changed
@@ -22,6 +23,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- `StreamTrait::play` is renamed to `start`.
- `InputCallbackInfo`/`OutputCallbackInfo` merged into `CallbackInfo`.
- `InputStreamTimestamp`/`OutputStreamTimestamp` merged into `StreamTimestamp`; `capture`/`playback` renamed `device`.
+- Renamed the `wasm-beep` and `audioworklet-beep` examples to `webaudio` and `audioworklet`.
- **ALSA**: Update `alsa` dependency to 0.12.
- **Linux**: `realtime` can now promote threads without requiring `realtime-dbus`.
@@ -38,6 +40,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- **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`.
+- **AudioWorklet**: Fix dropouts in output streams when the callback buffer grows.
- **JACK**: Channel enumeration is capped at the physical system port count again.
- **WASAPI**: Device enumeration no longer panics if the COM enumerator fails to initialize.
@@ -115,7 +118,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- `realtime` feature for real-time audio thread scheduling without a D-Bus build dependency.
- `StreamTrait::now()` to query the current instant on the stream's clock.
- `StreamTrait::buffer_size()` to query the stream's current buffer size in frames per callback.
-- `DeviceTrait::build_duplex_stream()`, `build_duplex_stream_raw()`, and `supports_duplex()` for synchronized capture and playback on a shared clock (no backend support yet).
- `SAMPLE_RATE_CD` (44100 Hz) and `SAMPLE_RATE_48K` (48000 Hz) constants.
- `SupportedStreamConfigRange::try_with_standard_sample_rate()` and `with_standard_sample_rate()`
to select 48 kHz or 44.1 kHz from a range.
diff --git a/Cargo.toml b/Cargo.toml
index 2810d6b91..06d86479f 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -236,6 +236,9 @@ name = "beep"
[[example]]
name = "enumerate"
+[[example]]
+name = "duplex"
+
[[example]]
name = "feedback"
diff --git a/README.md b/README.md
index 13ab29a1b..e7bf06599 100644
--- a/README.md
+++ b/README.md
@@ -63,14 +63,14 @@ The `audioworklet` backend additionally requires `-Zbuild-std` with atomics supp
| Feature | Platform | Description |
| ------- | -------- | ----------- |
| `asio` | Windows | ASIO backend for low-latency audio, bypassing the Windows audio stack. Requires ASIO drivers and LLVM/Clang. See the [ASIO setup guide](#compiling-for-asio). |
-| `audioworklet` | WebAssembly (`wasm32-unknown-unknown`) | Audio Worklet backend for lower-latency web audio than the default Web Audio API, running audio on a dedicated thread. Requires atomics support (`RUSTFLAGS="-C target-feature=+atomics,+bulk-memory,+mutable-globals"`) and `Cross-Origin` headers for `SharedArrayBuffer`. See the `audioworklet-beep` example. |
+| `audioworklet` | WebAssembly (`wasm32-unknown-unknown`) | Audio Worklet backend for lower-latency web audio than the default Web Audio API, running audio on a dedicated thread. Requires atomics support (`RUSTFLAGS="-C target-feature=+atomics,+bulk-memory,+mutable-globals"`) and `Cross-Origin` headers for `SharedArrayBuffer`. See the `audioworklet` example. |
| `custom` | All | User-defined backend implementations for audio systems not natively supported by CPAL. See `examples/custom.rs`. |
| `jack` | Linux, BSD, macOS, Windows | JACK Audio Connection Kit backend for pro-audio routing and inter-application connectivity. Requires `libjack-jackd2-dev` (Debian/Ubuntu) or `jack-devel` (Fedora). |
| `pipewire` | Linux, BSD | PipeWire media server backend. Requires `libpipewire-0.3-dev` (Debian/Ubuntu) or `pipewire-devel` (Fedora). |
| `pulseaudio` | Linux, BSD | PulseAudio sound server backend. Requires `libpulse-dev` (Debian/Ubuntu) or `pulseaudio-libs-devel` (Fedora). |
| `realtime` | Android, Linux, Windows | Raises the audio callback thread to real-time or high-priority scheduling for lower latency. On Linux, requires `CAP_SYS_NICE`, root, or an `rtprio` limit granted via `limits.conf` or systemd, unless `realtime-dbus` is also enabled. |
| `realtime-dbus` | Linux | Uses `rtkit` via D-Bus for RT scheduling on Linux desktop systems. Implies `realtime` on all platforms. Requires `libdbus-1-dev` on Linux. |
-| `wasm-bindgen` | WebAssembly (`wasm32-unknown-unknown`) | Web Audio API backend for browser-based audio; required for any WebAssembly audio support. See the `wasm-beep` example. |
+| `wasm-bindgen` | WebAssembly (`wasm32-unknown-unknown`) | Web Audio API backend for browser-based audio; required for any WebAssembly audio support. See the `webaudio` example. |
See the [beep example](examples/beep.rs) for selecting the backend at runtime.
diff --git a/examples/audioworklet-beep/index.html b/examples/audioworklet-beep/index.html
deleted file mode 100644
index 6e3139fc6..000000000
--- a/examples/audioworklet-beep/index.html
+++ /dev/null
@@ -1,17 +0,0 @@
-
-
-
-
-
- cpal AudioWorklet beep example
-
-
-
-
-
-
-
-
Recording plays your microphone back live. Wear headphones to avoid feedback howl.
-
-
-
\ No newline at end of file
diff --git a/examples/audioworklet-beep/.cargo/config.toml b/examples/audioworklet/.cargo/config.toml
similarity index 100%
rename from examples/audioworklet-beep/.cargo/config.toml
rename to examples/audioworklet/.cargo/config.toml
diff --git a/examples/audioworklet-beep/.gitignore b/examples/audioworklet/.gitignore
similarity index 100%
rename from examples/audioworklet-beep/.gitignore
rename to examples/audioworklet/.gitignore
diff --git a/examples/audioworklet-beep/Cargo.toml b/examples/audioworklet/Cargo.toml
similarity index 91%
rename from examples/audioworklet-beep/Cargo.toml
rename to examples/audioworklet/Cargo.toml
index ebf6bc2ec..b0b57dccc 100644
--- a/examples/audioworklet-beep/Cargo.toml
+++ b/examples/audioworklet/Cargo.toml
@@ -1,6 +1,6 @@
[package]
-name = "audioworklet-beep"
-description = "cpal beep example for WebAssembly on an AudioWorklet"
+name = "audioworklet"
+description = "cpal AudioWorklet example for WebAssembly"
version = "0.2.0"
edition = "2024"
rust-version = "1.85"
diff --git a/examples/audioworklet-beep/README.md b/examples/audioworklet/README.md
similarity index 100%
rename from examples/audioworklet-beep/README.md
rename to examples/audioworklet/README.md
diff --git a/examples/audioworklet-beep/Trunk.toml b/examples/audioworklet/Trunk.toml
similarity index 100%
rename from examples/audioworklet-beep/Trunk.toml
rename to examples/audioworklet/Trunk.toml
diff --git a/examples/audioworklet/index.html b/examples/audioworklet/index.html
new file mode 100644
index 000000000..35346584a
--- /dev/null
+++ b/examples/audioworklet/index.html
@@ -0,0 +1,28 @@
+
+
+
+
+
+ cpal AudioWorklet example
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
Both loopback modes play your microphone back live. Wear headphones to avoid feedback howl.
+
The two-stream mode bridges an independent input and output stream with a ring buffer. The
+ duplex mode runs both directions from one callback on one clock, so it needs no ring buffer
+ and the round trip is audibly shorter.
+
+
+
diff --git a/examples/audioworklet-beep/rust-toolchain.toml b/examples/audioworklet/rust-toolchain.toml
similarity index 100%
rename from examples/audioworklet-beep/rust-toolchain.toml
rename to examples/audioworklet/rust-toolchain.toml
diff --git a/examples/audioworklet-beep/src/lib.rs b/examples/audioworklet/src/lib.rs
similarity index 70%
rename from examples/audioworklet-beep/src/lib.rs
rename to examples/audioworklet/src/lib.rs
index d39979ef0..f03af811e 100644
--- a/examples/audioworklet-beep/src/lib.rs
+++ b/examples/audioworklet/src/lib.rs
@@ -1,8 +1,8 @@
use std::{cell::Cell, rc::Rc};
use cpal::{
- Device, Error, ErrorKind, FromSample, HostId, Sample, SampleFormat, SizedSample, Stream,
- StreamConfig,
+ Device, DuplexCallbackInfo, Error, ErrorKind, FromSample, HostId, Sample, SampleFormat,
+ SizedSample, Stream, StreamConfig,
traits::{DeviceTrait, HostTrait, StreamTrait},
};
use ringbuf::{
@@ -25,6 +25,8 @@ pub fn main_js() -> Result<(), JsValue> {
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();
+ let duplex_button = document.get_element_by_id("duplex").unwrap();
+ let stop_duplex_button = document.get_element_by_id("stop-duplex").unwrap();
// stream needs to be referenced from the "play" and "stop" closures
let stream = Rc::new(Cell::new(None));
@@ -76,6 +78,31 @@ pub fn main_js() -> Result<(), JsValue> {
closure.forget();
}
+ // duplex loopback is a single stream, so one slot holds the whole thing
+ let duplex_stream = Rc::new(Cell::new(None));
+
+ // set up duplex button
+ {
+ let duplex_stream = duplex_stream.clone();
+ let closure = Closure::::new(move |_event: web_sys::MouseEvent| {
+ duplex_stream.set(Some(duplex()));
+ });
+ duplex_button
+ .add_event_listener_with_callback("mousedown", closure.as_ref().unchecked_ref())?;
+ closure.forget();
+ }
+
+ // set up stop-duplex button
+ {
+ let closure = Closure::::new(move |_event: web_sys::MouseEvent| {
+ // stop the stream by dropping it; releases the microphone
+ duplex_stream.take();
+ });
+ stop_duplex_button
+ .add_event_listener_with_callback("mousedown", closure.as_ref().unchecked_ref())?;
+ closure.forget();
+ }
+
Ok(())
}
@@ -166,6 +193,56 @@ fn record() -> (Stream, Stream) {
(input_stream, output_stream)
}
+/// The same live microphone loopback as [`record`], but as one duplex stream instead of two
+/// independent ones. `AudioWorkletProcessor.process(inputs, outputs)` hands both directions to a
+/// single callback on a single clock, so no ring buffer is needed to bridge them and the round
+/// trip is a callback rather than a delay line. Wear headphones: routing a live mic to speakers
+/// risks feedback howl.
+fn duplex() -> Stream {
+ let host = cpal::host_from_id(HostId::AudioWorklet).expect("AudioWorklet host not available");
+
+ let device = host
+ .default_output_device()
+ .expect("failed to find a default output device");
+ assert!(
+ device.supports_duplex(),
+ "duplex streams need `navigator.mediaDevices`, which browsers expose only in a \
+ secure context: serve this page over HTTPS or from localhost"
+ );
+
+ let config = device.default_duplex_config().unwrap();
+
+ 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()),
+ };
+
+ // WebAudio processes exclusively in f32, so there is no sample format to match on here.
+ let input_channels = config.input_channels as usize;
+ let output_channels = config.output_channels as usize;
+ let stream = device
+ .build_duplex_stream(
+ config,
+ move |input: &[f32], output: &mut [f32], _: &DuplexCallbackInfo| {
+ // The two directions can carry different channel counts, so mix each captured
+ // frame down to mono and fan it back out across the output frame.
+ for (captured, rendered) in input
+ .chunks(input_channels)
+ .zip(output.chunks_mut(output_channels))
+ {
+ rendered.fill(captured.iter().sum::() / input_channels as f32);
+ }
+ },
+ err_fn,
+ None,
+ )
+ .unwrap();
+ stream.start().unwrap();
+ stream
+}
+
fn build_input(device: &Device, config: StreamConfig, mut producer: HeapProd) -> Stream
where
T: Sample + SizedSample,
diff --git a/examples/duplex.rs b/examples/duplex.rs
new file mode 100644
index 000000000..eab833941
--- /dev/null
+++ b/examples/duplex.rs
@@ -0,0 +1,187 @@
+//! Feeds the input stream directly back into the output stream, from a single duplex callback.
+//!
+//! A duplex stream drives both directions from one device callback on one clock, so neither the
+//! ring buffer nor the delay is needed. It requires a device that [`DeviceTrait::supports_duplex`].
+//!
+//! Where a platform exposes no natively duplex device, you may be able to compose one on the
+//! system: using an Aggregate Device on macOS (with Audio MIDI Setup), or an `asym` PCM in
+//! `~/.asoundrc` on ALSA.
+//!
+//! For simplicity this example requires both directions to use the same sample format, though
+//! [`DeviceTrait::build_duplex_stream`] does not.
+
+use clap::Parser;
+use cpal::{
+ Device, DuplexCallbackInfo, DuplexStreamConfig, Error, ErrorKind, HostId, SampleFormat,
+ SizedSample,
+ traits::{DeviceTrait, HostTrait, StreamTrait},
+};
+
+#[derive(Parser, Debug)]
+#[command(version, about = "CPAL duplex example", long_about = None)]
+struct Opt {
+ /// The duplex audio device to use. Defaults to the first device reporting duplex support.
+ #[arg(short, long, value_name = "DEVICE")]
+ device: Option,
+
+ /// Use the JACK host. Requires `--features jack`.
+ #[arg(long, default_value_t = false)]
+ jack: bool,
+
+ /// Use the PipeWire host. Requires `--features pipewire`.
+ #[arg(long, default_value_t = false)]
+ pipewire: bool,
+
+ /// Use the ASIO host. Requires `--features asio`.
+ #[arg(long, default_value_t = false)]
+ asio: bool,
+}
+
+fn main() -> anyhow::Result<()> {
+ let opt = Opt::parse();
+
+ // JACK/PipeWire/ASIO support must be enabled at compile time, and is
+ // only available on some platforms.
+ #[allow(unused_mut, unused_assignments)]
+ let mut jack_host_id: Result = Err(ErrorKind::HostUnavailable.into());
+ #[allow(unused_mut, unused_assignments)]
+ let mut pipewire_host_id: Result = Err(ErrorKind::HostUnavailable.into());
+ #[allow(unused_mut, unused_assignments)]
+ let mut asio_host_id: Result = Err(ErrorKind::HostUnavailable.into());
+
+ #[cfg(any(
+ target_os = "linux",
+ target_os = "dragonfly",
+ target_os = "freebsd",
+ target_os = "netbsd"
+ ))]
+ {
+ #[cfg(feature = "jack")]
+ {
+ jack_host_id = Ok(HostId::Jack);
+ }
+
+ #[cfg(feature = "pipewire")]
+ {
+ pipewire_host_id = Ok(HostId::PipeWire);
+ }
+ }
+
+ #[cfg(target_os = "windows")]
+ {
+ #[cfg(feature = "asio")]
+ {
+ asio_host_id = Ok(HostId::Asio);
+ }
+ }
+
+ // Manually check for flags. Can be passed through cargo with -- e.g.
+ // cargo run --release --example duplex --features jack -- --jack
+ let host = if opt.jack {
+ jack_host_id
+ .and_then(cpal::host_from_id)
+ .expect("make sure `--features jack` is specified, and the platform is supported")
+ } else if opt.pipewire {
+ pipewire_host_id
+ .and_then(cpal::host_from_id)
+ .expect("make sure `--features pipewire` is specified, and the platform is supported")
+ } else if opt.asio {
+ asio_host_id
+ .and_then(cpal::host_from_id)
+ .expect("make sure `--features asio` is specified, and the platform is supported")
+ } else {
+ cpal::default_host()
+ };
+
+ let device = if let Some(device) = opt.device {
+ let id = &device.parse().expect("failed to parse device id");
+ host.device_by_id(id).expect("failed to find device")
+ } else {
+ host.devices()?
+ .find(|device| device.supports_duplex())
+ .ok_or_else(|| anyhow::anyhow!("no device on this host reports duplex support."))?
+ };
+
+ if !device.supports_duplex() {
+ anyhow::bail!(
+ "device \"{}\" does not support duplex streams",
+ device.id()?
+ );
+ }
+ println!("Using duplex device: \"{}\"", device.id()?);
+
+ let input_config = device.default_input_config()?;
+ let output_config = device.default_output_config()?;
+ assert_eq!(
+ input_config.sample_format(),
+ output_config.sample_format(),
+ "both directions must share a sample format for this example"
+ );
+
+ let config = device.default_duplex_config()?;
+
+ match input_config.sample_format() {
+ SampleFormat::I8 => run::(&device, config),
+ SampleFormat::I16 => run::(&device, config),
+ SampleFormat::I32 => run::(&device, config),
+ SampleFormat::I64 => run::(&device, config),
+ SampleFormat::U8 => run::(&device, config),
+ SampleFormat::U16 => run::(&device, config),
+ SampleFormat::U32 => run::(&device, config),
+ SampleFormat::U64 => run::(&device, config),
+ SampleFormat::F32 => run::(&device, config),
+ SampleFormat::F64 => run::(&device, config),
+ sample_format => panic!("Unsupported sample format '{sample_format}'"),
+ }
+}
+
+fn run(device: &Device, config: DuplexStreamConfig) -> anyhow::Result<()>
+where
+ T: SizedSample + Send + 'static,
+{
+ let input_channels = config.input_channels as usize;
+ let output_channels = config.output_channels as usize;
+
+ println!(
+ "Attempting to build a duplex stream with {} samples and `{config:?}`.",
+ T::FORMAT
+ );
+ let shared_channels = input_channels.min(output_channels);
+ let stream = device.build_duplex_stream(
+ config,
+ move |input: &[T], output: &mut [T], _: &DuplexCallbackInfo| {
+ // Both directions arrive together, so captured audio goes straight out with nothing
+ // buffered in between.
+ //
+ // The channel counts do not need to match. The below maps channels one to one and
+ // silences any that are not shared. In any normal application you would implement
+ // proper mixing and gain compensation.
+ for (captured, rendered) in input
+ .chunks(input_channels)
+ .zip(output.chunks_mut(output_channels))
+ {
+ rendered[..shared_channels].copy_from_slice(&captured[..shared_channels]);
+ rendered[shared_channels..].fill(T::EQUILIBRIUM);
+ }
+ },
+ err_fn,
+ None,
+ )?;
+ println!("Successfully built the stream.");
+
+ stream.start()?;
+ println!("Playing for 10 seconds...");
+ std::thread::sleep(std::time::Duration::from_secs(10));
+ drop(stream);
+ println!("Done!");
+ Ok(())
+}
+
+fn err_fn(err: Error) {
+ match err.kind() {
+ ErrorKind::DeviceChanged | ErrorKind::RealtimeDenied => {
+ eprintln!("{err}")
+ }
+ _ => eprintln!("Stream error: {err}"),
+ }
+}
diff --git a/examples/feedback.rs b/examples/feedback.rs
index b5ee1ce47..b7488ac81 100644
--- a/examples/feedback.rs
+++ b/examples/feedback.rs
@@ -247,7 +247,7 @@ where
output_stream.start()?;
// Run for 10 seconds before closing.
- println!("Playing for 10 seconds... ");
+ println!("Playing for 10 seconds...");
std::thread::sleep(std::time::Duration::from_secs(10));
drop(input_stream);
drop(output_stream);
diff --git a/examples/wasm-beep/index.html b/examples/wasm-beep/index.html
deleted file mode 100644
index 344eecdf3..000000000
--- a/examples/wasm-beep/index.html
+++ /dev/null
@@ -1,14 +0,0 @@
-
-
-
-
- cpal beep example
-
-
-
-
-
-
-
Recording plays your microphone back live. Wear headphones to avoid feedback howl.
-
-
diff --git a/examples/wasm-beep/.gitignore b/examples/webaudio/.gitignore
similarity index 100%
rename from examples/wasm-beep/.gitignore
rename to examples/webaudio/.gitignore
diff --git a/examples/wasm-beep/Cargo.toml b/examples/webaudio/Cargo.toml
similarity index 93%
rename from examples/wasm-beep/Cargo.toml
rename to examples/webaudio/Cargo.toml
index eec5778f7..7739d5dff 100644
--- a/examples/wasm-beep/Cargo.toml
+++ b/examples/webaudio/Cargo.toml
@@ -1,6 +1,6 @@
[package]
-name = "wasm-beep"
-description = "cpal beep example for WebAssembly"
+name = "webaudio"
+description = "cpal Web Audio example for WebAssembly"
version = "0.2.0"
edition = "2024"
rust-version = "1.85"
diff --git a/examples/wasm-beep/README.md b/examples/webaudio/README.md
similarity index 100%
rename from examples/wasm-beep/README.md
rename to examples/webaudio/README.md
diff --git a/examples/webaudio/index.html b/examples/webaudio/index.html
new file mode 100644
index 000000000..2e97ac7d5
--- /dev/null
+++ b/examples/webaudio/index.html
@@ -0,0 +1,28 @@
+
+
+
+
+
+ cpal Web Audio example
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
Both loopback modes play your microphone back live. Wear headphones to avoid feedback howl.
+
The two-stream mode bridges an independent input and output stream with a ring buffer. The
+ duplex mode runs both directions from one callback on one clock, so it needs no ring buffer
+ and the round trip is audibly shorter.
+
+
+
diff --git a/examples/wasm-beep/src/lib.rs b/examples/webaudio/src/lib.rs
similarity index 71%
rename from examples/wasm-beep/src/lib.rs
rename to examples/webaudio/src/lib.rs
index 1e9c8f0b7..7a55b6674 100644
--- a/examples/wasm-beep/src/lib.rs
+++ b/examples/webaudio/src/lib.rs
@@ -1,7 +1,8 @@
use std::{cell::Cell, rc::Rc};
use cpal::{
- Device, Error, ErrorKind, FromSample, Sample, SampleFormat, SizedSample, Stream, StreamConfig,
+ Device, DuplexCallbackInfo, Error, ErrorKind, FromSample, Sample, SampleFormat, SizedSample,
+ Stream, StreamConfig,
traits::{DeviceTrait, HostTrait, StreamTrait},
};
use ringbuf::{
@@ -24,6 +25,8 @@ pub fn main_js() -> Result<(), JsValue> {
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();
+ let duplex_button = document.get_element_by_id("duplex").unwrap();
+ let stop_duplex_button = document.get_element_by_id("stop-duplex").unwrap();
// stream needs to be referenced from the "play" and "stop" closures
let stream = Rc::new(Cell::new(None));
@@ -75,6 +78,31 @@ pub fn main_js() -> Result<(), JsValue> {
closure.forget();
}
+ // duplex loopback is a single stream, so one slot holds the whole thing
+ let duplex_stream = Rc::new(Cell::new(None));
+
+ // set up duplex button
+ {
+ let duplex_stream = duplex_stream.clone();
+ let closure = Closure::::new(move |_event: web_sys::MouseEvent| {
+ duplex_stream.set(Some(duplex()));
+ });
+ duplex_button
+ .add_event_listener_with_callback("mousedown", closure.as_ref().unchecked_ref())?;
+ closure.forget();
+ }
+
+ // set up stop-duplex button
+ {
+ let closure = Closure::::new(move |_event: web_sys::MouseEvent| {
+ // stop the stream by dropping it; releases the microphone
+ duplex_stream.take();
+ });
+ stop_duplex_button
+ .add_event_listener_with_callback("mousedown", closure.as_ref().unchecked_ref())?;
+ closure.forget();
+ }
+
Ok(())
}
@@ -163,6 +191,54 @@ fn record() -> (Stream, Stream) {
(input_stream, output_stream)
}
+/// The same live microphone loopback as [`record`], but as one duplex stream instead of two
+/// independent ones. Both directions run from a single callback on a single clock, so no ring
+/// buffer is needed to bridge them and the round trip is a callback rather than a delay line.
+/// Wear headphones: routing a live mic to speakers risks feedback howl.
+fn duplex() -> Stream {
+ let host = cpal::default_host();
+ let device = host
+ .default_output_device()
+ .expect("failed to find a default output device");
+ assert!(
+ device.supports_duplex(),
+ "duplex streams need `navigator.mediaDevices`, which browsers expose only in a \
+ secure context: serve this page over HTTPS or from localhost"
+ );
+
+ let config = device.default_duplex_config().unwrap();
+
+ 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()),
+ };
+
+ // WebAudio processes exclusively in f32, so there is no sample format to match on here.
+ let input_channels = config.input_channels as usize;
+ let output_channels = config.output_channels as usize;
+ let stream = device
+ .build_duplex_stream(
+ config,
+ move |input: &[f32], output: &mut [f32], _: &DuplexCallbackInfo| {
+ // The two directions can carry different channel counts, so mix each captured
+ // frame down to mono and fan it back out across the output frame.
+ for (captured, rendered) in input
+ .chunks(input_channels)
+ .zip(output.chunks_mut(output_channels))
+ {
+ rendered.fill(captured.iter().sum::() / input_channels as f32);
+ }
+ },
+ err_fn,
+ None,
+ )
+ .unwrap();
+ stream.start().unwrap();
+ stream
+}
+
fn build_input(device: &Device, config: StreamConfig, mut producer: HeapProd) -> Stream
where
T: Sample + SizedSample,
diff --git a/src/device_description.rs b/src/device_description.rs
index f797aa00b..5618958e5 100644
--- a/src/device_description.rs
+++ b/src/device_description.rs
@@ -143,6 +143,9 @@ pub enum DeviceDirection {
Output,
/// Both input and output
+ ///
+ /// The device has endpoints in both directions. This does not automatically mean that both can
+ /// run from a single stream; see [`DeviceTrait::supports_duplex`](crate::traits::DeviceTrait::supports_duplex).
Duplex,
/// Direction unknown or not yet determined
diff --git a/src/duplex.rs b/src/duplex.rs
index 3800172ff..b98394f18 100644
--- a/src/duplex.rs
+++ b/src/duplex.rs
@@ -2,14 +2,13 @@ use crate::{BufferSize, CallbackInfo, ChannelCount, SampleRate};
/// Information relevant to a single call to the user's duplex stream data callback.
///
-/// Because a duplex stream's input and output share a single clock, `input.timestamp()` and
-/// `output.timestamp()` are drawn from the same time source. The two directions have independent
-/// buffers, so `input.xrun()` and `output.xrun()` can each report a glitch independently for the
-/// same invocation.
+/// `input.timestamp()` and `output.timestamp()` come from the same callback, so they are directly
+/// comparable. Both directions should run on one synchronized clock, but have independent buffers,
+/// so `input.xrun()` and `output.xrun()` can each report a glitch for the same invocation.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub struct DuplexCallbackInfo {
- input: CallbackInfo,
- output: CallbackInfo,
+ pub(crate) input: CallbackInfo,
+ pub(crate) output: CallbackInfo,
}
impl DuplexCallbackInfo {
diff --git a/src/host/audioworklet/mod.rs b/src/host/audioworklet/mod.rs
index aeb077b17..ff1558802 100644
--- a/src/host/audioworklet/mod.rs
+++ b/src/host/audioworklet/mod.rs
@@ -1,7 +1,7 @@
//! Audio Worklet backend implementation.
//!
//! Available on WebAssembly with the `audioworklet` feature. Requires atomics support.
-//! See the `audioworklet-beep` example for setup instructions.
+//! See the `audioworklet` example for setup instructions.
use std::{
cell::RefCell,
@@ -21,9 +21,9 @@ use wasm_bindgen::prelude::*;
use crate::{
BufferSize, CallbackInfo, ChannelCount, Data, DeviceDescription, DeviceDescriptionBuilder,
- DeviceDirection, DeviceId, Error, ErrorKind, FrameCount, Sample, SampleFormat, SampleRate,
- StreamConfig, StreamInstant, StreamTimestamp, SupportedBufferSize, SupportedStreamConfig,
- SupportedStreamConfigRange,
+ DeviceDirection, DeviceId, DuplexCallbackInfo, DuplexStreamConfig, Error, ErrorKind,
+ FrameCount, Sample, SampleFormat, SampleRate, StreamConfig, StreamInstant, StreamTimestamp,
+ SupportedBufferSize, SupportedStreamConfig, SupportedStreamConfigRange,
host::frames_to_duration,
traits::{DeviceTrait, HostTrait, StreamTrait},
};
@@ -77,6 +77,7 @@ 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";
+const DUPLEX_PROCESSOR_NAME: &str = "CpalDuplexProcessor";
fn render_quantum_size_supported() -> bool {
(|| -> Option {
@@ -149,6 +150,21 @@ fn validate_config(config: &StreamConfig, sample_format: SampleFormat) -> Result
Ok(())
}
+/// Applies [`validate_config`] to each direction of a duplex configuration.
+fn validate_duplex_config(
+ config: &DuplexStreamConfig,
+ input_sample_format: SampleFormat,
+ output_sample_format: SampleFormat,
+) -> Result<(), Error> {
+ let per_direction = |channels| StreamConfig {
+ channels,
+ sample_rate: config.sample_rate,
+ buffer_size: config.buffer_size,
+ };
+ validate_config(&per_direction(config.input_channels), input_sample_format)?;
+ validate_config(&per_direction(config.output_channels), output_sample_format)
+}
+
/// 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 {
@@ -271,6 +287,11 @@ impl DeviceTrait for Device {
))
}
+ /// One `AudioWorkletNode` renders both directions from a single `process()` call per quantum.
+ fn supports_duplex(&self) -> bool {
+ crate::host::is_get_user_media_available()
+ }
+
fn supported_input_configs(&self) -> Result {
// 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
@@ -765,6 +786,293 @@ impl DeviceTrait for Device {
buffer_size_frames,
})
}
+
+ /// Create a duplex stream.
+ ///
+ /// # Async completion
+ ///
+ /// Behaves like [`build_input_stream_raw`](Self::build_input_stream_raw): this returns `Ok`
+ /// once the [`AudioContext`] exists, before microphone permission has been resolved and
+ /// before the worklet module has loaded. Failures after that point are delivered to
+ /// `error_callback`, and [`start`](crate::traits::StreamTrait::start) /
+ /// [`pause`](crate::traits::StreamTrait::pause) calls made in the meantime are queued.
+ ///
+ /// [`AudioContext`]: web_sys::AudioContext
+ fn build_duplex_stream_raw(
+ &self,
+ config: DuplexStreamConfig,
+ input_sample_format: SampleFormat,
+ output_sample_format: SampleFormat,
+ data_callback: D,
+ error_callback: E,
+ _timeout: Option,
+ ) -> Result
+ where
+ D: FnMut(&Data, &mut Data, &DuplexCallbackInfo) + Send + 'static,
+ E: FnMut(Error) + Send + 'static,
+ {
+ validate_duplex_config(&config, input_sample_format, output_sample_format)?;
+ // Keep both `device` timestamps 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_duplex_callback(data_callback);
+
+ let input_channels = config.input_channels as u32;
+ let output_channels = config.output_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",
+ )
+ })?;
+
+ let destination = audio_context.destination();
+ if output_channels > destination.max_channel_count() {
+ return Err(Error::with_message(
+ ErrorKind::UnsupportedConfig,
+ format!(
+ "Output channel count {} exceeds the destination's maximum of {}",
+ config.output_channels,
+ destination.max_channel_count()
+ ),
+ ));
+ }
+ destination.set_channel_count(output_channels);
+
+ // 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 latency_nanos = Arc::new(AtomicU64::new(total_latency_nanos(&audio_context)));
+ let latency_nanos_cb = latency_nanos.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,
+ ),
+ 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);
+ options.set_number_of_outputs(1);
+ options.set_output_channel_count(&js_sys::Array::of1(&JsValue::from_f64(
+ output_channels as f64,
+ )));
+ // `config.input_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(input_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(),
+ &WasmAudioDuplexProcessor::new(Box::new(
+ move |input_interleaved,
+ output_interleaved,
+ 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 input = unsafe {
+ Data::from_parts(
+ input_interleaved.as_ptr() as *mut (),
+ input_interleaved.len(),
+ input_sample_format,
+ )
+ };
+ let mut output = unsafe {
+ Data::from_parts(
+ output_interleaved.as_mut_ptr() as *mut (),
+ output_interleaved.len(),
+ output_sample_format,
+ )
+ };
+
+ // One clock: both directions share the same `callback` instant.
+ let callback = StreamInstant::from_secs_f64(now);
+ let buffer_duration =
+ frames_to_duration(frame_size as FrameCount, sample_rate);
+ let latency =
+ Duration::from_nanos(latency_nanos_cb.load(Ordering::Relaxed));
+
+ let info = DuplexCallbackInfo::new(
+ CallbackInfo {
+ timestamp: StreamTimestamp {
+ callback,
+ device: callback
+ .checked_sub(buffer_duration)
+ .unwrap_or(callback),
+ },
+ xrun: false,
+ },
+ CallbackInfo {
+ timestamp: StreamTimestamp {
+ callback,
+ device: callback + (buffer_duration + latency),
+ },
+ xrun: false,
+ },
+ );
+ (data_callback)(&input, &mut output, &info);
+ },
+ ))
+ .pack()
+ .into(),
+ )));
+ let audio_worklet_node = web_sys::AudioWorkletNode::new_with_options(
+ &ctx,
+ DUPLEX_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 duplex processor failed to initialize or crashed",
+ ));
+ });
+ audio_worklet_node
+ .set_onprocessorerror(Some(on_processor_error.as_ref().unchecked_ref()));
+ on_processor_error.forget();
+
+ // Unlike the capture-only path, the node's output is the real playback signal,
+ // so it goes straight to the destination rather than through a muted gain.
+ source.connect_with_audio_node(&audio_worklet_node)?;
+ audio_worklet_node.connect_with_audio_node(&destination)?;
+
+ Ok((source, audio_worklet_node))
+ }
+ .await;
+
+ // Keep both alive until the Stream is dropped, or the source may be garbage-collected
+ // and silently kill the mic connection; see `build_input_stream_raw`.
+ let (_source, _audio_worklet_node) = 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);
+
+ // outputLatency can change at runtime (e.g. an output-device switch) but is only
+ // readable on the main thread, so poll it here and publish it to the worklet via the
+ // shared atomic.
+ let _latency_poller = web_sys::window().and_then(|window| {
+ let poll_ctx = audio_context.clone();
+ let poll_latency = latency_nanos.clone();
+ let closure = Closure::::new(move || {
+ poll_latency.store(total_latency_nanos(&poll_ctx), Ordering::Relaxed);
+ });
+ window
+ .set_interval_with_callback_and_timeout_and_arguments_0(
+ closure.as_ref().unchecked_ref(),
+ LATENCY_POLL_INTERVAL.as_millis() as i32,
+ )
+ .ok()
+ .map(|interval_id| LatencyPoller {
+ window,
+ interval_id,
+ _closure: closure,
+ })
+ });
+
+ // 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,
+ })
+ }
}
impl StreamTrait for Stream {
@@ -814,6 +1122,15 @@ impl Iterator for Devices {
}
}
+/// Grows `buffer` so it can hold `channels * frame_size` interleaved samples, never shrinking it,
+/// and returns the length of the region actually in use.
+fn resize_interleaved(buffer: &mut Vec, channels: u32, frame_size: u32) -> usize {
+ let len = channels as usize * frame_size as usize;
+ buffer.resize(len.max(buffer.len()), f32::EQUILIBRIUM);
+ len
+}
+
+// The interleaved buffer, plus frame size, sample rate, and current time.
type AudioProcessorCallback = Box;
/// WasmAudioProcessor provides an interface for the JavaScript code
@@ -843,31 +1160,19 @@ impl WasmAudioProcessor {
sample_rate: u32,
current_time: f64,
) -> u32 {
- let frame_size = frame_size as usize;
-
- // Ensure there's enough space in the output buffer
- // This likely only occurs once, or very few times.
- let interleaved_buffer_size = channels as usize * frame_size;
- self.interleaved_buffer.resize(
- interleaved_buffer_size.max(self.interleaved_buffer.len()),
- f32::EQUILIBRIUM,
- );
+ let interleaved_buffer_size =
+ resize_interleaved(&mut self.interleaved_buffer, channels, frame_size);
self.interleaved_buffer[..interleaved_buffer_size].fill(f32::EQUILIBRIUM);
(self.callback)(
&mut self.interleaved_buffer[..interleaved_buffer_size],
- frame_size as u32,
+ frame_size,
sample_rate,
current_time,
);
// Returns a pointer to the raw interleaved buffer to Javascript so
// it can deinterleave it into the output buffers.
- //
- // Deinterleaving is done on the Javascript side because it's simpler and it may be faster.
- // Doing it this way avoids an extra copy and the JS deinterleaving code
- // is likely heavily optimized by the browser's JS engine,
- // although I have not tested that assumption.
self.interleaved_buffer.as_mut_ptr() as _
}
@@ -917,11 +1222,7 @@ 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,
- );
+ resize_interleaved(&mut self.interleaved_buffer, channels, frame_size);
self.interleaved_buffer.as_mut_ptr() as _
}
@@ -956,6 +1257,86 @@ impl WasmAudioCaptureProcessor {
}
}
+// The interleaved captured buffer and an interleaved buffer to render into, plus frame size,
+// sample rate, and current time.
+type AudioDuplexCallback = Box;
+
+/// WasmAudioDuplexProcessor provides an interface for the JavaScript code running in the
+/// AudioWorklet to hand captured audio to Rust and take rendered audio back in one call.
+#[wasm_bindgen]
+pub struct WasmAudioDuplexProcessor {
+ input_buffer: Vec,
+ output_buffer: Vec,
+ callback: AudioDuplexCallback,
+}
+
+impl WasmAudioDuplexProcessor {
+ pub fn new(callback: AudioDuplexCallback) -> Self {
+ Self {
+ input_buffer: Vec::new(),
+ output_buffer: Vec::new(),
+ callback,
+ }
+ }
+}
+
+#[wasm_bindgen]
+impl WasmAudioDuplexProcessor {
+ /// Sizes both buffers for this quantum and returns a pointer for JS to interleave captured
+ /// audio into. Must be called to grow Wasm memory before [`process`](Self::process) runs.
+ pub fn prepare(&mut self, input_channels: u32, output_channels: u32, frame_size: u32) -> u32 {
+ resize_interleaved(&mut self.input_buffer, input_channels, frame_size);
+ resize_interleaved(&mut self.output_buffer, output_channels, frame_size);
+ self.input_buffer.as_mut_ptr() as _
+ }
+
+ /// Pointer for JS to deinterleave out. Valid only once [`prepare`](Self::prepare) has run.
+ pub fn output_buffer_ptr(&mut self) -> u32 {
+ self.output_buffer.as_mut_ptr() as _
+ }
+
+ /// Invokes the Rust callback with the captured audio, plus the output buffer to fill.
+ pub fn process(
+ &mut self,
+ input_channels: u32,
+ output_channels: u32,
+ frame_size: u32,
+ sample_rate: u32,
+ current_time: f64,
+ ) {
+ let input_len = input_channels as usize * frame_size as usize;
+ let output_len = output_channels as usize * frame_size as usize;
+
+ // Destructured so the callback can hold both buffers at once.
+ let Self {
+ input_buffer,
+ output_buffer,
+ callback,
+ } = self;
+ output_buffer[..output_len].fill(f32::EQUILIBRIUM);
+ callback(
+ &input_buffer[..input_len],
+ &mut output_buffer[..output_len],
+ 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 583181e6e..51d3e7708 100644
--- a/src/host/audioworklet/worklet.js
+++ b/src/host/audioworklet/worklet.js
@@ -1,61 +1,105 @@
-registerProcessor("CpalProcessor", class WasmProcessor extends AudioWorkletProcessor {
- constructor(options) {
+// Shared plumbing for the cpal processors. Interleaving and deinterleaving are done here rather
+// than in Rust because it avoids an extra copy and the JS engine optimizes these loops well.
+class CpalProcessorBase extends AudioWorkletProcessor {
+ constructor(options, ProcessorClass) {
super();
let [module, memory, handle] = options.processorOptions;
bindgen.initSync({ module, memory });
- this.processor = bindgen.WasmAudioProcessor.unpack(handle);
+ this.processor = ProcessorClass.unpack(handle);
+ this.name = ProcessorClass.name;
this.memory = memory;
this.wasm_memory = new Float32Array(memory.buffer);
}
- process(inputs, outputs) {
- // Check if memory grew and update view
+ // Growing Wasm memory detaches the old view, so it has to be re-taken after any call into
+ // Rust that may have allocated. Both accessors below do this for their caller.
+ refresh_memory() {
if (this.wasm_memory.buffer !== this.memory.buffer) {
this.wasm_memory = new Float32Array(this.memory.buffer);
}
+ }
- const channels = outputs[0];
+ // Resolves `ptr` against the current Wasm memory, or returns -1 (having logged) if the range
+ // it addresses does not fit.
+ sample_offset(ptr, samples) {
+ this.refresh_memory();
+ const start = ptr / Float32Array.BYTES_PER_ELEMENT;
+ if (start + samples > this.wasm_memory.length) {
+ console.error(`${this.name}: Audio buffer out of bounds! Ptr:`, ptr, "Len:", samples);
+ return -1;
+ }
+ return start;
+ }
+
+ // Reads sequentially from `channels` and writes strided into Wasm at byte pointer `ptr`.
+ // Returns false if the buffer does not fit in Wasm memory.
+ interleave(channels, ptr, frame_size) {
const channels_count = channels.length;
- const frame_size = channels[0].length;
- const interleaved_ptr = this.processor.process(
- channels_count,
- frame_size,
- sampleRate,
- currentTime
- );
+ const start = this.sample_offset(ptr, frame_size * channels_count);
+ if (start < 0) {
+ return false;
+ }
- const interleaved_start = interleaved_ptr / Float32Array.BYTES_PER_ELEMENT;
const interleaved = this.wasm_memory;
+ for (let ch = 0; ch < channels_count; ch++) {
+ const channel = channels[ch];
+ let write_pos = start + ch;
- const total_samples = frame_size * channels_count;
- if (interleaved_start + total_samples > this.wasm_memory.length) {
- console.error("CpalProcessor: Audio buffer out of bounds! Ptr:", interleaved_ptr, "Len:", total_samples);
- return false; // Safely stop the node
+ for (let i = 0; i < frame_size; i++) {
+ interleaved[write_pos] = channel[i];
+ write_pos += channels_count;
+ }
+ }
+ return true;
+ }
+
+ // Reads strided from Wasm at byte pointer `ptr` and writes sequentially into `channels`.
+ // Returns false if the buffer does not fit in Wasm memory.
+ deinterleave(channels, ptr, frame_size) {
+ const channels_count = channels.length;
+ const start = this.sample_offset(ptr, frame_size * channels_count);
+ if (start < 0) {
+ return false;
}
- // Deinterleave: read strided from Wasm, write sequential to output
+ const interleaved = this.wasm_memory;
for (let ch = 0; ch < channels_count; ch++) {
const channel = channels[ch];
- let read_pos = interleaved_start + ch;
+ let read_pos = start + ch;
for (let i = 0; i < frame_size; i++) {
channel[i] = interleaved[read_pos];
read_pos += channels_count;
}
}
-
return true;
}
+}
+
+registerProcessor("CpalProcessor", class WasmProcessor extends CpalProcessorBase {
+ constructor(options) {
+ super(options, bindgen.WasmAudioProcessor);
+ }
+
+ process(inputs, outputs) {
+ const channels = outputs[0];
+ const frame_size = channels[0].length;
+
+ const interleaved_ptr = this.processor.process(
+ channels.length,
+ frame_size,
+ sampleRate,
+ currentTime
+ );
+
+ // Safely stop the node if the buffer does not fit.
+ return this.deinterleave(channels, interleaved_ptr, frame_size);
+ }
});
-registerProcessor("CpalCaptureProcessor", class WasmCaptureProcessor extends AudioWorkletProcessor {
+registerProcessor("CpalCaptureProcessor", class WasmCaptureProcessor extends CpalProcessorBase {
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);
+ super(options, bindgen.WasmAudioCaptureProcessor);
}
process(inputs) {
@@ -68,32 +112,52 @@ registerProcessor("CpalCaptureProcessor", class WasmCaptureProcessor extends Aud
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);
+ if (!this.interleave(channels, interleaved_ptr, frame_size)) {
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;
+ this.processor.process_captured(channels_count, frame_size, sampleRate, currentTime);
- for (let i = 0; i < frame_size; i++) {
- this.wasm_memory[write_pos] = channel[i];
- write_pos += channels_count;
- }
+ return true;
+ }
+});
+
+registerProcessor("CpalDuplexProcessor", class WasmDuplexProcessor extends CpalProcessorBase {
+ constructor(options) {
+ super(options, bindgen.WasmAudioDuplexProcessor);
+ }
+
+ process(inputs, outputs) {
+ const output_channels = outputs[0];
+ const output_channels_count = output_channels.length;
+ const frame_size = output_channels[0].length;
+
+ // inputs[0] is empty until the microphone source is connected. Keep rendering output
+ // with a zero-channel input rather than stalling the graph waiting for it.
+ const input_channels = inputs[0];
+ const input_channels_count = input_channels.length;
+
+ const input_ptr = this.processor.prepare(
+ input_channels_count,
+ output_channels_count,
+ frame_size
+ );
+ if (!this.interleave(input_channels, input_ptr, frame_size)) {
+ return false; // Safely stop the node
}
- this.processor.process_captured(channels_count, frame_size, sampleRate, currentTime);
+ // prepare() sized both buffers, so this cannot grow Wasm memory again. Read it before
+ // process() runs the user callback, which can.
+ const output_ptr = this.processor.output_buffer_ptr();
- return true;
+ this.processor.process(
+ input_channels_count,
+ output_channels_count,
+ frame_size,
+ sampleRate,
+ currentTime
+ );
+
+ return this.deinterleave(output_channels, output_ptr, frame_size);
}
});
diff --git a/src/host/mod.rs b/src/host/mod.rs
index b109acb59..b1df58e1d 100644
--- a/src/host/mod.rs
+++ b/src/host/mod.rs
@@ -307,6 +307,26 @@ where
}
}
+/// Wraps a duplex data callback so neither direction's `device` timestamp regresses across
+/// callbacks. The two directions are clamped independently.
+#[allow(dead_code)]
+pub(crate) fn monotonic_duplex_callback(
+ mut data_callback: D,
+) -> impl FnMut(&crate::Data, &mut crate::Data, &crate::DuplexCallbackInfo) + Send + 'static
+where
+ D: FnMut(&crate::Data, &mut crate::Data, &crate::DuplexCallbackInfo) + Send + 'static,
+{
+ let mut input_floor = 0u64;
+ let mut output_floor = 0u64;
+ move |input, output, info| {
+ let mut info = *info;
+ info.input.timestamp.device = non_decreasing(&mut input_floor, info.input.timestamp.device);
+ info.output.timestamp.device =
+ non_decreasing(&mut output_floor, info.output.timestamp.device);
+ data_callback(input, output, &info);
+ }
+}
+
/// Maps a rejected `getUserMedia()` promise to a [`crate::Error`], based on the DOMException
/// `name` the browser rejects with.
///
diff --git a/src/host/webaudio/mod.rs b/src/host/webaudio/mod.rs
index 6b2f3a299..f8cc3d700 100644
--- a/src/host/webaudio/mod.rs
+++ b/src/host/webaudio/mod.rs
@@ -25,11 +25,13 @@ use futures_util::StreamExt as _;
type OutputDataCallbackArc = Arc>;
type InputDataCallbackArc = Arc>;
+type DuplexDataCallbackArc = Arc>;
type CaptureGraphResult = Result<
(
MediaStreamAudioSourceNode,
ScriptProcessorNode,
- GainNode,
+ // Needed by capture-only paths.
+ Option,
Closure,
),
JsValue,
@@ -46,9 +48,9 @@ use self::{
};
use crate::{
BufferSize, CallbackInfo, ChannelCount, Data, DeviceDescription, DeviceDescriptionBuilder,
- DeviceDirection, DeviceId, Error, ErrorKind, FrameCount, Sample, SampleFormat, SampleRate,
- StreamConfig, StreamInstant, StreamTimestamp, SupportedBufferSize, SupportedStreamConfig,
- SupportedStreamConfigRange,
+ DeviceDirection, DeviceId, DuplexCallbackInfo, DuplexStreamConfig, Error, ErrorKind,
+ FrameCount, Sample, SampleFormat, SampleRate, StreamConfig, StreamInstant, StreamTimestamp,
+ SupportedBufferSize, SupportedStreamConfig, SupportedStreamConfigRange,
host::ErrorCallbackArc,
traits::{DeviceTrait, HostTrait, StreamTrait},
};
@@ -109,15 +111,15 @@ 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.
+/// Keeps the getUserMedia() capture graph alive for the lifetime of an input or duplex
+/// [`Stream`], and calls [`stop_tracks`](crate::host::stop_tracks) on drop.
#[cfg(not(target_feature = "atomics"))]
struct CaptureHandles {
media_stream: MediaStream,
_source: MediaStreamAudioSourceNode,
_processor: ScriptProcessorNode,
- _mute_gain: GainNode,
+ /// `None` on the duplex path, whose output goes straight to the destination.
+ _mute_gain: Option,
_on_audio_process: Closure,
}
@@ -275,6 +277,11 @@ impl DeviceTrait for Device {
Self::default_output_config(self)
}
+ /// One `ScriptProcessorNode` renders both directions from a single `onaudioprocess` event.
+ fn supports_duplex(&self) -> bool {
+ is_get_user_media_available()
+ }
+
/// Create an input stream capturing microphone audio via `getUserMedia()`.
///
/// # Async completion
@@ -306,18 +313,7 @@ impl DeviceTrait for Device {
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_size_frames = script_processor_buffer_size(config.buffer_size)?;
let buffer_duration_secs = buffer_time_step_secs(buffer_size_frames, config.sample_rate);
let data_callback = crate::host::monotonic_input_callback(data_callback);
@@ -465,7 +461,7 @@ impl DeviceTrait for Device {
processor.set_onaudioprocess(Some(on_audio_process.as_ref().unchecked_ref()));
- Ok((source, processor, mute_gain, on_audio_process))
+ Ok((source, processor, Some(mute_gain), on_audio_process))
};
let (source, processor, mute_gain, on_audio_process) = match build_graph() {
@@ -946,6 +942,375 @@ impl DeviceTrait for Device {
})
}
}
+
+ /// Create a duplex stream.
+ ///
+ /// # Async completion
+ ///
+ /// Behaves like [`build_input_stream_raw`](Self::build_input_stream_raw): this returns `Ok`
+ /// once the [`AudioContext`] exists, before the browser has granted or denied microphone
+ /// access. Failures after that point are delivered to `error_callback`.
+ fn build_duplex_stream_raw(
+ &self,
+ config: DuplexStreamConfig,
+ input_sample_format: SampleFormat,
+ output_sample_format: SampleFormat,
+ data_callback: D,
+ error_callback: E,
+ _timeout: Option,
+ ) -> Result
+ where
+ D: FnMut(&Data, &mut Data, &DuplexCallbackInfo) + Send + 'static,
+ E: FnMut(Error) + Send + 'static,
+ {
+ validate_duplex_config(&config, input_sample_format, output_sample_format)?;
+
+ let input_channels = config.input_channels as usize;
+ let output_channels = config.output_channels as usize;
+ let buffer_size_frames = script_processor_buffer_size(config.buffer_size)?;
+ let buffer_duration_secs = buffer_time_step_secs(buffer_size_frames, config.sample_rate);
+
+ // Keep both `device` timestamps monotonic: outputLatency can drop (e.g. the page calls
+ // `setSinkId()` to switch output devices), which would pull `device` backward.
+ let data_callback = crate::host::monotonic_duplex_callback(data_callback);
+ let data_callback: DuplexDataCallbackArc = 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",
+ )
+ })?;
+
+ let destination = ctx.destination();
+ if config.output_channels as u32 > destination.max_channel_count() {
+ return Err(Error::with_message(
+ ErrorKind::UnsupportedConfig,
+ format!(
+ "Output channel count {} exceeds the destination's maximum of {}",
+ config.output_channels,
+ destination.max_channel_count()
+ ),
+ ));
+ }
+ destination.set_channel_count(config.output_channels as u32);
+
+ // baseLatency is fixed for the lifetime of the AudioContext.
+ let base_latency_secs = js_sys::Reflect::get(ctx.as_ref(), &JsValue::from("baseLatency"))
+ .ok()
+ .and_then(|v| v.as_f64())
+ .unwrap_or(0.0);
+
+ // SAFETY: see the SAFETY note in `build_output_stream_raw`; 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(not(target_feature = "atomics"))]
+ let stream_config = StreamConfig {
+ channels: config.output_channels,
+ sample_rate: config.sample_rate,
+ buffer_size: config.buffer_size,
+ };
+
+ #[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,
+ input_channels as u32,
+ output_channels as u32,
+ )?;
+ source.connect_with_audio_node(&processor)?;
+ processor.connect_with_audio_node(&ctx_task.destination())?;
+
+ let mut input_buffer_interleaved =
+ vec![f32::EQUILIBRIUM; input_channels * buffer_size_frames];
+ let mut output_buffer_interleaved =
+ vec![f32::EQUILIBRIUM; output_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();
+
+ #[cfg(target_feature = "atomics")]
+ let temporary_channel_array_view = {
+ let temporary_channel_array = js_sys::ArrayBuffer::new(
+ (std::mem::size_of::() * buffer_size_frames) as u32,
+ );
+ js_sys::Float32Array::new(&temporary_channel_array)
+ };
+
+ 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, output_buffer) =
+ match (event.input_buffer(), event.output_buffer()) {
+ (Ok(input), Ok(output)) => (input, output),
+ _ => {
+ (error_callback_cb.lock().unwrap_or_else(|e| e.into_inner()))(
+ Error::with_message(
+ ErrorKind::StreamInvalidated,
+ "Failed to access the duplex audio buffers",
+ ),
+ );
+ return;
+ }
+ };
+
+ // Deinterleave from the browser's per-channel buffers into our interleaved
+ // scratch buffer.
+ for channel in 0..input_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 {
+ input_buffer_interleaved[input_channels * i + channel] =
+ temporary_channel_buffer[i];
+ }
+ }
+
+ output_buffer_interleaved.fill(f32::EQUILIBRIUM);
+
+ let input = unsafe {
+ Data::from_parts(
+ input_buffer_interleaved.as_mut_ptr() as *mut (),
+ input_buffer_interleaved.len(),
+ input_sample_format,
+ )
+ };
+ let mut output = unsafe {
+ Data::from_parts(
+ output_buffer_interleaved.as_mut_ptr() as *mut (),
+ output_buffer_interleaved.len(),
+ output_sample_format,
+ )
+ };
+
+ // outputLatency can change at runtime, so read it each callback.
+ let output_latency_secs =
+ js_sys::Reflect::get(ctx_cb.as_ref(), &JsValue::from("outputLatency"))
+ .ok()
+ .and_then(|v| v.as_f64())
+ .unwrap_or(0.0);
+ let total_hw_latency_secs = {
+ let sum = base_latency_secs + output_latency_secs;
+ if sum.is_finite() { sum.max(0.0) } else { 0.0 }
+ };
+
+ // One clock: both directions share the same `callback` instant.
+ let callback = StreamInstant::from_secs_f64(now);
+ let info = DuplexCallbackInfo::new(
+ CallbackInfo {
+ timestamp: StreamTimestamp {
+ callback,
+ device: StreamInstant::from_secs_f64(
+ (now - buffer_duration_secs).max(0.0),
+ ),
+ },
+ xrun: false,
+ },
+ CallbackInfo {
+ timestamp: StreamTimestamp {
+ callback,
+ device: StreamInstant::from_secs_f64(
+ event.playback_time() + total_hw_latency_secs,
+ ),
+ },
+ xrun: false,
+ },
+ );
+
+ match data_callback_cb.lock() {
+ Ok(mut data_callback) => {
+ (data_callback.deref_mut())(&input, &mut output, &info)
+ }
+ Err(_) => {
+ (error_callback_cb.lock().unwrap_or_else(|e| e.into_inner()))(
+ Error::with_message(
+ ErrorKind::StreamInvalidated,
+ "Stream lock poisoned",
+ ),
+ );
+ return;
+ }
+ }
+
+ // Interleaved scratch back out into the browser's per-channel buffers.
+ for channel in 0..output_channels {
+ for i in 0..buffer_size_frames {
+ temporary_channel_buffer[i] =
+ output_buffer_interleaved[output_channels * i + channel];
+ }
+
+ #[cfg(not(target_feature = "atomics"))]
+ let copied = output_buffer
+ .copy_to_channel(&temporary_channel_buffer, channel as i32);
+
+ // copyToChannel cannot be directly copied into from a SharedArrayBuffer,
+ // which WASM memory is backed by if the 'atomics' flag is enabled.
+ #[cfg(target_feature = "atomics")]
+ let copied = {
+ temporary_channel_array_view.copy_from(&temporary_channel_buffer);
+ output_buffer
+ .unchecked_ref::()
+ .copy_to_channel(&temporary_channel_array_view, channel as i32)
+ };
+
+ if copied.is_err() {
+ (error_callback_cb.lock().unwrap_or_else(|e| e.into_inner()))(
+ Error::with_message(
+ ErrorKind::StreamInvalidated,
+ "Failed to copy rendered audio",
+ ),
+ );
+ return;
+ }
+ }
+ })
+ as Box);
+
+ processor.set_onaudioprocess(Some(on_audio_process.as_ref().unchecked_ref()));
+
+ // The processor's output is the playback signal, so it needs no muted gain.
+ Ok((source, processor, None, 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 duplex graph: {js_err:?}"),
+ ));
+ return;
+ }
+ };
+
+ #[cfg(not(target_feature = "atomics"))]
+ {
+ capture.set(Some(CaptureHandles {
+ media_stream,
+ _source: source,
+ _processor: processor,
+ _mute_gain,
+ _on_audio_process: on_audio_process,
+ }));
+ }
+
+ #[cfg(target_feature = "atomics")]
+ {
+ let _source = source;
+ let _processor = processor;
+ 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,
+ // The ScriptProcessorNode fires on its own once the context is running, so the
+ // buffer-source scheduling the output path needs has no counterpart here.
+ on_ended_closures: Vec::new(),
+ config: stream_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,
+ })
+ }
+ }
}
// Without atomics: AudioContext is accessible directly from Stream.
@@ -1090,6 +1455,38 @@ fn validate_config(config: &StreamConfig, sample_format: SampleFormat) -> Result
Ok(())
}
+/// Gets a configured buffer size that `createScriptProcessor()` accepts.
+fn script_processor_buffer_size(buffer_size: BufferSize) -> Result {
+ let frames = match buffer_size {
+ BufferSize::Fixed(v) => v as usize,
+ BufferSize::Default => DEFAULT_BUFFER_SIZE,
+ };
+ if !SCRIPT_PROCESSOR_VALID_BUFFER_SIZES.contains(&frames) {
+ return Err(Error::with_message(
+ ErrorKind::UnsupportedConfig,
+ format!(
+ "Buffer size {frames} is not supported; must be one of {SCRIPT_PROCESSOR_VALID_BUFFER_SIZES:?}"
+ ),
+ ));
+ }
+ Ok(frames)
+}
+
+/// Applies [`validate_config`] to each direction of a duplex configuration.
+fn validate_duplex_config(
+ config: &DuplexStreamConfig,
+ input_sample_format: SampleFormat,
+ output_sample_format: SampleFormat,
+) -> Result<(), Error> {
+ let per_direction = |channels| StreamConfig {
+ channels,
+ sample_rate: config.sample_rate,
+ buffer_size: config.buffer_size,
+ };
+ validate_config(&per_direction(config.input_channels), input_sample_format)?;
+ validate_config(&per_direction(config.output_channels), output_sample_format)
+}
+
/// 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)
diff --git a/src/traits.rs b/src/traits.rs
index 2a9004ad5..4eec778cd 100644
--- a/src/traits.rs
+++ b/src/traits.rs
@@ -12,8 +12,9 @@ use std::{
};
use crate::{
- CallbackInfo, Data, DeviceDescription, DeviceId, DuplexCallbackInfo, DuplexStreamConfig, Error,
- ErrorKind, InputDevices, OutputDevices, SampleFormat, SizedSample, StreamConfig, StreamInstant,
+ BufferSize, CallbackInfo, ChannelCount, Data, DeviceDescription, DeviceId, DuplexCallbackInfo,
+ DuplexStreamConfig, Error, ErrorKind, InputDevices, OutputDevices, SAMPLE_RATE_48K,
+ SAMPLE_RATE_CD, SampleFormat, SampleRate, SizedSample, StreamConfig, StreamInstant,
SupportedStreamConfig, SupportedStreamConfigRange,
};
@@ -178,14 +179,11 @@ pub trait DeviceTrait: PartialEq + Eq + Hash + Debug + Display + Send + Sync {
.is_ok_and(|mut iter| iter.next().is_some())
}
- /// True if the device can build a synchronized duplex stream where the captured input and
- /// rendered output share a single clock.
+ /// True if the device supports audio input and output from one device-level callback,
+ /// otherwise false
///
- /// Returning `true` is a contract that input and output sides will run from one device-level
- /// callback, or an OS driver aggregate (such as an Aggregate Device on macOS).
- ///
- /// The default implementation returns `false`; hosts that can guarantee a shared clock should
- /// override.
+ /// This is stricter than [`DeviceDirection::Duplex`](crate::DeviceDirection::Duplex), which
+ /// only means the device has both directions.
fn supports_duplex(&self) -> bool {
false
}
@@ -244,6 +242,129 @@ pub trait DeviceTrait: PartialEq + Eq + Hash + Debug + Display + Send + Sync {
/// [`ErrorKind::UnsupportedOperation`]: crate::ErrorKind::UnsupportedOperation
fn default_output_config(&self) -> Result;
+ /// The default duplex stream configuration for the device.
+ ///
+ /// The default implementation prefers a channel count shared by both
+ /// [`supported_input_configs`](Self::supported_input_configs) and
+ /// [`supported_output_configs`](Self::supported_output_configs) (e.g. a stereo interface used
+ /// for stereo passthrough), reconciled on a shared sample rate. Only when no channel count is
+ /// achievable by both directions does it fall back to each direction's own preferred channel
+ /// count.
+ ///
+ /// # Errors
+ ///
+ /// - [`ErrorKind::DeviceNotAvailable`] if the device has been disconnected.
+ /// - [`ErrorKind::UnsupportedConfig`] if no sample rate is supported by both directions.
+ /// - [`ErrorKind::UnsupportedOperation`] if the host or device does not support duplex
+ /// streams.
+ ///
+ /// [`ErrorKind::DeviceNotAvailable`]: crate::ErrorKind::DeviceNotAvailable
+ /// [`ErrorKind::UnsupportedConfig`]: crate::ErrorKind::UnsupportedConfig
+ /// [`ErrorKind::UnsupportedOperation`]: crate::ErrorKind::UnsupportedOperation
+ fn default_duplex_config(&self) -> Result {
+ if !self.supports_duplex() {
+ return Err(Error::with_message(
+ ErrorKind::UnsupportedOperation,
+ "duplex streams are not supported by this device",
+ ));
+ }
+
+ let inputs: Vec<_> = self.supported_input_configs()?.collect();
+ let outputs: Vec<_> = self.supported_output_configs()?.collect();
+
+ let mut candidate_rates: Vec = [
+ self.default_input_config().ok().map(|c| c.sample_rate()),
+ self.default_output_config().ok().map(|c| c.sample_rate()),
+ ]
+ .into_iter()
+ .flatten()
+ .collect();
+ candidate_rates.dedup();
+ candidate_rates.sort_unstable_by(|a, b| b.cmp(a));
+ candidate_rates.extend([SAMPLE_RATE_48K, SAMPLE_RATE_CD]);
+
+ // Otherwise fall back to the highest rate at which the two directions actually overlap.
+ let shared_sample_rate =
+ |inputs: &[SupportedStreamConfigRange], outputs: &[SupportedStreamConfigRange]| {
+ let overlaps = |rate: SampleRate| {
+ inputs.iter().any(|r| r.contains_rate(rate))
+ && outputs.iter().any(|r| r.contains_rate(rate))
+ };
+ candidate_rates
+ .iter()
+ .copied()
+ .find(|&rate| overlaps(rate))
+ .or_else(|| {
+ inputs
+ .iter()
+ .flat_map(|i| {
+ outputs.iter().filter_map(move |o| {
+ let hi = i.max_sample_rate.min(o.max_sample_rate);
+ (i.min_sample_rate.max(o.min_sample_rate) <= hi).then_some(hi)
+ })
+ })
+ .max()
+ })
+ };
+
+ // cpal's own channel preference order (see `cmp_default_heuristics`): stereo, then mono,
+ // then ascending channel count.
+ let channel_rank = |channels: ChannelCount| (channels == 2, channels == 1, channels);
+
+ let matched_channels = inputs
+ .iter()
+ .map(|r| r.channels())
+ .filter(|&c| outputs.iter().any(|r| r.channels() == c))
+ .max_by_key(|&c| channel_rank(c));
+ if let Some(channels) = matched_channels {
+ let matched_inputs: Vec<_> = inputs
+ .iter()
+ .copied()
+ .filter(|r| r.channels() == channels)
+ .collect();
+ let matched_outputs: Vec<_> = outputs
+ .iter()
+ .copied()
+ .filter(|r| r.channels() == channels)
+ .collect();
+ if let Some(sample_rate) = shared_sample_rate(&matched_inputs, &matched_outputs) {
+ return Ok(DuplexStreamConfig {
+ input_channels: channels,
+ output_channels: channels,
+ sample_rate,
+ buffer_size: BufferSize::Default,
+ });
+ }
+ }
+
+ // No channel count is achievable by both directions (or none shares a rate at that
+ // count): fall back to each direction's own best channel count, reconciled only on a
+ // shared sample rate.
+ let sample_rate = shared_sample_rate(&inputs, &outputs).ok_or_else(|| {
+ Error::with_message(
+ ErrorKind::UnsupportedConfig,
+ "no sample rate is supported by both input and output",
+ )
+ })?;
+ let best_channels = |configs: &[SupportedStreamConfigRange], none_supported_message| {
+ configs
+ .iter()
+ .filter(|r| r.contains_rate(sample_rate))
+ .max_by(|a, b| a.cmp_default_heuristics(b))
+ .map(|r| r.channels())
+ .ok_or_else(|| {
+ Error::with_message(ErrorKind::UnsupportedConfig, none_supported_message)
+ })
+ };
+
+ Ok(DuplexStreamConfig {
+ input_channels: best_channels(&inputs, "no supported input configuration")?,
+ output_channels: best_channels(&outputs, "no supported output configuration")?,
+ sample_rate,
+ buffer_size: BufferSize::Default,
+ })
+ }
+
/// Create an input stream.
///
/// # Parameters
@@ -450,6 +571,9 @@ pub trait DeviceTrait: PartialEq + Eq + Hash + Debug + Display + Send + Sync {
/// or OS provided bidirectional aggregate device (macOS). macOS Aggregate device drift
/// compensation is not required.
///
+ /// cpal does not compensate for drift between the two directions, and does not mix or remap
+ /// channels between them.
+ ///
/// # Parameters
///
/// * `config` - Channels, sample rate, and buffer size shared by both directions.