feat: promote linux native audio path
This commit is contained in:
Edison Jwa
Generated
+1293
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,17 @@
|
||||
[package]
|
||||
name = "linux_voice_test"
|
||||
version = "0.1.0"
|
||||
edition = "2021"
|
||||
|
||||
[workspace]
|
||||
|
||||
[dependencies]
|
||||
ctrlc = "3.5"
|
||||
|
||||
[target.'cfg(target_os = "linux")'.dependencies]
|
||||
libpulse-binding = "2.30"
|
||||
libpulse-simple-binding = "2.29"
|
||||
pipewire = { version = "0.10", features = ["v0_3_44"] }
|
||||
|
||||
[target.'cfg(target_os = "windows")'.dependencies]
|
||||
cpal = "0.16"
|
||||
@@ -0,0 +1,37 @@
|
||||
# linux_voice_test
|
||||
|
||||
Purpose: prove a Linux-native voice loopback path that enumerates input/output
|
||||
devices and can target PipeWire or PulseAudio directly instead of relying on
|
||||
`cpal`'s generic ALSA path.
|
||||
|
||||
Scope:
|
||||
- enumerate Linux capture and playback devices through PipeWire or PulseAudio
|
||||
- open a selected input and output device pair
|
||||
- run local microphone-to-speaker loopback at 48 kHz mono f32
|
||||
|
||||
Out of scope:
|
||||
- Chanora product integration
|
||||
- Opus encode/decode
|
||||
- jitter buffer, per-user playback controls, or persistent settings
|
||||
- packaging and service lifecycle
|
||||
|
||||
Current validation:
|
||||
- `cargo check` succeeds on the Fedora-based host after isolating the PoC from
|
||||
the workspace with a local `[workspace]` table
|
||||
- `pkg-config --modversion libpipewire-0.3` and `pkg-config --modversion
|
||||
libspa-0.2` both succeed on that host; `pipewire-0.3` is not the correct
|
||||
module name
|
||||
- `cargo run -- --list-devices` succeeds and lists real PipeWire devices on the
|
||||
host
|
||||
|
||||
Reproduction:
|
||||
|
||||
```bash
|
||||
cd poc/linux_voice_test
|
||||
cargo check
|
||||
cargo run -- --list-devices
|
||||
cargo run -- --backend pipewire --input 0 --output 0
|
||||
```
|
||||
|
||||
See `VERIFICATION.md` for the exact commands and observed results captured in
|
||||
this repository session.
|
||||
@@ -0,0 +1,59 @@
|
||||
# linux_voice_test verification
|
||||
|
||||
Date: 2026-05-25
|
||||
Host: Fedora-based Linux sandbox host
|
||||
|
||||
## Commands
|
||||
|
||||
```bash
|
||||
pkg-config --modversion libpipewire-0.3
|
||||
pkg-config --modversion libspa-0.2
|
||||
cd poc/linux_voice_test
|
||||
cargo check
|
||||
cargo run -- --list-devices
|
||||
```
|
||||
|
||||
## Observed results
|
||||
|
||||
`pkg-config` resolved the product Linux capture dependencies successfully:
|
||||
|
||||
```text
|
||||
libpipewire-0.3 -> 1.6.5
|
||||
libspa-0.2 -> 0.2
|
||||
```
|
||||
|
||||
Note: `pipewire-0.3` is not the correct pkg-config module name for the Rust
|
||||
PipeWire crates used here; the expected module is `libpipewire-0.3`.
|
||||
|
||||
`cargo check` completed successfully after adding an empty `[workspace]` table
|
||||
to `poc/linux_voice_test/Cargo.toml` so the PoC can build independently from
|
||||
the root workspace.
|
||||
|
||||
`cargo run -- --list-devices` completed successfully and printed:
|
||||
|
||||
```text
|
||||
Backend: pipewire
|
||||
Input devices:
|
||||
[0] Brio 500 Analog Stereo [43]
|
||||
[1] Built-in Audio Analog Stereo [34]
|
||||
Backend: pipewire
|
||||
Output devices:
|
||||
[0] Built-in Audio Digital Stereo (IEC958) [46]
|
||||
[1] Navi 21/23 HDMI/DP Audio Controller Digital Stereo (HDMI 4) [50]
|
||||
```
|
||||
|
||||
## What this does validate
|
||||
|
||||
- PipeWire crate dependencies compile on this host
|
||||
- the host exposes the exact `libpipewire-0.3` and `libspa-0.2` pkg-config
|
||||
modules used by the product Linux capture backend
|
||||
- PipeWire node enumeration works for both input and output
|
||||
- the PoC can target real device node ids reported by the running desktop audio
|
||||
graph
|
||||
|
||||
## What this does not validate
|
||||
|
||||
- PulseAudio fallback runtime on this host
|
||||
- full microphone-to-speaker loopback quality over an extended session
|
||||
- Chanora product audio-engine integration
|
||||
- push-to-talk, VAD, Opus, jitter buffering, or per-user playback controls
|
||||
@@ -0,0 +1,29 @@
|
||||
use std::collections::VecDeque;
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
pub const SAMPLE_RATE: u32 = 48_000;
|
||||
#[cfg(target_os = "linux")]
|
||||
pub const CHANNELS: u32 = 1;
|
||||
#[cfg(target_os = "linux")]
|
||||
pub const BUFFER_FRAMES: usize = 256;
|
||||
pub const RING_CAPACITY_SAMPLES: usize = 48_000 * 2;
|
||||
|
||||
pub type SharedBuffer = Arc<Mutex<VecDeque<f32>>>;
|
||||
|
||||
pub fn make_shared_buffer() -> SharedBuffer {
|
||||
Arc::new(Mutex::new(VecDeque::with_capacity(RING_CAPACITY_SAMPLES)))
|
||||
}
|
||||
|
||||
pub fn push_sample(buffer: &SharedBuffer, sample: f32) {
|
||||
let mut queue = buffer.lock().expect("audio buffer mutex poisoned");
|
||||
if queue.len() >= RING_CAPACITY_SAMPLES {
|
||||
let _ = queue.pop_front();
|
||||
}
|
||||
queue.push_back(sample);
|
||||
}
|
||||
|
||||
pub fn pop_sample(buffer: &SharedBuffer) -> f32 {
|
||||
let mut queue = buffer.lock().expect("audio buffer mutex poisoned");
|
||||
queue.pop_front().unwrap_or(0.0)
|
||||
}
|
||||
@@ -0,0 +1,318 @@
|
||||
use crate::audio_buffer::{
|
||||
make_shared_buffer, pop_sample, push_sample, SharedBuffer, CHANNELS, SAMPLE_RATE,
|
||||
};
|
||||
use crate::linux_shared::{
|
||||
sort_devices, wait_for_startup, LinuxAudioBackend, LinuxDevice, LinuxLoopbackSession,
|
||||
ThreadedLoopbackSession,
|
||||
};
|
||||
use pipewire as pw;
|
||||
use pw::spa::pod::Pod;
|
||||
use pw::{properties::properties, spa};
|
||||
use std::cell::{Cell, RefCell};
|
||||
use std::rc::Rc;
|
||||
use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
|
||||
use std::sync::{mpsc, Arc};
|
||||
use std::thread::{self, JoinHandle};
|
||||
use std::time::Duration;
|
||||
|
||||
pub struct PipeWireBackend;
|
||||
|
||||
impl LinuxAudioBackend for PipeWireBackend {
|
||||
fn name(&self) -> &'static str {
|
||||
"pipewire"
|
||||
}
|
||||
|
||||
fn list_input_devices(&self) -> Result<Vec<LinuxDevice>, String> {
|
||||
list_nodes("Audio/Source")
|
||||
}
|
||||
|
||||
fn list_output_devices(&self) -> Result<Vec<LinuxDevice>, String> {
|
||||
list_nodes("Audio/Sink")
|
||||
}
|
||||
|
||||
fn start_loopback(
|
||||
&self,
|
||||
input: &LinuxDevice,
|
||||
output: &LinuxDevice,
|
||||
stop: Arc<AtomicBool>,
|
||||
) -> Result<Box<dyn LinuxLoopbackSession>, String> {
|
||||
let input_id = input
|
||||
.id
|
||||
.parse::<u32>()
|
||||
.map_err(|_| format!("invalid PipeWire input node id `{}`", input.id))?;
|
||||
let output_id = output
|
||||
.id
|
||||
.parse::<u32>()
|
||||
.map_err(|_| format!("invalid PipeWire output node id `{}`", output.id))?;
|
||||
|
||||
let buffer = make_shared_buffer();
|
||||
let capture = spawn_capture(buffer.clone(), input_id, stop.clone())?;
|
||||
let playback = match spawn_playback(buffer, output_id, stop.clone()) {
|
||||
Ok(handle) => handle,
|
||||
Err(err) => {
|
||||
stop.store(true, AtomicOrdering::SeqCst);
|
||||
let _ = capture.join();
|
||||
return Err(err);
|
||||
}
|
||||
};
|
||||
|
||||
Ok(Box::new(ThreadedLoopbackSession {
|
||||
capture,
|
||||
playback,
|
||||
stop,
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
fn list_nodes(class_prefix: &str) -> Result<Vec<LinuxDevice>, String> {
|
||||
let mainloop = pw::main_loop::MainLoopRc::new(None).map_err(|err| err.to_string())?;
|
||||
let context = pw::context::ContextRc::new(&mainloop, None).map_err(|err| err.to_string())?;
|
||||
let core = context.connect_rc(None).map_err(|err| err.to_string())?;
|
||||
let registry = core.get_registry().map_err(|err| err.to_string())?;
|
||||
|
||||
let done = Rc::new(Cell::new(false));
|
||||
let devices = Rc::new(RefCell::new(Vec::<LinuxDevice>::new()));
|
||||
|
||||
let devices_for_registry = devices.clone();
|
||||
let class_prefix = class_prefix.to_string();
|
||||
let class_prefix_for_registry = class_prefix.clone();
|
||||
let _registry_listener = registry
|
||||
.add_listener_local()
|
||||
.global(move |global| {
|
||||
if global.type_ != pw::types::ObjectType::Node {
|
||||
return;
|
||||
}
|
||||
|
||||
let Some(props) = global.props else {
|
||||
return;
|
||||
};
|
||||
|
||||
let Some(media_class) = props.get(*pw::keys::MEDIA_CLASS) else {
|
||||
return;
|
||||
};
|
||||
|
||||
if !media_class.starts_with(class_prefix_for_registry.as_str()) {
|
||||
return;
|
||||
}
|
||||
|
||||
let label = props
|
||||
.get(*pw::keys::NODE_DESCRIPTION)
|
||||
.or_else(|| props.get(*pw::keys::NODE_NICK))
|
||||
.or_else(|| props.get(*pw::keys::NODE_NAME))
|
||||
.unwrap_or("Unnamed PipeWire node")
|
||||
.to_string();
|
||||
|
||||
devices_for_registry.borrow_mut().push(LinuxDevice {
|
||||
id: global.id.to_string(),
|
||||
label,
|
||||
});
|
||||
})
|
||||
.register();
|
||||
|
||||
let pending = core.sync(0).map_err(|err| err.to_string())?;
|
||||
let done_for_core = done.clone();
|
||||
let loop_for_core = mainloop.clone();
|
||||
let _core_listener = core
|
||||
.add_listener_local()
|
||||
.done(move |id, seq| {
|
||||
if id == pw::core::PW_ID_CORE && seq == pending {
|
||||
done_for_core.set(true);
|
||||
loop_for_core.quit();
|
||||
}
|
||||
})
|
||||
.error(move |_id, _seq, _res, message| {
|
||||
eprintln!("PipeWire core error: {message}");
|
||||
})
|
||||
.register();
|
||||
|
||||
while !done.get() {
|
||||
mainloop.run();
|
||||
}
|
||||
|
||||
let mut devices = devices.borrow().clone();
|
||||
devices.sort_by(|left, right| sort_devices(left, right));
|
||||
|
||||
if devices.is_empty() {
|
||||
return Err(format!("no PipeWire nodes found for {class_prefix}"));
|
||||
}
|
||||
|
||||
Ok(devices)
|
||||
}
|
||||
|
||||
fn spawn_capture(
|
||||
buffer: SharedBuffer,
|
||||
source_id: u32,
|
||||
stop: Arc<AtomicBool>,
|
||||
) -> Result<JoinHandle<Result<(), String>>, String> {
|
||||
let (ready_tx, ready_rx) = mpsc::sync_channel::<Result<(), String>>(1);
|
||||
let handle = thread::spawn(move || {
|
||||
let mainloop = pw::main_loop::MainLoopRc::new(None).map_err(|err| err.to_string())?;
|
||||
let context =
|
||||
pw::context::ContextRc::new(&mainloop, None).map_err(|err| err.to_string())?;
|
||||
let core = context.connect_rc(None).map_err(|err| err.to_string())?;
|
||||
|
||||
let mut props = properties! {
|
||||
*pw::keys::MEDIA_TYPE => "Audio",
|
||||
*pw::keys::MEDIA_CATEGORY => "Capture",
|
||||
*pw::keys::MEDIA_ROLE => "Communication",
|
||||
};
|
||||
let target = source_id.to_string();
|
||||
props.insert(*pw::keys::TARGET_OBJECT, target.as_str());
|
||||
|
||||
let stream = pw::stream::StreamBox::new(&core, "linux-voice-test-capture", props)
|
||||
.map_err(|err| err.to_string())?;
|
||||
|
||||
let _listener = stream
|
||||
.add_local_listener_with_user_data(buffer)
|
||||
.process(|stream, queue| {
|
||||
if let Some(mut buffer) = stream.dequeue_buffer() {
|
||||
let datas = buffer.datas_mut();
|
||||
if datas.is_empty() {
|
||||
return;
|
||||
}
|
||||
|
||||
let data_ref = &mut datas[0];
|
||||
let byte_count = data_ref.chunk().size() as usize;
|
||||
let Some(bytes) = data_ref.data() else {
|
||||
return;
|
||||
};
|
||||
|
||||
for chunk in bytes[..byte_count].chunks_exact(4) {
|
||||
let sample = f32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]);
|
||||
push_sample(queue, sample);
|
||||
}
|
||||
}
|
||||
})
|
||||
.register()
|
||||
.map_err(|err| err.to_string())?;
|
||||
|
||||
let values = build_format_param()?;
|
||||
let mut params = [Pod::from_bytes(&values).ok_or("invalid PipeWire capture format pod")?];
|
||||
|
||||
stream
|
||||
.connect(
|
||||
spa::utils::Direction::Input,
|
||||
None,
|
||||
pw::stream::StreamFlags::AUTOCONNECT
|
||||
| pw::stream::StreamFlags::MAP_BUFFERS
|
||||
| pw::stream::StreamFlags::RT_PROCESS,
|
||||
&mut params,
|
||||
)
|
||||
.map_err(|err| err.to_string())?;
|
||||
|
||||
let _ = ready_tx.send(Ok(()));
|
||||
|
||||
while !stop.load(AtomicOrdering::SeqCst) {
|
||||
let _ = mainloop
|
||||
.loop_()
|
||||
.iterate(pw::loop_::Timeout::Finite(Duration::from_millis(100)));
|
||||
}
|
||||
|
||||
mainloop.quit();
|
||||
Ok(())
|
||||
});
|
||||
|
||||
wait_for_startup(ready_rx, handle, "PipeWire capture")
|
||||
}
|
||||
|
||||
fn spawn_playback(
|
||||
buffer: SharedBuffer,
|
||||
sink_id: u32,
|
||||
stop: Arc<AtomicBool>,
|
||||
) -> Result<JoinHandle<Result<(), String>>, String> {
|
||||
let (ready_tx, ready_rx) = mpsc::sync_channel::<Result<(), String>>(1);
|
||||
let handle = thread::spawn(move || {
|
||||
let mainloop = pw::main_loop::MainLoopRc::new(None).map_err(|err| err.to_string())?;
|
||||
let context =
|
||||
pw::context::ContextRc::new(&mainloop, None).map_err(|err| err.to_string())?;
|
||||
let core = context.connect_rc(None).map_err(|err| err.to_string())?;
|
||||
|
||||
let mut props = properties! {
|
||||
*pw::keys::MEDIA_TYPE => "Audio",
|
||||
*pw::keys::MEDIA_CATEGORY => "Playback",
|
||||
*pw::keys::MEDIA_ROLE => "Communication",
|
||||
};
|
||||
let target = sink_id.to_string();
|
||||
props.insert(*pw::keys::TARGET_OBJECT, target.as_str());
|
||||
|
||||
let stream = pw::stream::StreamBox::new(&core, "linux-voice-test-playback", props)
|
||||
.map_err(|err| err.to_string())?;
|
||||
|
||||
let _listener = stream
|
||||
.add_local_listener_with_user_data(buffer)
|
||||
.process(|stream, queue| {
|
||||
if let Some(mut buffer) = stream.dequeue_buffer() {
|
||||
let datas = buffer.datas_mut();
|
||||
if datas.is_empty() {
|
||||
return;
|
||||
}
|
||||
|
||||
let data_ref = &mut datas[0];
|
||||
let byte_len = {
|
||||
let Some(bytes) = data_ref.data() else {
|
||||
return;
|
||||
};
|
||||
|
||||
for chunk in bytes.chunks_exact_mut(4) {
|
||||
chunk.copy_from_slice(&pop_sample(queue).to_le_bytes());
|
||||
}
|
||||
|
||||
bytes.len()
|
||||
};
|
||||
let chunk = data_ref.chunk_mut();
|
||||
*chunk.offset_mut() = 0;
|
||||
*chunk.stride_mut() = 4;
|
||||
*chunk.size_mut() = byte_len as u32;
|
||||
}
|
||||
})
|
||||
.register()
|
||||
.map_err(|err| err.to_string())?;
|
||||
|
||||
let values = build_format_param()?;
|
||||
let mut params = [Pod::from_bytes(&values).ok_or("invalid PipeWire playback format pod")?];
|
||||
|
||||
stream
|
||||
.connect(
|
||||
spa::utils::Direction::Output,
|
||||
None,
|
||||
pw::stream::StreamFlags::AUTOCONNECT
|
||||
| pw::stream::StreamFlags::MAP_BUFFERS
|
||||
| pw::stream::StreamFlags::RT_PROCESS,
|
||||
&mut params,
|
||||
)
|
||||
.map_err(|err| err.to_string())?;
|
||||
|
||||
let _ = ready_tx.send(Ok(()));
|
||||
|
||||
while !stop.load(AtomicOrdering::SeqCst) {
|
||||
let _ = mainloop
|
||||
.loop_()
|
||||
.iterate(pw::loop_::Timeout::Finite(Duration::from_millis(100)));
|
||||
}
|
||||
|
||||
mainloop.quit();
|
||||
Ok(())
|
||||
});
|
||||
|
||||
wait_for_startup(ready_rx, handle, "PipeWire playback")
|
||||
}
|
||||
|
||||
fn build_format_param() -> Result<Vec<u8>, String> {
|
||||
let mut audio_info = spa::param::audio::AudioInfoRaw::new();
|
||||
audio_info.set_format(spa::param::audio::AudioFormat::F32LE);
|
||||
audio_info.set_rate(SAMPLE_RATE);
|
||||
audio_info.set_channels(CHANNELS);
|
||||
|
||||
let object = pw::spa::pod::Object {
|
||||
type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(),
|
||||
id: pw::spa::param::ParamType::EnumFormat.as_raw(),
|
||||
properties: audio_info.into(),
|
||||
};
|
||||
|
||||
pw::spa::pod::serialize::PodSerializer::serialize(
|
||||
std::io::Cursor::new(Vec::new()),
|
||||
&pw::spa::pod::Value::Object(object),
|
||||
)
|
||||
.map_err(|err| err.to_string())
|
||||
.map(|serializer| serializer.0.into_inner())
|
||||
}
|
||||
@@ -0,0 +1,309 @@
|
||||
use crate::audio_buffer::{
|
||||
make_shared_buffer, pop_sample, push_sample, SharedBuffer, BUFFER_FRAMES, CHANNELS, SAMPLE_RATE,
|
||||
};
|
||||
use crate::linux_shared::{
|
||||
sort_devices, wait_for_startup, LinuxAudioBackend, LinuxDevice, LinuxLoopbackSession,
|
||||
ThreadedLoopbackSession,
|
||||
};
|
||||
use libpulse_binding as pulse;
|
||||
use libpulse_simple_binding as psimple;
|
||||
use pulse::callbacks::ListResult;
|
||||
use pulse::context::{self, Context};
|
||||
use pulse::mainloop::threaded::Mainloop;
|
||||
use pulse::proplist::Proplist;
|
||||
use pulse::sample::{Format, Spec};
|
||||
use pulse::stream::Direction;
|
||||
use std::cell::{Cell, RefCell};
|
||||
use std::rc::Rc;
|
||||
use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
|
||||
use std::sync::{mpsc, Arc};
|
||||
use std::thread::{self, JoinHandle};
|
||||
|
||||
pub struct PulseAudioBackend;
|
||||
|
||||
impl LinuxAudioBackend for PulseAudioBackend {
|
||||
fn name(&self) -> &'static str {
|
||||
"pulseaudio"
|
||||
}
|
||||
|
||||
fn list_input_devices(&self) -> Result<Vec<LinuxDevice>, String> {
|
||||
with_pulse_context(|context, mainloop| {
|
||||
let done = Rc::new(Cell::new(false));
|
||||
let devices = Rc::new(RefCell::new(Vec::<LinuxDevice>::new()));
|
||||
|
||||
let done_for_cb = done.clone();
|
||||
let devices_for_cb = devices.clone();
|
||||
let mainloop_for_cb = mainloop.clone();
|
||||
let introspector = context.borrow().introspect();
|
||||
let _operation = introspector.get_source_info_list(move |result| match result {
|
||||
ListResult::Item(info) => {
|
||||
if info.monitor_of_sink.is_some() {
|
||||
return;
|
||||
}
|
||||
|
||||
let Some(name) = info.name.as_ref() else {
|
||||
return;
|
||||
};
|
||||
|
||||
let label = info
|
||||
.description
|
||||
.as_ref()
|
||||
.map(|value| value.to_string())
|
||||
.unwrap_or_else(|| name.to_string());
|
||||
|
||||
devices_for_cb.borrow_mut().push(LinuxDevice {
|
||||
id: name.to_string(),
|
||||
label,
|
||||
});
|
||||
}
|
||||
ListResult::End | ListResult::Error => {
|
||||
done_for_cb.set(true);
|
||||
unsafe { (*mainloop_for_cb.as_ptr()).signal(false) };
|
||||
}
|
||||
});
|
||||
|
||||
while !done.get() {
|
||||
mainloop.borrow_mut().wait();
|
||||
}
|
||||
|
||||
let mut devices = devices.borrow().clone();
|
||||
devices.sort_by(|left, right| sort_devices(left, right));
|
||||
|
||||
if devices.is_empty() {
|
||||
return Err("no PulseAudio sources found".to_string());
|
||||
}
|
||||
|
||||
Ok(devices)
|
||||
})
|
||||
}
|
||||
|
||||
fn list_output_devices(&self) -> Result<Vec<LinuxDevice>, String> {
|
||||
with_pulse_context(|context, mainloop| {
|
||||
let done = Rc::new(Cell::new(false));
|
||||
let devices = Rc::new(RefCell::new(Vec::<LinuxDevice>::new()));
|
||||
|
||||
let done_for_cb = done.clone();
|
||||
let devices_for_cb = devices.clone();
|
||||
let mainloop_for_cb = mainloop.clone();
|
||||
let introspector = context.borrow().introspect();
|
||||
let _operation = introspector.get_sink_info_list(move |result| match result {
|
||||
ListResult::Item(info) => {
|
||||
let Some(name) = info.name.as_ref() else {
|
||||
return;
|
||||
};
|
||||
|
||||
let label = info
|
||||
.description
|
||||
.as_ref()
|
||||
.map(|value| value.to_string())
|
||||
.unwrap_or_else(|| name.to_string());
|
||||
|
||||
devices_for_cb.borrow_mut().push(LinuxDevice {
|
||||
id: name.to_string(),
|
||||
label,
|
||||
});
|
||||
}
|
||||
ListResult::End | ListResult::Error => {
|
||||
done_for_cb.set(true);
|
||||
unsafe { (*mainloop_for_cb.as_ptr()).signal(false) };
|
||||
}
|
||||
});
|
||||
|
||||
while !done.get() {
|
||||
mainloop.borrow_mut().wait();
|
||||
}
|
||||
|
||||
let mut devices = devices.borrow().clone();
|
||||
devices.sort_by(|left, right| sort_devices(left, right));
|
||||
|
||||
if devices.is_empty() {
|
||||
return Err("no PulseAudio sinks found".to_string());
|
||||
}
|
||||
|
||||
Ok(devices)
|
||||
})
|
||||
}
|
||||
|
||||
fn start_loopback(
|
||||
&self,
|
||||
input: &LinuxDevice,
|
||||
output: &LinuxDevice,
|
||||
stop: Arc<AtomicBool>,
|
||||
) -> Result<Box<dyn LinuxLoopbackSession>, String> {
|
||||
let buffer = make_shared_buffer();
|
||||
let capture = spawn_capture(buffer.clone(), input.id.clone(), stop.clone())?;
|
||||
let playback = match spawn_playback(buffer, output.id.clone(), stop.clone()) {
|
||||
Ok(handle) => handle,
|
||||
Err(err) => {
|
||||
stop.store(true, AtomicOrdering::SeqCst);
|
||||
let _ = capture.join();
|
||||
return Err(err);
|
||||
}
|
||||
};
|
||||
|
||||
Ok(Box::new(ThreadedLoopbackSession {
|
||||
capture,
|
||||
playback,
|
||||
stop,
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
fn with_pulse_context<T, F>(operation: F) -> Result<T, String>
|
||||
where
|
||||
F: FnOnce(&Rc<RefCell<Context>>, &Rc<RefCell<Mainloop>>) -> Result<T, String>,
|
||||
{
|
||||
let proplist = build_proplist()?;
|
||||
let mainloop = Rc::new(RefCell::new(
|
||||
Mainloop::new().ok_or("failed to create PulseAudio mainloop")?,
|
||||
));
|
||||
let context = Rc::new(RefCell::new(
|
||||
Context::new_with_proplist(&*mainloop.borrow(), "linux-voice-test", &proplist)
|
||||
.ok_or("failed to create PulseAudio context")?,
|
||||
));
|
||||
|
||||
let mainloop_for_cb = mainloop.clone();
|
||||
let context_for_cb = context.clone();
|
||||
context
|
||||
.borrow_mut()
|
||||
.set_state_callback(Some(Box::new(move || {
|
||||
let state = unsafe { (*context_for_cb.as_ptr()).get_state() };
|
||||
match state {
|
||||
context::State::Ready | context::State::Failed | context::State::Terminated => {
|
||||
unsafe { (*mainloop_for_cb.as_ptr()).signal(false) };
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
})));
|
||||
|
||||
context
|
||||
.borrow_mut()
|
||||
.connect(None, context::FlagSet::NOFLAGS, None)
|
||||
.map_err(pa_error)?;
|
||||
|
||||
mainloop.borrow_mut().lock();
|
||||
mainloop.borrow_mut().start().map_err(pa_error)?;
|
||||
|
||||
loop {
|
||||
match context.borrow().get_state() {
|
||||
context::State::Ready => break,
|
||||
context::State::Failed | context::State::Terminated => {
|
||||
mainloop.borrow_mut().unlock();
|
||||
mainloop.borrow_mut().stop();
|
||||
return Err(format!(
|
||||
"PulseAudio context state {:?}",
|
||||
context.borrow().get_state()
|
||||
));
|
||||
}
|
||||
_ => mainloop.borrow_mut().wait(),
|
||||
}
|
||||
}
|
||||
|
||||
context.borrow_mut().set_state_callback(None);
|
||||
|
||||
let result = operation(&context, &mainloop);
|
||||
mainloop.borrow_mut().unlock();
|
||||
mainloop.borrow_mut().stop();
|
||||
result
|
||||
}
|
||||
|
||||
fn spawn_capture(
|
||||
buffer: SharedBuffer,
|
||||
source_name: String,
|
||||
stop: Arc<AtomicBool>,
|
||||
) -> Result<JoinHandle<Result<(), String>>, String> {
|
||||
let (ready_tx, ready_rx) = mpsc::sync_channel::<Result<(), String>>(1);
|
||||
let handle = thread::spawn(move || {
|
||||
let spec = pulse_spec();
|
||||
let stream = psimple::Simple::new(
|
||||
None,
|
||||
"linux-voice-test",
|
||||
Direction::Record,
|
||||
Some(source_name.as_str()),
|
||||
"linux-voice-test-record",
|
||||
&spec,
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.map_err(pa_error)?;
|
||||
|
||||
let _ = ready_tx.send(Ok(()));
|
||||
|
||||
let mut bytes = vec![0_u8; BUFFER_FRAMES * std::mem::size_of::<f32>()];
|
||||
while !stop.load(AtomicOrdering::SeqCst) {
|
||||
stream.read(&mut bytes).map_err(pa_error)?;
|
||||
for chunk in bytes.chunks_exact(4) {
|
||||
let sample = f32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]);
|
||||
push_sample(&buffer, sample);
|
||||
}
|
||||
}
|
||||
|
||||
stream.flush().map_err(pa_error)?;
|
||||
Ok(())
|
||||
});
|
||||
|
||||
wait_for_startup(ready_rx, handle, "PulseAudio capture")
|
||||
}
|
||||
|
||||
fn spawn_playback(
|
||||
buffer: SharedBuffer,
|
||||
sink_name: String,
|
||||
stop: Arc<AtomicBool>,
|
||||
) -> Result<JoinHandle<Result<(), String>>, String> {
|
||||
let (ready_tx, ready_rx) = mpsc::sync_channel::<Result<(), String>>(1);
|
||||
let handle = thread::spawn(move || {
|
||||
let spec = pulse_spec();
|
||||
let stream = psimple::Simple::new(
|
||||
None,
|
||||
"linux-voice-test",
|
||||
Direction::Playback,
|
||||
Some(sink_name.as_str()),
|
||||
"linux-voice-test-playback",
|
||||
&spec,
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.map_err(pa_error)?;
|
||||
|
||||
let _ = ready_tx.send(Ok(()));
|
||||
|
||||
let mut bytes = vec![0_u8; BUFFER_FRAMES * std::mem::size_of::<f32>()];
|
||||
while !stop.load(AtomicOrdering::SeqCst) {
|
||||
for chunk in bytes.chunks_exact_mut(4) {
|
||||
chunk.copy_from_slice(&pop_sample(&buffer).to_le_bytes());
|
||||
}
|
||||
stream.write(&bytes).map_err(pa_error)?;
|
||||
}
|
||||
|
||||
stream.drain().map_err(pa_error)?;
|
||||
Ok(())
|
||||
});
|
||||
|
||||
wait_for_startup(ready_rx, handle, "PulseAudio playback")
|
||||
}
|
||||
|
||||
fn build_proplist() -> Result<Proplist, String> {
|
||||
let mut proplist = Proplist::new().ok_or("failed to create PulseAudio proplist")?;
|
||||
proplist
|
||||
.set_str(
|
||||
pulse::proplist::properties::APPLICATION_NAME,
|
||||
"linux-voice-test",
|
||||
)
|
||||
.map_err(|_| "failed to set PulseAudio application name".to_string())?;
|
||||
Ok(proplist)
|
||||
}
|
||||
|
||||
fn pulse_spec() -> Spec {
|
||||
let spec = Spec {
|
||||
format: Format::F32le,
|
||||
channels: CHANNELS as u8,
|
||||
rate: SAMPLE_RATE,
|
||||
};
|
||||
debug_assert!(spec.is_valid());
|
||||
spec
|
||||
}
|
||||
|
||||
fn pa_error(err: pulse::error::PAErr) -> String {
|
||||
err.to_string()
|
||||
.unwrap_or_else(|| format!("PulseAudio error {err:?}"))
|
||||
}
|
||||
@@ -0,0 +1,384 @@
|
||||
use crate::linux_pipewire::PipeWireBackend;
|
||||
use crate::linux_pulseaudio::PulseAudioBackend;
|
||||
use std::env;
|
||||
use std::fmt;
|
||||
use std::io::{self, Write};
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::mpsc;
|
||||
use std::sync::Arc;
|
||||
use std::thread;
|
||||
use std::thread::JoinHandle;
|
||||
use std::time::Duration;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct LinuxDevice {
|
||||
pub id: String,
|
||||
pub label: String,
|
||||
}
|
||||
|
||||
impl fmt::Display for LinuxDevice {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
write!(f, "{} [{}]", self.label, self.id)
|
||||
}
|
||||
}
|
||||
|
||||
pub trait LinuxLoopbackSession {
|
||||
fn wait(self: Box<Self>) -> Result<(), String>;
|
||||
}
|
||||
|
||||
pub trait LinuxAudioBackend {
|
||||
fn name(&self) -> &'static str;
|
||||
fn list_input_devices(&self) -> Result<Vec<LinuxDevice>, String>;
|
||||
fn list_output_devices(&self) -> Result<Vec<LinuxDevice>, String>;
|
||||
fn start_loopback(
|
||||
&self,
|
||||
input: &LinuxDevice,
|
||||
output: &LinuxDevice,
|
||||
stop: Arc<AtomicBool>,
|
||||
) -> Result<Box<dyn LinuxLoopbackSession>, String>;
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
enum LinuxBackendKind {
|
||||
Auto,
|
||||
PipeWire,
|
||||
PulseAudio,
|
||||
}
|
||||
|
||||
const USAGE: &str = concat!(
|
||||
"usage: cargo run -- [--backend auto|pipewire|pulseaudio] [--list-devices] ",
|
||||
"[--input <index|id|label>] [--output <index|id|label>]\n\n",
|
||||
"Examples:\n",
|
||||
" cargo run -- --list-devices\n",
|
||||
" cargo run -- --backend pipewire --list-devices\n",
|
||||
" cargo run -- --input 0 --output 1\n",
|
||||
" cargo run -- --backend pulseaudio --input alsa_input.pci-0000_00_1f.3.analog-stereo ",
|
||||
"--output alsa_output.pci-0000_00_1f.3.analog-stereo"
|
||||
);
|
||||
|
||||
pub fn run_from_env() -> Result<(), String> {
|
||||
let args: Vec<String> = env::args().skip(1).collect();
|
||||
let cli = parse_cli(&args)?;
|
||||
let backend = create_backend(cli.backend)?;
|
||||
let inputs = backend.list_input_devices()?;
|
||||
let outputs = backend.list_output_devices()?;
|
||||
|
||||
if cli.list_devices {
|
||||
print_devices(backend.name(), "Input devices:", &inputs);
|
||||
print_devices(backend.name(), "Output devices:", &outputs);
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let input = select_device("input", &inputs, cli.input.as_deref())?;
|
||||
let output = select_device("output", &outputs, cli.output.as_deref())?;
|
||||
|
||||
let stop = Arc::new(AtomicBool::new(false));
|
||||
let stop_for_handler = stop.clone();
|
||||
ctrlc::set_handler(move || {
|
||||
stop_for_handler.store(true, Ordering::SeqCst);
|
||||
})
|
||||
.map_err(|err| err.to_string())?;
|
||||
|
||||
println!("Backend: {}", backend.name());
|
||||
println!("Input: {input}");
|
||||
println!("Output: {output}");
|
||||
println!("Streaming. Press Ctrl+C to stop.");
|
||||
|
||||
let session = backend.start_loopback(&input, &output, stop.clone())?;
|
||||
|
||||
while !stop.load(Ordering::SeqCst) {
|
||||
thread::sleep(Duration::from_millis(100));
|
||||
}
|
||||
|
||||
session.wait()
|
||||
}
|
||||
|
||||
struct LinuxCli {
|
||||
backend: LinuxBackendKind,
|
||||
list_devices: bool,
|
||||
input: Option<String>,
|
||||
output: Option<String>,
|
||||
}
|
||||
|
||||
fn parse_cli(args: &[String]) -> Result<LinuxCli, String> {
|
||||
let mut backend = LinuxBackendKind::Auto;
|
||||
let mut list_devices = false;
|
||||
let mut input = None;
|
||||
let mut output = None;
|
||||
let mut index = 0;
|
||||
|
||||
while index < args.len() {
|
||||
match args[index].as_str() {
|
||||
"--backend" => {
|
||||
index += 1;
|
||||
let value = args
|
||||
.get(index)
|
||||
.ok_or_else(|| format!("missing value for --backend\n{USAGE}"))?;
|
||||
backend = match value.as_str() {
|
||||
"auto" => LinuxBackendKind::Auto,
|
||||
"pipewire" => LinuxBackendKind::PipeWire,
|
||||
"pulseaudio" => LinuxBackendKind::PulseAudio,
|
||||
_ => return Err(format!("unsupported backend `{value}`\n{USAGE}")),
|
||||
};
|
||||
}
|
||||
"--list-devices" => {
|
||||
list_devices = true;
|
||||
}
|
||||
"--input" => {
|
||||
index += 1;
|
||||
input = Some(
|
||||
args.get(index)
|
||||
.ok_or_else(|| format!("missing value for --input\n{USAGE}"))?
|
||||
.clone(),
|
||||
);
|
||||
}
|
||||
"--output" => {
|
||||
index += 1;
|
||||
output = Some(
|
||||
args.get(index)
|
||||
.ok_or_else(|| format!("missing value for --output\n{USAGE}"))?
|
||||
.clone(),
|
||||
);
|
||||
}
|
||||
"--help" | "-h" => {
|
||||
return Err(USAGE.to_string());
|
||||
}
|
||||
other => {
|
||||
return Err(format!("unexpected argument `{other}`\n{USAGE}"));
|
||||
}
|
||||
}
|
||||
|
||||
index += 1;
|
||||
}
|
||||
|
||||
Ok(LinuxCli {
|
||||
backend,
|
||||
list_devices,
|
||||
input,
|
||||
output,
|
||||
})
|
||||
}
|
||||
|
||||
fn create_backend(kind: LinuxBackendKind) -> Result<Box<dyn LinuxAudioBackend>, String> {
|
||||
match kind {
|
||||
LinuxBackendKind::Auto => {
|
||||
let mut errors = Vec::new();
|
||||
|
||||
match create_backend(LinuxBackendKind::PipeWire) {
|
||||
Ok(backend) => return Ok(backend),
|
||||
Err(err) => errors.push(format!("pipewire: {err}")),
|
||||
}
|
||||
|
||||
match create_backend(LinuxBackendKind::PulseAudio) {
|
||||
Ok(backend) => return Ok(backend),
|
||||
Err(err) => errors.push(format!("pulseaudio: {err}")),
|
||||
}
|
||||
|
||||
Err(format!(
|
||||
"failed to initialize a Linux audio backend\n{}",
|
||||
errors.join("\n")
|
||||
))
|
||||
}
|
||||
LinuxBackendKind::PipeWire => {
|
||||
pipewire::init();
|
||||
Ok(Box::new(PipeWireBackend))
|
||||
}
|
||||
LinuxBackendKind::PulseAudio => Ok(Box::new(PulseAudioBackend)),
|
||||
}
|
||||
}
|
||||
|
||||
fn select_device(
|
||||
kind: &str,
|
||||
items: &[LinuxDevice],
|
||||
selector: Option<&str>,
|
||||
) -> Result<LinuxDevice, String> {
|
||||
match selector {
|
||||
Some(selector) => resolve_device_selector(kind, items, selector),
|
||||
None => prompt_choice(&format!("{kind} devices:"), items),
|
||||
}
|
||||
}
|
||||
|
||||
fn resolve_device_selector(
|
||||
kind: &str,
|
||||
items: &[LinuxDevice],
|
||||
selector: &str,
|
||||
) -> Result<LinuxDevice, String> {
|
||||
if items.is_empty() {
|
||||
return Err(format!("no {kind} devices available"));
|
||||
}
|
||||
|
||||
if let Ok(index) = selector.parse::<usize>() {
|
||||
return items
|
||||
.get(index)
|
||||
.cloned()
|
||||
.ok_or_else(|| format!("{kind} device index `{selector}` is out of range"));
|
||||
}
|
||||
|
||||
let mut matches = items
|
||||
.iter()
|
||||
.filter(|item| item.id == selector || item.label == selector)
|
||||
.cloned();
|
||||
|
||||
let Some(device) = matches.next() else {
|
||||
return Err(format!(
|
||||
"no {kind} device matched `{selector}`\nRun with `--list-devices` to inspect valid choices."
|
||||
));
|
||||
};
|
||||
|
||||
if matches.next().is_some() {
|
||||
return Err(format!(
|
||||
"multiple {kind} devices matched `{selector}`\nUse the numeric index from `--list-devices` or the exact device id."
|
||||
));
|
||||
}
|
||||
|
||||
Ok(device)
|
||||
}
|
||||
|
||||
fn print_devices(backend_name: &str, title: &str, items: &[LinuxDevice]) {
|
||||
println!("Backend: {backend_name}");
|
||||
println!("{title}");
|
||||
for (index, item) in items.iter().enumerate() {
|
||||
println!(" [{index}] {item}");
|
||||
}
|
||||
}
|
||||
|
||||
fn prompt_choice(title: &str, items: &[LinuxDevice]) -> Result<LinuxDevice, String> {
|
||||
if items.is_empty() {
|
||||
return Err(format!("no items available for `{title}`"));
|
||||
}
|
||||
|
||||
println!("{title}");
|
||||
for (index, item) in items.iter().enumerate() {
|
||||
println!(" [{index}] {item}");
|
||||
}
|
||||
|
||||
print!("Select number: ");
|
||||
io::stdout().flush().map_err(|err| err.to_string())?;
|
||||
|
||||
let mut input = String::new();
|
||||
io::stdin()
|
||||
.read_line(&mut input)
|
||||
.map_err(|err| err.to_string())?;
|
||||
|
||||
let choice = input
|
||||
.trim()
|
||||
.parse::<usize>()
|
||||
.map_err(|_| "invalid selection".to_string())?;
|
||||
|
||||
items
|
||||
.get(choice)
|
||||
.cloned()
|
||||
.ok_or_else(|| "selection out of range".to_string())
|
||||
}
|
||||
|
||||
pub struct ThreadedLoopbackSession {
|
||||
pub capture: JoinHandle<Result<(), String>>,
|
||||
pub playback: JoinHandle<Result<(), String>>,
|
||||
pub stop: Arc<AtomicBool>,
|
||||
}
|
||||
|
||||
impl LinuxLoopbackSession for ThreadedLoopbackSession {
|
||||
fn wait(self: Box<Self>) -> Result<(), String> {
|
||||
let ThreadedLoopbackSession {
|
||||
capture,
|
||||
playback,
|
||||
stop,
|
||||
} = *self;
|
||||
stop.store(true, Ordering::SeqCst);
|
||||
join_backend_thread(capture)?;
|
||||
join_backend_thread(playback)?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
pub fn wait_for_startup(
|
||||
ready_rx: mpsc::Receiver<Result<(), String>>,
|
||||
handle: JoinHandle<Result<(), String>>,
|
||||
label: &str,
|
||||
) -> Result<JoinHandle<Result<(), String>>, String> {
|
||||
match ready_rx.recv() {
|
||||
Ok(Ok(())) => Ok(handle),
|
||||
Ok(Err(err)) => {
|
||||
let _ = handle.join();
|
||||
Err(err)
|
||||
}
|
||||
Err(err) => {
|
||||
let _ = handle.join();
|
||||
Err(format!("{label} startup failed: {err}"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn join_backend_thread(handle: JoinHandle<Result<(), String>>) -> Result<(), String> {
|
||||
match handle.join() {
|
||||
Ok(result) => result,
|
||||
Err(_) => Err("backend thread panicked".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn sort_devices(left: &LinuxDevice, right: &LinuxDevice) -> std::cmp::Ordering {
|
||||
left.label
|
||||
.cmp(&right.label)
|
||||
.then_with(|| left.id.cmp(&right.id))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn device(id: &str, label: &str) -> LinuxDevice {
|
||||
LinuxDevice {
|
||||
id: id.to_string(),
|
||||
label: label.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_cli_defaults_to_auto_backend() {
|
||||
let cli = parse_cli(&[]).expect("parse_cli should accept empty args");
|
||||
assert!(matches!(cli.backend, LinuxBackendKind::Auto));
|
||||
assert!(!cli.list_devices);
|
||||
assert_eq!(cli.input, None);
|
||||
assert_eq!(cli.output, None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_cli_reads_non_interactive_selection() {
|
||||
let cli = parse_cli(&[
|
||||
"--backend".into(),
|
||||
"pulseaudio".into(),
|
||||
"--input".into(),
|
||||
"1".into(),
|
||||
"--output".into(),
|
||||
"sink-id".into(),
|
||||
"--list-devices".into(),
|
||||
])
|
||||
.expect("parse_cli should parse explicit flags");
|
||||
|
||||
assert!(matches!(cli.backend, LinuxBackendKind::PulseAudio));
|
||||
assert!(cli.list_devices);
|
||||
assert_eq!(cli.input.as_deref(), Some("1"));
|
||||
assert_eq!(cli.output.as_deref(), Some("sink-id"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_device_selector_supports_index_and_exact_id() {
|
||||
let devices = vec![device("mic-id", "Mic"), device("sink-id", "Sink")];
|
||||
|
||||
let by_index = resolve_device_selector("input", &devices, "0")
|
||||
.expect("selector should resolve by index");
|
||||
assert_eq!(by_index.id, "mic-id");
|
||||
|
||||
let by_id = resolve_device_selector("output", &devices, "sink-id")
|
||||
.expect("selector should resolve by id");
|
||||
assert_eq!(by_id.label, "Sink");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_device_selector_rejects_ambiguous_labels() {
|
||||
let devices = vec![device("mic-1", "Mic"), device("mic-2", "Mic")];
|
||||
let err = resolve_device_selector("input", &devices, "Mic")
|
||||
.expect_err("selector should reject ambiguous labels");
|
||||
assert!(err.contains("multiple input devices matched `Mic`"));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
mod audio_buffer;
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
mod linux_pipewire;
|
||||
#[cfg(target_os = "linux")]
|
||||
mod linux_pulseaudio;
|
||||
#[cfg(target_os = "linux")]
|
||||
mod linux_shared;
|
||||
|
||||
#[cfg(target_os = "windows")]
|
||||
mod windows_cpal;
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
fn main() -> Result<(), String> {
|
||||
linux_shared::run_from_env()
|
||||
}
|
||||
|
||||
#[cfg(target_os = "windows")]
|
||||
fn main() -> Result<(), String> {
|
||||
windows_cpal::run()
|
||||
}
|
||||
|
||||
#[cfg(not(any(target_os = "linux", target_os = "windows")))]
|
||||
fn main() {
|
||||
eprintln!("This POC currently supports Windows via cpal and Linux via PipeWire or PulseAudio.");
|
||||
}
|
||||
@@ -0,0 +1,168 @@
|
||||
use crate::audio_buffer::{make_shared_buffer, pop_sample, push_sample};
|
||||
use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
|
||||
use cpal::{SampleFormat, Stream, StreamConfig};
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::thread;
|
||||
use std::time::Duration;
|
||||
|
||||
pub fn run() -> Result<(), String> {
|
||||
let host = cpal::default_host();
|
||||
let input = host
|
||||
.default_input_device()
|
||||
.ok_or("no default input device available")?;
|
||||
let output = host
|
||||
.default_output_device()
|
||||
.ok_or("no default output device available")?;
|
||||
|
||||
println!(
|
||||
"Windows cpal input: {}",
|
||||
input.name().map_err(|err| err.to_string())?
|
||||
);
|
||||
println!(
|
||||
"Windows cpal output: {}",
|
||||
output.name().map_err(|err| err.to_string())?
|
||||
);
|
||||
|
||||
let input_config = input
|
||||
.default_input_config()
|
||||
.map_err(|err| err.to_string())?;
|
||||
let output_config = output
|
||||
.default_output_config()
|
||||
.map_err(|err| err.to_string())?;
|
||||
let config: StreamConfig = input_config.config();
|
||||
let buffer = make_shared_buffer();
|
||||
|
||||
let input_stream = build_input_stream(
|
||||
&input,
|
||||
&config,
|
||||
input_config.sample_format(),
|
||||
buffer.clone(),
|
||||
)?;
|
||||
let output_stream = build_output_stream(
|
||||
&output,
|
||||
&config,
|
||||
output_config.sample_format(),
|
||||
buffer.clone(),
|
||||
)?;
|
||||
|
||||
let stop = Arc::new(AtomicBool::new(false));
|
||||
let stop_for_handler = stop.clone();
|
||||
ctrlc::set_handler(move || {
|
||||
stop_for_handler.store(true, Ordering::SeqCst);
|
||||
})
|
||||
.map_err(|err| err.to_string())?;
|
||||
|
||||
input_stream.play().map_err(|err| err.to_string())?;
|
||||
output_stream.play().map_err(|err| err.to_string())?;
|
||||
|
||||
println!("Streaming with cpal. Press Ctrl+C to stop.");
|
||||
while !stop.load(Ordering::SeqCst) {
|
||||
thread::sleep(Duration::from_millis(100));
|
||||
}
|
||||
|
||||
drop(input_stream);
|
||||
drop(output_stream);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn build_input_stream(
|
||||
device: &cpal::Device,
|
||||
config: &StreamConfig,
|
||||
sample_format: SampleFormat,
|
||||
buffer: crate::audio_buffer::SharedBuffer,
|
||||
) -> Result<Stream, String> {
|
||||
let err_fn = |err| eprintln!("cpal input stream error: {err}");
|
||||
|
||||
match sample_format {
|
||||
SampleFormat::F32 => device
|
||||
.build_input_stream(
|
||||
config,
|
||||
move |data: &[f32], _| {
|
||||
for &sample in data {
|
||||
push_sample(&buffer, sample);
|
||||
}
|
||||
},
|
||||
err_fn,
|
||||
None,
|
||||
)
|
||||
.map_err(|err| err.to_string()),
|
||||
SampleFormat::I16 => device
|
||||
.build_input_stream(
|
||||
config,
|
||||
move |data: &[i16], _| {
|
||||
for &sample in data {
|
||||
push_sample(&buffer, sample as f32 / i16::MAX as f32);
|
||||
}
|
||||
},
|
||||
err_fn,
|
||||
None,
|
||||
)
|
||||
.map_err(|err| err.to_string()),
|
||||
SampleFormat::U16 => device
|
||||
.build_input_stream(
|
||||
config,
|
||||
move |data: &[u16], _| {
|
||||
for &sample in data {
|
||||
let normalized = (sample as f32 / u16::MAX as f32) * 2.0 - 1.0;
|
||||
push_sample(&buffer, normalized);
|
||||
}
|
||||
},
|
||||
err_fn,
|
||||
None,
|
||||
)
|
||||
.map_err(|err| err.to_string()),
|
||||
other => Err(format!("unsupported cpal input sample format: {other:?}")),
|
||||
}
|
||||
}
|
||||
|
||||
fn build_output_stream(
|
||||
device: &cpal::Device,
|
||||
config: &StreamConfig,
|
||||
sample_format: SampleFormat,
|
||||
buffer: crate::audio_buffer::SharedBuffer,
|
||||
) -> Result<Stream, String> {
|
||||
let err_fn = |err| eprintln!("cpal output stream error: {err}");
|
||||
|
||||
match sample_format {
|
||||
SampleFormat::F32 => device
|
||||
.build_output_stream(
|
||||
config,
|
||||
move |data: &mut [f32], _| {
|
||||
for sample in data {
|
||||
*sample = pop_sample(&buffer);
|
||||
}
|
||||
},
|
||||
err_fn,
|
||||
None,
|
||||
)
|
||||
.map_err(|err| err.to_string()),
|
||||
SampleFormat::I16 => device
|
||||
.build_output_stream(
|
||||
config,
|
||||
move |data: &mut [i16], _| {
|
||||
for sample in data {
|
||||
let value = pop_sample(&buffer).clamp(-1.0, 1.0);
|
||||
*sample = (value * i16::MAX as f32) as i16;
|
||||
}
|
||||
},
|
||||
err_fn,
|
||||
None,
|
||||
)
|
||||
.map_err(|err| err.to_string()),
|
||||
SampleFormat::U16 => device
|
||||
.build_output_stream(
|
||||
config,
|
||||
move |data: &mut [u16], _| {
|
||||
for sample in data {
|
||||
let value = pop_sample(&buffer).clamp(-1.0, 1.0);
|
||||
*sample = (((value + 1.0) * 0.5) * u16::MAX as f32) as u16;
|
||||
}
|
||||
},
|
||||
err_fn,
|
||||
None,
|
||||
)
|
||||
.map_err(|err| err.to_string()),
|
||||
other => Err(format!("unsupported cpal output sample format: {other:?}")),
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user