Compare commits
15
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
39b5dafd57 | ||
|
|
a78860db15 | ||
|
|
5f52aa1506 | ||
|
|
d92d0f6f6b | ||
|
|
551767f9f5 | ||
|
|
fa90cd3ce9 | ||
|
|
660261a9a5 | ||
|
|
3a74fd0230 | ||
|
|
2dbb1ea316 | ||
|
|
c902db2e90 | ||
|
|
83e5881768 | ||
|
|
8424b44dec | ||
|
|
33e49a8ca7 | ||
|
|
393c1c7f09 | ||
|
|
1bf14ba08f |
@@ -0,0 +1,42 @@
|
|||||||
|
name: CI
|
||||||
|
|
||||||
|
# Runs on the self-hosted host-mode runner on the desktop (label `arch`). The
|
||||||
|
# gitbutter VPS only queues the job; all compile/test compute happens locally.
|
||||||
|
on:
|
||||||
|
push:
|
||||||
|
branches: [main]
|
||||||
|
pull_request:
|
||||||
|
workflow_dispatch:
|
||||||
|
|
||||||
|
jobs:
|
||||||
|
check:
|
||||||
|
runs-on: arch
|
||||||
|
steps:
|
||||||
|
- name: Checkout
|
||||||
|
uses: actions/checkout@v4
|
||||||
|
|
||||||
|
- name: Toolchain versions
|
||||||
|
run: |
|
||||||
|
rustc --version
|
||||||
|
cargo --version
|
||||||
|
cargo clippy --version
|
||||||
|
cargo deny --version
|
||||||
|
cargo audit --version
|
||||||
|
|
||||||
|
- name: Format check
|
||||||
|
run: cargo fmt --all -- --check
|
||||||
|
|
||||||
|
- name: Clippy (all targets, warnings as errors)
|
||||||
|
run: cargo clippy --all-targets -- -D warnings
|
||||||
|
|
||||||
|
- name: Tests
|
||||||
|
run: cargo test --all-targets
|
||||||
|
|
||||||
|
- name: Doc tests
|
||||||
|
run: cargo test --doc
|
||||||
|
|
||||||
|
- name: cargo-deny (advisories, bans, licenses, sources)
|
||||||
|
run: cargo deny check
|
||||||
|
|
||||||
|
- name: cargo-audit
|
||||||
|
run: cargo audit
|
||||||
Generated
+4
-4
@@ -200,9 +200,9 @@ checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000"
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "anyhow"
|
name = "anyhow"
|
||||||
version = "1.0.102"
|
version = "1.0.103"
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
|
checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3"
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "arbitrary"
|
name = "arbitrary"
|
||||||
@@ -3682,9 +3682,9 @@ checksum = "6b947ae49db0d222b1dbc6b113ce7248a3fc3a6ca21b696717bfc000ba4484d8"
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "memmap2"
|
name = "memmap2"
|
||||||
version = "0.9.10"
|
version = "0.9.11"
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "714098028fe011992e1c3962653c96b2d578c4b4bce9036e15ff220319b1e0e3"
|
checksum = "d1219ed1b7f229ee7104d281dd01d6802fe28bb6e95d292942c4daacdeb798c0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"libc",
|
"libc",
|
||||||
]
|
]
|
||||||
|
|||||||
@@ -0,0 +1,22 @@
|
|||||||
|
use std::process::Command;
|
||||||
|
|
||||||
|
fn main() {
|
||||||
|
println!("cargo:rerun-if-changed=.git/HEAD");
|
||||||
|
if let Ok(head) = std::fs::read_to_string(".git/HEAD")
|
||||||
|
&& let Some(reference) = head.strip_prefix("ref: ")
|
||||||
|
{
|
||||||
|
println!("cargo:rerun-if-changed=.git/{}", reference.trim());
|
||||||
|
}
|
||||||
|
|
||||||
|
let short = Command::new("git")
|
||||||
|
.args(["rev-parse", "--short=8", "HEAD"])
|
||||||
|
.output()
|
||||||
|
.ok()
|
||||||
|
.filter(|output| output.status.success())
|
||||||
|
.and_then(|output| String::from_utf8(output.stdout).ok())
|
||||||
|
.map(|value| value.trim().to_string())
|
||||||
|
.filter(|value| !value.is_empty())
|
||||||
|
.unwrap_or_else(|| "unknown".to_string());
|
||||||
|
|
||||||
|
println!("cargo:rustc-env=PEERSPEAK_GIT_SHORT={short}");
|
||||||
|
}
|
||||||
+1
-1
@@ -1,7 +1,7 @@
|
|||||||
# Maintainer: mollusk <jitty+lc1iz0dc@protonmail.com>
|
# Maintainer: mollusk <jitty+lc1iz0dc@protonmail.com>
|
||||||
pkgname=peerspeak-git
|
pkgname=peerspeak-git
|
||||||
_pkgname=peerspeak
|
_pkgname=peerspeak
|
||||||
pkgver=0.5.0.r0.g0000000
|
pkgver=0.6.1.r310.g660261a
|
||||||
pkgrel=1
|
pkgrel=1
|
||||||
pkgdesc="Decentralized peer-to-peer voice chat (Rust/iroh/PipeWire/Opus/iced)"
|
pkgdesc="Decentralized peer-to-peer voice chat (Rust/iroh/PipeWire/Opus/iced)"
|
||||||
arch=('x86_64')
|
arch=('x86_64')
|
||||||
|
|||||||
+52
-1
@@ -4,7 +4,7 @@ use crate::audio::clip_player::{
|
|||||||
};
|
};
|
||||||
use crate::audio::eq::{EQ_GAIN_DB_MAX, EQ_GAIN_DB_MIN, EqSettings};
|
use crate::audio::eq::{EQ_GAIN_DB_MAX, EQ_GAIN_DB_MIN, EqSettings};
|
||||||
use crate::audio::{AudioDevice, enumerate_audio_devices};
|
use crate::audio::{AudioDevice, enumerate_audio_devices};
|
||||||
use crate::config::{AppConfig, NetworkMode, RecordingMode, RoomLayout};
|
use crate::config::{AppConfig, AudioProfile, NetworkMode, RecordingMode, RoomLayout};
|
||||||
use crate::core::{
|
use crate::core::{
|
||||||
CoreController,
|
CoreController,
|
||||||
messages::{CoreCommand, UiEvent},
|
messages::{CoreCommand, UiEvent},
|
||||||
@@ -34,6 +34,17 @@ use tokio::sync::Mutex;
|
|||||||
|
|
||||||
static UI_RX: OnceLock<Mutex<Option<tokio::sync::mpsc::Receiver<UiEvent>>>> = OnceLock::new();
|
static UI_RX: OnceLock<Mutex<Option<tokio::sync::mpsc::Receiver<UiEvent>>>> = OnceLock::new();
|
||||||
|
|
||||||
|
const APP_VERSION: &str = env!("CARGO_PKG_VERSION");
|
||||||
|
const GIT_SHORT: &str = env!("PEERSPEAK_GIT_SHORT");
|
||||||
|
|
||||||
|
fn app_build_label() -> String {
|
||||||
|
if GIT_SHORT == "unknown" {
|
||||||
|
format!("PeerSpeak v{APP_VERSION}")
|
||||||
|
} else {
|
||||||
|
format!("PeerSpeak v{APP_VERSION} ({GIT_SHORT})")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||||
pub enum Screen {
|
pub enum Screen {
|
||||||
Home,
|
Home,
|
||||||
@@ -366,6 +377,7 @@ pub enum AppMessage {
|
|||||||
/// immediately but does not persist (saved once on release via NoiseGateChanged).
|
/// immediately but does not persist (saved once on release via NoiseGateChanged).
|
||||||
NoiseGateDragging(f32),
|
NoiseGateDragging(f32),
|
||||||
NetworkModeSelected(NetworkMode),
|
NetworkModeSelected(NetworkMode),
|
||||||
|
AudioProfileSelected(AudioProfile),
|
||||||
RecordingModeSelected(RecordingMode),
|
RecordingModeSelected(RecordingMode),
|
||||||
/// Choose the friends presence posture (W7): invisible / normal / discoverable.
|
/// Choose the friends presence posture (W7): invisible / normal / discoverable.
|
||||||
PresenceModeSelected(PresenceMode),
|
PresenceModeSelected(PresenceMode),
|
||||||
@@ -908,6 +920,7 @@ impl Default for AppState {
|
|||||||
let _ = controller.send(CoreCommand::SetInputVolume(config.input_volume));
|
let _ = controller.send(CoreCommand::SetInputVolume(config.input_volume));
|
||||||
let _ = controller.send(CoreCommand::SetOutputVolume(config.output_volume));
|
let _ = controller.send(CoreCommand::SetOutputVolume(config.output_volume));
|
||||||
let _ = controller.send(CoreCommand::SetNetworkMode(config.network_mode));
|
let _ = controller.send(CoreCommand::SetNetworkMode(config.network_mode));
|
||||||
|
let _ = controller.send(CoreCommand::SetAudioProfile(config.audio_profile));
|
||||||
let _ = controller.send(CoreCommand::SetRecordingMode(config.recording_mode));
|
let _ = controller.send(CoreCommand::SetRecordingMode(config.recording_mode));
|
||||||
let _ = controller.send(CoreCommand::SetPixelpassPath(config.pixelpass_path.clone()));
|
let _ = controller.send(CoreCommand::SetPixelpassPath(config.pixelpass_path.clone()));
|
||||||
let _ = controller.send(CoreCommand::SetPresenceMode(config.presence_mode));
|
let _ = controller.send(CoreCommand::SetPresenceMode(config.presence_mode));
|
||||||
@@ -1100,6 +1113,7 @@ fn effective_background_bytes(
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn run_gui() -> iced::Result {
|
pub fn run_gui() -> iced::Result {
|
||||||
|
crate::log_msg(&format!("{} starting", app_build_label()));
|
||||||
// Restore the last window size (saved on close). Position is restored too,
|
// Restore the last window size (saved on close). Position is restored too,
|
||||||
// but only on X11 — Wayland's xdg-shell gives clients no way to set their own
|
// but only on X11 — Wayland's xdg-shell gives clients no way to set their own
|
||||||
// position, so we center there and leave placement to the compositor.
|
// position, so we center there and leave placement to the compositor.
|
||||||
@@ -2038,6 +2052,12 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
|
|||||||
// Applied on the next join, since the endpoint is rebuilt then.
|
// Applied on the next join, since the endpoint is rebuilt then.
|
||||||
let _ = state.controller.send(CoreCommand::SetNetworkMode(mode));
|
let _ = state.controller.send(CoreCommand::SetNetworkMode(mode));
|
||||||
}
|
}
|
||||||
|
AppMessage::AudioProfileSelected(profile) => {
|
||||||
|
state.config.audio_profile = profile;
|
||||||
|
state.config.save();
|
||||||
|
// Applies live to the running encoder, and to the next call.
|
||||||
|
let _ = state.controller.send(CoreCommand::SetAudioProfile(profile));
|
||||||
|
}
|
||||||
AppMessage::RecordingModeSelected(mode) => {
|
AppMessage::RecordingModeSelected(mode) => {
|
||||||
state.config.recording_mode = mode;
|
state.config.recording_mode = mode;
|
||||||
state.config.save();
|
state.config.save();
|
||||||
@@ -3111,6 +3131,19 @@ fn network_mode_hint(mode: NetworkMode) -> &'static str {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// One-line explanation of an audio/network profile for the settings picker (W12).
|
||||||
|
fn audio_profile_hint(profile: AudioProfile) -> &'static str {
|
||||||
|
match profile {
|
||||||
|
AudioProfile::LowLatency => {
|
||||||
|
"Lowest delay, no loss recovery. Best on a clean LAN or wired link."
|
||||||
|
}
|
||||||
|
AudioProfile::Balanced => "Default: voice quality with light loss recovery.",
|
||||||
|
AudioProfile::BadNetwork => {
|
||||||
|
"Most resilient on a lossy/congested link: heavier loss recovery, lower bitrate."
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// One-line explanation of a recording mode for the settings picker.
|
/// One-line explanation of a recording mode for the settings picker.
|
||||||
fn recording_mode_hint(mode: RecordingMode) -> &'static str {
|
fn recording_mode_hint(mode: RecordingMode) -> &'static str {
|
||||||
match mode {
|
match mode {
|
||||||
@@ -3806,6 +3839,7 @@ fn connect_card(state: &AppState) -> Element<'_, AppMessage> {
|
|||||||
let subtitle = text("NAT-traversing full-mesh voice chat")
|
let subtitle = text("NAT-traversing full-mesh voice chat")
|
||||||
.size(16)
|
.size(16)
|
||||||
.color(color_subtext);
|
.color(color_subtext);
|
||||||
|
let build_label = text(app_build_label()).size(11).color(color_subtext);
|
||||||
|
|
||||||
let nickname_input = column![
|
let nickname_input = column![
|
||||||
text("Nickname").size(14).color(color_subtext),
|
text("Nickname").size(14).color(color_subtext),
|
||||||
@@ -3861,6 +3895,7 @@ fn connect_card(state: &AppState) -> Element<'_, AppMessage> {
|
|||||||
column![
|
column![
|
||||||
logo,
|
logo,
|
||||||
subtitle,
|
subtitle,
|
||||||
|
build_label,
|
||||||
vertical_space(20.0),
|
vertical_space(20.0),
|
||||||
nickname_input,
|
nickname_input,
|
||||||
vertical_space(16.0),
|
vertical_space(16.0),
|
||||||
@@ -4967,6 +5002,17 @@ fn view(state: &AppState) -> Element<'_, AppMessage> {
|
|||||||
control
|
control
|
||||||
},
|
},
|
||||||
].spacing(8).width(iced::Length::Fill),
|
].spacing(8).width(iced::Length::Fill),
|
||||||
|
vertical_space(section_gap),
|
||||||
|
section_header("Connection quality"),
|
||||||
|
column![
|
||||||
|
pick_list(
|
||||||
|
&AudioProfile::ALL[..],
|
||||||
|
Some(state.config.audio_profile),
|
||||||
|
AppMessage::AudioProfileSelected,
|
||||||
|
).width(iced::Length::Fill),
|
||||||
|
text(audio_profile_hint(state.config.audio_profile)).size(11).color(color_subtext),
|
||||||
|
text("Applies immediately, even mid-call.").size(11).color(color_subtext),
|
||||||
|
].spacing(4).width(iced::Length::Fill),
|
||||||
]
|
]
|
||||||
.spacing(10)
|
.spacing(10)
|
||||||
.width(iced::Length::Fill)
|
.width(iced::Length::Fill)
|
||||||
@@ -5240,6 +5286,9 @@ fn view(state: &AppState) -> Element<'_, AppMessage> {
|
|||||||
for category in SettingsCategory::ALL {
|
for category in SettingsCategory::ALL {
|
||||||
settings_nav = settings_nav.push(category_button(category));
|
settings_nav = settings_nav.push(category_button(category));
|
||||||
}
|
}
|
||||||
|
settings_nav = settings_nav
|
||||||
|
.push(iced::widget::Space::new().height(iced::Length::Fill))
|
||||||
|
.push(text(app_build_label()).size(11).color(color_subtext));
|
||||||
let settings_nav = container(settings_nav)
|
let settings_nav = container(settings_nav)
|
||||||
.padding(12)
|
.padding(12)
|
||||||
.width(iced::Length::Fixed(220.0))
|
.width(iced::Length::Fixed(220.0))
|
||||||
@@ -5258,6 +5307,8 @@ fn view(state: &AppState) -> Element<'_, AppMessage> {
|
|||||||
.width(iced::Length::Fill),
|
.width(iced::Length::Fill),
|
||||||
vertical_space(10.0),
|
vertical_space(10.0),
|
||||||
settings_body,
|
settings_body,
|
||||||
|
vertical_space(10.0),
|
||||||
|
text(app_build_label()).size(11).color(color_subtext),
|
||||||
]
|
]
|
||||||
.spacing(8)
|
.spacing(8)
|
||||||
.width(iced::Length::Fill),
|
.width(iced::Length::Fill),
|
||||||
|
|||||||
+428
-84
@@ -12,9 +12,11 @@
|
|||||||
//! This module is pure plumbing over [`WavWriter`]: no audio decode, no
|
//! This module is pure plumbing over [`WavWriter`]: no audio decode, no
|
||||||
//! networking, no realtime work. The mixer (a non-RT task) drives it.
|
//! networking, no realtime work. The mixer (a non-RT task) drives it.
|
||||||
|
|
||||||
use std::collections::{HashMap, VecDeque};
|
use std::collections::{HashMap, HashSet, VecDeque};
|
||||||
use std::io;
|
use std::io;
|
||||||
use std::path::{Path, PathBuf};
|
use std::path::{Path, PathBuf};
|
||||||
|
use std::sync::mpsc::{self, SyncSender, TrySendError};
|
||||||
|
use std::thread::{self, JoinHandle};
|
||||||
|
|
||||||
use iroh::EndpointId;
|
use iroh::EndpointId;
|
||||||
|
|
||||||
@@ -24,6 +26,8 @@ use crate::core::jitter::FRAME_SAMPLES;
|
|||||||
/// Cap on the silence chunk written at once when pre-padding a late joiner, so a
|
/// Cap on the silence chunk written at once when pre-padding a late joiner, so a
|
||||||
/// long-running call can't trigger a single multi-hundred-MB allocation.
|
/// long-running call can't trigger a single multi-hundred-MB allocation.
|
||||||
const SILENCE_CHUNK: usize = FRAME_SAMPLES * 256;
|
const SILENCE_CHUNK: usize = FRAME_SAMPLES * 256;
|
||||||
|
const WRITER_QUEUE_CYCLES: usize = 256;
|
||||||
|
const DROP_LOG_INTERVAL_CYCLES: u64 = 256;
|
||||||
|
|
||||||
/// Cap on buffered mic samples (~200ms @ 48kHz). Bounds how far the mic track
|
/// Cap on buffered mic samples (~200ms @ 48kHz). Bounds how far the mic track
|
||||||
/// can drift if the capture clock runs ahead of the mixer cycle; past it the
|
/// can drift if the capture clock runs ahead of the mixer cycle; past it the
|
||||||
@@ -56,41 +60,6 @@ pub fn create_session_dir(base: &Path, now_unix_secs: u64) -> io::Result<PathBuf
|
|||||||
))
|
))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// One output track: its WAV writer plus whether it has been written *this*
|
|
||||||
/// cycle (so `end_cycle` knows which tracks to pad with silence).
|
|
||||||
struct Track {
|
|
||||||
writer: WavWriter,
|
|
||||||
written_this_cycle: bool,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl Track {
|
|
||||||
fn create(path: &Path) -> io::Result<Self> {
|
|
||||||
Ok(Self {
|
|
||||||
writer: WavWriter::new(path)?,
|
|
||||||
written_this_cycle: false,
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Append `frame` fitted to exactly `frame_samples` (zero-padded if short),
|
|
||||||
/// and mark the track as written for this cycle.
|
|
||||||
fn write_frame(&mut self, frame: &[i16], frame_samples: usize) -> io::Result<()> {
|
|
||||||
self.writer.write_samples(&fit(frame, frame_samples))?;
|
|
||||||
self.written_this_cycle = true;
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Append `samples` of silence (no cycle-marking — used for padding).
|
|
||||||
fn write_silence(&mut self, samples: usize) -> io::Result<()> {
|
|
||||||
let mut remaining = samples;
|
|
||||||
while remaining > 0 {
|
|
||||||
let n = remaining.min(SILENCE_CHUNK);
|
|
||||||
self.writer.write_samples(&vec![0i16; n])?;
|
|
||||||
remaining -= n;
|
|
||||||
}
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Return `frame` resized to exactly `n` samples: truncated if longer (shouldn't
|
/// Return `frame` resized to exactly `n` samples: truncated if longer (shouldn't
|
||||||
/// happen — Opus frames are uniform), zero-padded if shorter.
|
/// happen — Opus frames are uniform), zero-padded if shorter.
|
||||||
fn fit(frame: &[i16], n: usize) -> Vec<i16> {
|
fn fit(frame: &[i16], n: usize) -> Vec<i16> {
|
||||||
@@ -126,41 +95,210 @@ pub fn track_filename(name: &str, id: &EndpointId) -> String {
|
|||||||
format!("{slug}-{short}.wav")
|
format!("{slug}-{short}.wav")
|
||||||
}
|
}
|
||||||
|
|
||||||
/// A live multitrack recording: per-peer stems + your mic, plus an optional
|
#[derive(Default)]
|
||||||
/// mixed track, all under one session directory and clocked together.
|
struct PendingCycle {
|
||||||
pub struct MultitrackRecorder {
|
new_peers: Vec<NewPeer>,
|
||||||
dir: PathBuf,
|
peer_frames: HashMap<EndpointId, Vec<i16>>,
|
||||||
frame_samples: usize,
|
mix_frame: Option<Vec<i16>>,
|
||||||
/// Cycles recorded so far = the shared length (in frames) of every track.
|
|
||||||
cycles: u64,
|
|
||||||
peers: HashMap<EndpointId, Track>,
|
|
||||||
/// Your mic track. Fed asynchronously from the capture thread via
|
|
||||||
/// [`push_mic`](MultitrackRecorder::push_mic) into `mic_fifo`, then drained
|
|
||||||
/// one frame per `end_cycle` so it aligns with the cycle clock.
|
|
||||||
mic: WavWriter,
|
|
||||||
mic_fifo: VecDeque<i16>,
|
|
||||||
/// Present in "Both" mode (stems + mixed), absent in "stems only".
|
|
||||||
mix: Option<Track>,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
impl MultitrackRecorder {
|
struct NewPeer {
|
||||||
/// Create a recording in `dir` (which must already exist). `with_mix` adds
|
id: EndpointId,
|
||||||
/// the convenience mixed track (`mix.wav`). Your mic is always `me.wav`.
|
filename: String,
|
||||||
pub fn create(dir: &Path, frame_samples: usize, with_mix: bool) -> io::Result<Self> {
|
}
|
||||||
|
|
||||||
|
struct CycleBatch {
|
||||||
|
new_peers: Vec<NewPeer>,
|
||||||
|
mic_frame: Vec<i16>,
|
||||||
|
mix_frame: Option<Vec<i16>>,
|
||||||
|
peer_frames: HashMap<EndpointId, Vec<i16>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
trait SampleWriter {
|
||||||
|
fn write_samples(&mut self, samples: &[i16]) -> io::Result<()>;
|
||||||
|
fn finalize(self) -> io::Result<()>;
|
||||||
|
}
|
||||||
|
|
||||||
|
impl SampleWriter for WavWriter {
|
||||||
|
fn write_samples(&mut self, samples: &[i16]) -> io::Result<()> {
|
||||||
|
WavWriter::write_samples(self, samples)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn finalize(self) -> io::Result<()> {
|
||||||
|
WavWriter::finalize(self)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct WriterState<W> {
|
||||||
|
dir: PathBuf,
|
||||||
|
frame_samples: usize,
|
||||||
|
peers: HashMap<EndpointId, W>,
|
||||||
|
mic: W,
|
||||||
|
mix: Option<W>,
|
||||||
|
cycles_written: u64,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl WriterState<WavWriter> {
|
||||||
|
fn create(dir: &Path, frame_samples: usize, with_mix: bool) -> io::Result<Self> {
|
||||||
let mic = WavWriter::new(&dir.join("me.wav"))?;
|
let mic = WavWriter::new(&dir.join("me.wav"))?;
|
||||||
let mix = if with_mix {
|
let mix = if with_mix {
|
||||||
Some(Track::create(&dir.join("mix.wav"))?)
|
Some(WavWriter::new(&dir.join("mix.wav"))?)
|
||||||
} else {
|
} else {
|
||||||
None
|
None
|
||||||
};
|
};
|
||||||
Ok(Self {
|
Ok(Self {
|
||||||
dir: dir.to_path_buf(),
|
dir: dir.to_path_buf(),
|
||||||
frame_samples,
|
frame_samples,
|
||||||
cycles: 0,
|
|
||||||
peers: HashMap::new(),
|
peers: HashMap::new(),
|
||||||
mic,
|
mic,
|
||||||
mic_fifo: VecDeque::new(),
|
|
||||||
mix,
|
mix,
|
||||||
|
cycles_written: 0,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<W: SampleWriter> WriterState<W> {
|
||||||
|
fn apply_batch<F>(&mut self, batch: &CycleBatch, mut create_peer: F) -> io::Result<()>
|
||||||
|
where
|
||||||
|
F: FnMut(&Path) -> io::Result<W>,
|
||||||
|
{
|
||||||
|
for peer in &batch.new_peers {
|
||||||
|
if !self.peers.contains_key(&peer.id) {
|
||||||
|
let writer = create_peer(&self.dir.join(&peer.filename))?;
|
||||||
|
self.peers.insert(peer.id, writer);
|
||||||
|
let pad = self.back_pad_samples()?;
|
||||||
|
let writer = self.peers.get_mut(&peer.id).unwrap();
|
||||||
|
Self::write_silence(writer, pad)?;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
self.mic.write_samples(&batch.mic_frame)?;
|
||||||
|
if let Some(mix) = self.mix.as_mut() {
|
||||||
|
if let Some(frame) = batch.mix_frame.as_deref() {
|
||||||
|
mix.write_samples(frame)?;
|
||||||
|
} else {
|
||||||
|
Self::write_silence(mix, self.frame_samples)?;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let silence = vec![0i16; self.frame_samples];
|
||||||
|
for (id, writer) in &mut self.peers {
|
||||||
|
let frame = batch
|
||||||
|
.peer_frames
|
||||||
|
.get(id)
|
||||||
|
.map(Vec::as_slice)
|
||||||
|
.unwrap_or(&silence);
|
||||||
|
writer.write_samples(frame)?;
|
||||||
|
}
|
||||||
|
self.cycles_written += 1;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn back_pad_samples(&self) -> io::Result<usize> {
|
||||||
|
let cycles = usize::try_from(self.cycles_written)
|
||||||
|
.map_err(|_| io::Error::other("multitrack recording too long"))?;
|
||||||
|
cycles
|
||||||
|
.checked_mul(self.frame_samples)
|
||||||
|
.ok_or_else(|| io::Error::other("multitrack recording too long"))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn write_silence(writer: &mut W, samples: usize) -> io::Result<()> {
|
||||||
|
let mut remaining = samples;
|
||||||
|
let silence = vec![0i16; remaining.min(SILENCE_CHUNK)];
|
||||||
|
while remaining > 0 {
|
||||||
|
let n = remaining.min(silence.len());
|
||||||
|
writer.write_samples(&silence[..n])?;
|
||||||
|
remaining -= n;
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn finalize(self) -> io::Result<()> {
|
||||||
|
let mut first_finalize_error = None;
|
||||||
|
record_first_error(&mut first_finalize_error, self.mic.finalize());
|
||||||
|
if let Some(mix) = self.mix {
|
||||||
|
record_first_error(&mut first_finalize_error, mix.finalize());
|
||||||
|
}
|
||||||
|
for writer in self.peers.into_values() {
|
||||||
|
record_first_error(&mut first_finalize_error, writer.finalize());
|
||||||
|
}
|
||||||
|
if let Some(e) = first_finalize_error {
|
||||||
|
Err(e)
|
||||||
|
} else {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn record_first_error(slot: &mut Option<io::Error>, result: io::Result<()>) {
|
||||||
|
if slot.is_none()
|
||||||
|
&& let Err(e) = result
|
||||||
|
{
|
||||||
|
*slot = Some(e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Applies whole-cycle batches on the writer thread. Each applied batch appends
|
||||||
|
/// exactly `frame_samples` to every existing track, and a dropped batch never
|
||||||
|
/// reaches this loop for any track, so stem lengths stay equal even when the
|
||||||
|
/// bounded queue applies back-pressure.
|
||||||
|
fn writer_thread_main(
|
||||||
|
mut state: WriterState<WavWriter>,
|
||||||
|
batch_rx: mpsc::Receiver<CycleBatch>,
|
||||||
|
) -> io::Result<()> {
|
||||||
|
let mut first_write_error = None;
|
||||||
|
|
||||||
|
for batch in batch_rx {
|
||||||
|
if first_write_error.is_none()
|
||||||
|
&& let Err(e) = state.apply_batch(&batch, WavWriter::new)
|
||||||
|
{
|
||||||
|
first_write_error = Some(e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let finalize_result = state.finalize();
|
||||||
|
if let Some(e) = first_write_error {
|
||||||
|
Err(e)
|
||||||
|
} else {
|
||||||
|
finalize_result
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A live multitrack recording: per-peer stems + your mic, plus an optional
|
||||||
|
/// mixed track, all under one session directory and clocked together.
|
||||||
|
pub struct MultitrackRecorder {
|
||||||
|
dir: PathBuf,
|
||||||
|
frame_samples: usize,
|
||||||
|
known_peers: HashSet<EndpointId>,
|
||||||
|
/// Your mic track. Fed asynchronously from the capture thread via
|
||||||
|
/// [`push_mic`](MultitrackRecorder::push_mic) into `mic_fifo`, then drained
|
||||||
|
/// one frame per `end_cycle` so it aligns with the cycle clock.
|
||||||
|
mic_fifo: VecDeque<i16>,
|
||||||
|
/// Present in "Both" mode (stems + mixed), absent in "stems only".
|
||||||
|
with_mix: bool,
|
||||||
|
batch_tx: SyncSender<CycleBatch>,
|
||||||
|
writer_thread: JoinHandle<io::Result<()>>,
|
||||||
|
dropped_cycles: u64,
|
||||||
|
pending: PendingCycle,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl MultitrackRecorder {
|
||||||
|
/// Create a recording in `dir` (which must already exist). `with_mix` adds
|
||||||
|
/// the convenience mixed track (`mix.wav`). Your mic is always `me.wav`.
|
||||||
|
pub fn create(dir: &Path, frame_samples: usize, with_mix: bool) -> io::Result<Self> {
|
||||||
|
let writer_state = WriterState::create(dir, frame_samples, with_mix)?;
|
||||||
|
let (batch_tx, batch_rx) = mpsc::sync_channel(WRITER_QUEUE_CYCLES);
|
||||||
|
let writer_thread = thread::spawn(move || writer_thread_main(writer_state, batch_rx));
|
||||||
|
Ok(Self {
|
||||||
|
dir: dir.to_path_buf(),
|
||||||
|
frame_samples,
|
||||||
|
known_peers: HashSet::new(),
|
||||||
|
mic_fifo: VecDeque::new(),
|
||||||
|
with_mix,
|
||||||
|
batch_tx,
|
||||||
|
writer_thread,
|
||||||
|
dropped_cycles: 0,
|
||||||
|
pending: PendingCycle::default(),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -173,12 +311,14 @@ impl MultitrackRecorder {
|
|||||||
/// so it aligns with the others. Idempotent: a peer already tracked is left
|
/// so it aligns with the others. Idempotent: a peer already tracked is left
|
||||||
/// as-is (re-announce / name change doesn't restart their file).
|
/// as-is (re-announce / name change doesn't restart their file).
|
||||||
pub fn add_peer(&mut self, id: EndpointId, name: &str) -> io::Result<()> {
|
pub fn add_peer(&mut self, id: EndpointId, name: &str) -> io::Result<()> {
|
||||||
if self.peers.contains_key(&id) {
|
if self.known_peers.contains(&id) {
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
let mut track = Track::create(&self.dir.join(track_filename(name, &id)))?;
|
self.known_peers.insert(id);
|
||||||
track.write_silence(self.cycles as usize * self.frame_samples)?;
|
self.pending.new_peers.push(NewPeer {
|
||||||
self.peers.insert(id, track);
|
id,
|
||||||
|
filename: track_filename(name, &id),
|
||||||
|
});
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -186,11 +326,13 @@ impl MultitrackRecorder {
|
|||||||
/// registered yet (write raced ahead of the join event), auto-register it
|
/// registered yet (write raced ahead of the join event), auto-register it
|
||||||
/// with an id-only name so no audio is dropped.
|
/// with an id-only name so no audio is dropped.
|
||||||
pub fn write_peer(&mut self, id: EndpointId, frame: &[i16]) -> io::Result<()> {
|
pub fn write_peer(&mut self, id: EndpointId, frame: &[i16]) -> io::Result<()> {
|
||||||
if !self.peers.contains_key(&id) {
|
if !self.known_peers.contains(&id) {
|
||||||
self.add_peer(id, "")?;
|
self.add_peer(id, "")?;
|
||||||
}
|
}
|
||||||
let fs = self.frame_samples;
|
self.pending
|
||||||
self.peers.get_mut(&id).unwrap().write_frame(frame, fs)
|
.peer_frames
|
||||||
|
.insert(id, fit(frame, self.frame_samples));
|
||||||
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Buffer a frame of your transmitted mic audio (called from the capture
|
/// Buffer a frame of your transmitted mic audio (called from the capture
|
||||||
@@ -215,9 +357,8 @@ impl MultitrackRecorder {
|
|||||||
/// Record the finished mixed-bus frame for the current cycle (no-op in
|
/// Record the finished mixed-bus frame for the current cycle (no-op in
|
||||||
/// stems-only mode).
|
/// stems-only mode).
|
||||||
pub fn write_mix(&mut self, frame: &[i16]) -> io::Result<()> {
|
pub fn write_mix(&mut self, frame: &[i16]) -> io::Result<()> {
|
||||||
let fs = self.frame_samples;
|
if self.with_mix {
|
||||||
if let Some(mix) = self.mix.as_mut() {
|
self.pending.mix_frame = Some(fit(frame, self.frame_samples));
|
||||||
mix.write_frame(frame, fs)?;
|
|
||||||
}
|
}
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
@@ -230,28 +371,63 @@ impl MultitrackRecorder {
|
|||||||
// Mic: always one frame per cycle, drained from the FIFO (silence on
|
// Mic: always one frame per cycle, drained from the FIFO (silence on
|
||||||
// underrun), so it tracks the cycle clock like the peer stems.
|
// underrun), so it tracks the cycle clock like the peer stems.
|
||||||
let mic_frame = self.drain_mic(fs);
|
let mic_frame = self.drain_mic(fs);
|
||||||
self.mic.write_samples(&mic_frame)?;
|
let mut pending = std::mem::take(&mut self.pending);
|
||||||
// Peers + the optional mix track: pad any not written this cycle.
|
pending.new_peers.sort_by(|a, b| {
|
||||||
for track in self.peers.values_mut().chain(self.mix.as_mut()) {
|
a.filename
|
||||||
if !track.written_this_cycle {
|
.cmp(&b.filename)
|
||||||
track.write_silence(fs)?;
|
.then_with(|| a.id.to_string().cmp(&b.id.to_string()))
|
||||||
|
});
|
||||||
|
let batch = CycleBatch {
|
||||||
|
new_peers: pending.new_peers,
|
||||||
|
mic_frame,
|
||||||
|
mix_frame: if self.with_mix {
|
||||||
|
pending.mix_frame
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
},
|
||||||
|
peer_frames: pending.peer_frames,
|
||||||
|
};
|
||||||
|
match self.batch_tx.try_send(batch) {
|
||||||
|
Ok(()) => Ok(()),
|
||||||
|
Err(TrySendError::Full(batch)) => {
|
||||||
|
for peer in &batch.new_peers {
|
||||||
|
self.known_peers.remove(&peer.id);
|
||||||
|
}
|
||||||
|
self.dropped_cycles = self.dropped_cycles.saturating_add(1);
|
||||||
|
if self.dropped_cycles == 1
|
||||||
|
|| self.dropped_cycles.is_multiple_of(DROP_LOG_INTERVAL_CYCLES)
|
||||||
|
{
|
||||||
|
crate::log_msg(&format!(
|
||||||
|
"multitrack recording: writer queue full; dropped {} cycle(s)",
|
||||||
|
self.dropped_cycles
|
||||||
|
));
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
}
|
}
|
||||||
track.written_this_cycle = false;
|
Err(TrySendError::Disconnected(_)) => Err(io::Error::new(
|
||||||
|
io::ErrorKind::BrokenPipe,
|
||||||
|
"multitrack writer thread stopped",
|
||||||
|
)),
|
||||||
}
|
}
|
||||||
self.cycles += 1;
|
|
||||||
Ok(())
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Finalize every track's WAV header. Consumes the recorder.
|
/// Finalize every track's WAV header. Consumes the recorder.
|
||||||
pub fn finalize(self) -> io::Result<()> {
|
pub fn finalize(self) -> io::Result<()> {
|
||||||
self.mic.finalize()?;
|
let Self {
|
||||||
if let Some(mix) = self.mix {
|
dir: _,
|
||||||
mix.writer.finalize()?;
|
frame_samples: _,
|
||||||
}
|
known_peers: _,
|
||||||
for (_, track) in self.peers {
|
mic_fifo: _,
|
||||||
track.writer.finalize()?;
|
with_mix: _,
|
||||||
}
|
batch_tx,
|
||||||
Ok(())
|
writer_thread,
|
||||||
|
dropped_cycles: _,
|
||||||
|
pending: _,
|
||||||
|
} = self;
|
||||||
|
drop(batch_tx);
|
||||||
|
writer_thread
|
||||||
|
.join()
|
||||||
|
.unwrap_or_else(|_| Err(io::Error::other("multitrack writer thread panicked")))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -277,6 +453,51 @@ mod tests {
|
|||||||
d
|
d
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Default)]
|
||||||
|
struct TestWriter {
|
||||||
|
samples: Vec<i16>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl SampleWriter for TestWriter {
|
||||||
|
fn write_samples(&mut self, samples: &[i16]) -> io::Result<()> {
|
||||||
|
self.samples.extend_from_slice(samples);
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn finalize(self) -> io::Result<()> {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn test_writer_state(frame_samples: usize, with_mix: bool) -> WriterState<TestWriter> {
|
||||||
|
WriterState {
|
||||||
|
dir: PathBuf::new(),
|
||||||
|
frame_samples,
|
||||||
|
peers: HashMap::new(),
|
||||||
|
mic: TestWriter::default(),
|
||||||
|
mix: if with_mix {
|
||||||
|
Some(TestWriter::default())
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
},
|
||||||
|
cycles_written: 0,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn test_batch(
|
||||||
|
new_peers: Vec<NewPeer>,
|
||||||
|
mic_frame: Vec<i16>,
|
||||||
|
mix_frame: Option<Vec<i16>>,
|
||||||
|
peer_frames: Vec<(EndpointId, Vec<i16>)>,
|
||||||
|
) -> CycleBatch {
|
||||||
|
CycleBatch {
|
||||||
|
new_peers,
|
||||||
|
mic_frame,
|
||||||
|
mix_frame,
|
||||||
|
peer_frames: peer_frames.into_iter().collect(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn fit_pads_and_truncates() {
|
fn fit_pads_and_truncates() {
|
||||||
assert_eq!(fit(&[1, 2], 4), vec![1, 2, 0, 0]);
|
assert_eq!(fit(&[1, 2], 4), vec![1, 2, 0, 0]);
|
||||||
@@ -311,6 +532,110 @@ mod tests {
|
|||||||
let _ = std::fs::remove_dir_all(&base);
|
let _ = std::fs::remove_dir_all(&base);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn apply_batch_advances_existing_tracks_and_back_pads_late_peer() {
|
||||||
|
let frame = 3;
|
||||||
|
let early = an_id();
|
||||||
|
let late = an_id();
|
||||||
|
let mut state = test_writer_state(frame, true);
|
||||||
|
state.cycles_written = 2;
|
||||||
|
state.mic.samples = vec![8; 2 * frame];
|
||||||
|
state.mix.as_mut().unwrap().samples = vec![6; 2 * frame];
|
||||||
|
state.peers.insert(
|
||||||
|
early,
|
||||||
|
TestWriter {
|
||||||
|
samples: vec![1; 2 * frame],
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
|
let batch = test_batch(
|
||||||
|
vec![NewPeer {
|
||||||
|
id: late,
|
||||||
|
filename: "late.wav".to_string(),
|
||||||
|
}],
|
||||||
|
vec![9; frame],
|
||||||
|
None,
|
||||||
|
vec![(early, vec![2; frame]), (late, vec![7; frame])],
|
||||||
|
);
|
||||||
|
state
|
||||||
|
.apply_batch(&batch, |_| Ok(TestWriter::default()))
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
assert_eq!(state.cycles_written, 3);
|
||||||
|
assert_eq!(state.mic.samples.len(), 3 * frame);
|
||||||
|
assert_eq!(state.mix.as_ref().unwrap().samples.len(), 3 * frame);
|
||||||
|
assert_eq!(
|
||||||
|
&state.mix.as_ref().unwrap().samples[2 * frame..],
|
||||||
|
&[0, 0, 0]
|
||||||
|
);
|
||||||
|
assert_eq!(state.peers.get(&early).unwrap().samples.len(), 3 * frame);
|
||||||
|
assert_eq!(
|
||||||
|
&state.peers.get(&early).unwrap().samples[2 * frame..],
|
||||||
|
&[2, 2, 2]
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
state.peers.get(&late).unwrap().samples,
|
||||||
|
vec![0, 0, 0, 0, 0, 0, 7, 7, 7],
|
||||||
|
"late peer is back-padded by completed cycles before this batch"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn skipped_batches_keep_all_tracks_equal_length() {
|
||||||
|
let frame = 2;
|
||||||
|
let p1 = an_id();
|
||||||
|
let p2 = an_id();
|
||||||
|
let mut state = test_writer_state(frame, true);
|
||||||
|
|
||||||
|
let first = test_batch(
|
||||||
|
vec![
|
||||||
|
NewPeer {
|
||||||
|
id: p1,
|
||||||
|
filename: "p1.wav".to_string(),
|
||||||
|
},
|
||||||
|
NewPeer {
|
||||||
|
id: p2,
|
||||||
|
filename: "p2.wav".to_string(),
|
||||||
|
},
|
||||||
|
],
|
||||||
|
vec![1; frame],
|
||||||
|
Some(vec![5; frame]),
|
||||||
|
vec![(p1, vec![10; frame]), (p2, vec![20; frame])],
|
||||||
|
);
|
||||||
|
state
|
||||||
|
.apply_batch(&first, |_| Ok(TestWriter::default()))
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let _dropped_cycle = test_batch(
|
||||||
|
Vec::new(),
|
||||||
|
vec![2; frame],
|
||||||
|
Some(vec![6; frame]),
|
||||||
|
vec![(p1, vec![11; frame])],
|
||||||
|
);
|
||||||
|
|
||||||
|
let after_drop = test_batch(
|
||||||
|
Vec::new(),
|
||||||
|
vec![3; frame],
|
||||||
|
None,
|
||||||
|
vec![(p1, vec![12; frame])],
|
||||||
|
);
|
||||||
|
state
|
||||||
|
.apply_batch(&after_drop, |_| Ok(TestWriter::default()))
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let expected = 2 * frame;
|
||||||
|
assert_eq!(state.cycles_written, 2);
|
||||||
|
assert_eq!(state.mic.samples.len(), expected);
|
||||||
|
assert_eq!(state.mix.as_ref().unwrap().samples.len(), expected);
|
||||||
|
assert_eq!(state.peers.get(&p1).unwrap().samples.len(), expected);
|
||||||
|
assert_eq!(state.peers.get(&p2).unwrap().samples.len(), expected);
|
||||||
|
assert_eq!(
|
||||||
|
&state.peers.get(&p2).unwrap().samples[frame..],
|
||||||
|
&[0, 0],
|
||||||
|
"peer absent from an applied batch gets silence for that cycle"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn all_tracks_equal_length_after_n_cycles() {
|
fn all_tracks_equal_length_after_n_cycles() {
|
||||||
let dir = tmpdir("equal");
|
let dir = tmpdir("equal");
|
||||||
@@ -406,4 +731,23 @@ mod tests {
|
|||||||
"no mix track in stems-only mode"
|
"no mix track in stems-only mode"
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(unix)]
|
||||||
|
#[test]
|
||||||
|
fn async_peer_create_error_surfaces_at_finalize() {
|
||||||
|
use std::os::unix::fs::PermissionsExt;
|
||||||
|
|
||||||
|
let dir = tmpdir("asyncerr");
|
||||||
|
let mut rec = MultitrackRecorder::create(&dir, 4, false).unwrap();
|
||||||
|
std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o500)).unwrap();
|
||||||
|
|
||||||
|
rec.add_peer(an_id(), "blocked").unwrap();
|
||||||
|
rec.end_cycle().unwrap();
|
||||||
|
let result = rec.finalize();
|
||||||
|
|
||||||
|
std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700)).unwrap();
|
||||||
|
let err = result.unwrap_err();
|
||||||
|
assert_eq!(err.kind(), io::ErrorKind::PermissionDenied);
|
||||||
|
let _ = std::fs::remove_dir_all(&dir);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -20,6 +20,9 @@ pub trait AudioDecoder: Send {
|
|||||||
/// If `compressed` is `None` (or `Some(&[])`), it indicates packet loss,
|
/// If `compressed` is `None` (or `Some(&[])`), it indicates packet loss,
|
||||||
/// enabling the decoder to perform packet loss concealment (PLC).
|
/// enabling the decoder to perform packet loss concealment (PLC).
|
||||||
fn decode(&mut self, compressed: Option<&[u8]>) -> Result<Vec<i16>, CodecError>;
|
fn decode(&mut self, compressed: Option<&[u8]>) -> Result<Vec<i16>, CodecError>;
|
||||||
|
|
||||||
|
/// Reconstructs the previous lost frame from the next packet's in-band FEC.
|
||||||
|
fn decode_fec(&mut self, next_payload: &[u8]) -> Result<Vec<i16>, CodecError>;
|
||||||
}
|
}
|
||||||
|
|
||||||
pub mod opus_impl;
|
pub mod opus_impl;
|
||||||
|
|||||||
+120
-1
@@ -1,5 +1,49 @@
|
|||||||
use crate::codec::{AudioDecoder, AudioEncoder, CodecError};
|
use crate::codec::{AudioDecoder, AudioEncoder, CodecError};
|
||||||
use opus::{Application, Channels, Decoder, Encoder};
|
use crate::config::AudioProfile;
|
||||||
|
use opus::{Application, Bitrate, Channels, Decoder, Encoder};
|
||||||
|
|
||||||
|
/// Concrete libopus encoder settings derived from an [`AudioProfile`]. Plain
|
||||||
|
/// data, so the profile→params mapping ([`opus_params`]) stays a pure,
|
||||||
|
/// unit-testable function (W12).
|
||||||
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||||
|
pub struct OpusParams {
|
||||||
|
/// Target bitrate in bits/sec.
|
||||||
|
pub bitrate: i32,
|
||||||
|
/// Enable in-band forward error correction (loss redundancy in the bitstream).
|
||||||
|
pub inband_fec: bool,
|
||||||
|
/// Expected packet-loss percentage (0..=100); tunes how much FEC libopus adds.
|
||||||
|
pub packet_loss_perc: i32,
|
||||||
|
/// Discontinuous transmission: stop sending during silence to save bandwidth.
|
||||||
|
pub dtx: bool,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Map a named profile to concrete Opus parameters. Pure — the W12 testable seam.
|
||||||
|
///
|
||||||
|
/// `BadNetwork` deliberately runs a *lower* bitrate than `Balanced`: in-band FEC
|
||||||
|
/// redundancy is carried inside the same bitstream, so trimming the base bitrate
|
||||||
|
/// leaves headroom for the redundancy on a congested link.
|
||||||
|
pub fn opus_params(profile: AudioProfile) -> OpusParams {
|
||||||
|
match profile {
|
||||||
|
AudioProfile::LowLatency => OpusParams {
|
||||||
|
bitrate: 24_000,
|
||||||
|
inband_fec: false,
|
||||||
|
packet_loss_perc: 0,
|
||||||
|
dtx: false,
|
||||||
|
},
|
||||||
|
AudioProfile::Balanced => OpusParams {
|
||||||
|
bitrate: 32_000,
|
||||||
|
inband_fec: true,
|
||||||
|
packet_loss_perc: 10,
|
||||||
|
dtx: false,
|
||||||
|
},
|
||||||
|
AudioProfile::BadNetwork => OpusParams {
|
||||||
|
bitrate: 20_000,
|
||||||
|
inband_fec: true,
|
||||||
|
packet_loss_perc: 25,
|
||||||
|
dtx: false,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
pub struct OpusEncoder {
|
pub struct OpusEncoder {
|
||||||
encoder: Encoder,
|
encoder: Encoder,
|
||||||
@@ -17,6 +61,29 @@ impl OpusEncoder {
|
|||||||
.map_err(|e| CodecError::Init(format!("Failed to create Opus encoder: {}", e)))?;
|
.map_err(|e| CodecError::Init(format!("Failed to create Opus encoder: {}", e)))?;
|
||||||
Ok(Self { encoder })
|
Ok(Self { encoder })
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Apply concrete codec parameters to the live encoder. Safe to call between
|
||||||
|
/// frames, so the user can switch profile mid-call.
|
||||||
|
pub fn apply_params(&mut self, params: &OpusParams) -> Result<(), CodecError> {
|
||||||
|
self.encoder
|
||||||
|
.set_bitrate(Bitrate::Bits(params.bitrate))
|
||||||
|
.map_err(|e| CodecError::Init(format!("set_bitrate: {}", e)))?;
|
||||||
|
self.encoder
|
||||||
|
.set_inband_fec(params.inband_fec)
|
||||||
|
.map_err(|e| CodecError::Init(format!("set_inband_fec: {}", e)))?;
|
||||||
|
self.encoder
|
||||||
|
.set_packet_loss_perc(params.packet_loss_perc)
|
||||||
|
.map_err(|e| CodecError::Init(format!("set_packet_loss_perc: {}", e)))?;
|
||||||
|
self.encoder
|
||||||
|
.set_dtx(params.dtx)
|
||||||
|
.map_err(|e| CodecError::Init(format!("set_dtx: {}", e)))?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Apply a named [`AudioProfile`] (shorthand for `apply_params(&opus_params(p))`).
|
||||||
|
pub fn apply_profile(&mut self, profile: AudioProfile) -> Result<(), CodecError> {
|
||||||
|
self.apply_params(&opus_params(profile))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl AudioEncoder for OpusEncoder {
|
impl AudioEncoder for OpusEncoder {
|
||||||
@@ -95,12 +162,64 @@ impl AudioDecoder for OpusDecoder {
|
|||||||
pcm.truncate(decoded_per_channel * channels_count);
|
pcm.truncate(decoded_per_channel * channels_count);
|
||||||
Ok(pcm)
|
Ok(pcm)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn decode_fec(&mut self, next_payload: &[u8]) -> Result<Vec<i16>, CodecError> {
|
||||||
|
let channels_count = self.channels_count();
|
||||||
|
let mut pcm = vec![0i16; self.frame_samples * channels_count];
|
||||||
|
|
||||||
|
let decoded_per_channel = self
|
||||||
|
.decoder
|
||||||
|
.decode(next_payload, &mut pcm, true)
|
||||||
|
.map_err(|e| CodecError::Decode(format!("Opus FEC decoding failed: {}", e)))?;
|
||||||
|
|
||||||
|
pcm.truncate(decoded_per_channel * channels_count);
|
||||||
|
Ok(pcm)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_opus_params_mapping() {
|
||||||
|
let low = opus_params(AudioProfile::LowLatency);
|
||||||
|
let bal = opus_params(AudioProfile::Balanced);
|
||||||
|
let bad = opus_params(AudioProfile::BadNetwork);
|
||||||
|
|
||||||
|
// LowLatency has no loss redundancy; the other two do.
|
||||||
|
assert!(!low.inband_fec);
|
||||||
|
assert_eq!(low.packet_loss_perc, 0);
|
||||||
|
assert!(bal.inband_fec);
|
||||||
|
assert!(bad.inband_fec);
|
||||||
|
|
||||||
|
// Capture-side gating suppresses silence; no profile adds Opus DTX.
|
||||||
|
assert!(!low.dtx && !bal.dtx && !bad.dtx);
|
||||||
|
assert!(bad.packet_loss_perc > bal.packet_loss_perc);
|
||||||
|
|
||||||
|
// BadNetwork trims base bitrate to make room for FEC redundancy.
|
||||||
|
assert!(bad.bitrate < bal.bitrate);
|
||||||
|
|
||||||
|
// All bitrates are sane positive voice rates.
|
||||||
|
for p in [low, bal, bad] {
|
||||||
|
assert!(p.bitrate > 0 && p.bitrate <= 64_000);
|
||||||
|
assert!((0..=100).contains(&p.packet_loss_perc));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_apply_profile_sets_bitrate() {
|
||||||
|
let mut encoder = OpusEncoder::new(48000, Channels::Mono, Application::Voip).unwrap();
|
||||||
|
// Every profile applies cleanly to a real encoder...
|
||||||
|
for profile in AudioProfile::ALL {
|
||||||
|
encoder.apply_profile(profile).unwrap();
|
||||||
|
}
|
||||||
|
// ...and the last-applied bitrate is reflected by the encoder.
|
||||||
|
encoder.apply_profile(AudioProfile::Balanced).unwrap();
|
||||||
|
let want = opus_params(AudioProfile::Balanced).bitrate;
|
||||||
|
assert_eq!(encoder.encoder.get_bitrate().unwrap(), Bitrate::Bits(want));
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_round_trip() {
|
fn test_round_trip() {
|
||||||
let mut encoder = OpusEncoder::new(48000, Channels::Mono, Application::Voip).unwrap();
|
let mut encoder = OpusEncoder::new(48000, Channels::Mono, Application::Voip).unwrap();
|
||||||
|
|||||||
@@ -93,6 +93,60 @@ impl std::fmt::Display for RecordingMode {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Named Opus encoder / network-resilience policy (W12). The user picks a
|
||||||
|
/// profile instead of raw codec knobs; the concrete libopus parameters live in
|
||||||
|
/// `codec::opus_impl::opus_params`. Applies live to the running encoder.
|
||||||
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
|
||||||
|
pub enum AudioProfile {
|
||||||
|
/// Lowest mouth-to-ear delay: modest bitrate, no FEC redundancy. Best on a
|
||||||
|
/// clean LAN / low-loss link where added latency matters more than loss.
|
||||||
|
LowLatency,
|
||||||
|
/// Sensible default: voice bitrate with in-band FEC for light packet loss.
|
||||||
|
#[default]
|
||||||
|
Balanced,
|
||||||
|
/// Maximum resilience on a lossy/congested link: in-band FEC tuned for heavy
|
||||||
|
/// loss, at a lower bitrate to leave headroom for the redundancy.
|
||||||
|
BadNetwork,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl AudioProfile {
|
||||||
|
/// All variants, in picker display order.
|
||||||
|
pub const ALL: [AudioProfile; 3] = [
|
||||||
|
AudioProfile::LowLatency,
|
||||||
|
AudioProfile::Balanced,
|
||||||
|
AudioProfile::BadNetwork,
|
||||||
|
];
|
||||||
|
|
||||||
|
/// Compact discriminant for handing the profile to the capture thread via an
|
||||||
|
/// atomic. Pairs with [`AudioProfile::from_u8`].
|
||||||
|
pub fn as_u8(self) -> u8 {
|
||||||
|
match self {
|
||||||
|
AudioProfile::LowLatency => 0,
|
||||||
|
AudioProfile::Balanced => 1,
|
||||||
|
AudioProfile::BadNetwork => 2,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Inverse of [`AudioProfile::as_u8`]; unknown values fall back to the default.
|
||||||
|
pub fn from_u8(v: u8) -> AudioProfile {
|
||||||
|
match v {
|
||||||
|
0 => AudioProfile::LowLatency,
|
||||||
|
2 => AudioProfile::BadNetwork,
|
||||||
|
_ => AudioProfile::Balanced,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl std::fmt::Display for AudioProfile {
|
||||||
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
|
f.write_str(match self {
|
||||||
|
AudioProfile::LowLatency => "Low latency",
|
||||||
|
AudioProfile::Balanced => "Balanced",
|
||||||
|
AudioProfile::BadNetwork => "Bad network",
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
impl std::fmt::Display for RoomLayout {
|
impl std::fmt::Display for RoomLayout {
|
||||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
f.write_str(match self {
|
f.write_str(match self {
|
||||||
@@ -199,6 +253,10 @@ pub struct AppConfig {
|
|||||||
pub clip_volume_universal: bool,
|
pub clip_volume_universal: bool,
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
pub network_mode: NetworkMode,
|
pub network_mode: NetworkMode,
|
||||||
|
/// Opus encoder / network-resilience profile (W12). Applies live to the
|
||||||
|
/// running encoder; default `Balanced`.
|
||||||
|
#[serde(default)]
|
||||||
|
pub audio_profile: AudioProfile,
|
||||||
/// Presence posture for the friends idle listener (W7): invisible / normal /
|
/// Presence posture for the friends idle listener (W7): invisible / normal /
|
||||||
/// discoverable. Default `Normal` = answer friends only, no DNS beacon.
|
/// discoverable. Default `Normal` = answer friends only, no DNS beacon.
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
@@ -380,6 +438,7 @@ impl Default for AppConfig {
|
|||||||
show_player_bar: true,
|
show_player_bar: true,
|
||||||
clip_volume_universal: true,
|
clip_volume_universal: true,
|
||||||
network_mode: NetworkMode::default(),
|
network_mode: NetworkMode::default(),
|
||||||
|
audio_profile: AudioProfile::default(),
|
||||||
presence_mode: crate::presence::PresenceMode::default(),
|
presence_mode: crate::presence::PresenceMode::default(),
|
||||||
echo_cancellation_enabled: false,
|
echo_cancellation_enabled: false,
|
||||||
notifications_enabled: true,
|
notifications_enabled: true,
|
||||||
@@ -1019,6 +1078,38 @@ mod tests {
|
|||||||
assert_ne!(display_0, display_2);
|
assert_ne!(display_0, display_2);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_audio_profile() {
|
||||||
|
// Default is Balanced.
|
||||||
|
assert_eq!(AudioProfile::default(), AudioProfile::Balanced);
|
||||||
|
|
||||||
|
// ALL holds the three variants.
|
||||||
|
assert_eq!(AudioProfile::ALL.len(), 3);
|
||||||
|
assert!(AudioProfile::ALL.contains(&AudioProfile::LowLatency));
|
||||||
|
assert!(AudioProfile::ALL.contains(&AudioProfile::Balanced));
|
||||||
|
assert!(AudioProfile::ALL.contains(&AudioProfile::BadNetwork));
|
||||||
|
|
||||||
|
// as_u8 / from_u8 round-trip every variant, and unknown bytes fall back
|
||||||
|
// to the default rather than panicking.
|
||||||
|
for p in AudioProfile::ALL {
|
||||||
|
assert_eq!(AudioProfile::from_u8(p.as_u8()), p);
|
||||||
|
}
|
||||||
|
assert_eq!(AudioProfile::from_u8(99), AudioProfile::Balanced);
|
||||||
|
|
||||||
|
// serde round-trips, and Display strings are non-empty + distinct.
|
||||||
|
let mut labels = Vec::new();
|
||||||
|
for p in AudioProfile::ALL {
|
||||||
|
let s = serde_json::to_string(&p).unwrap();
|
||||||
|
assert_eq!(serde_json::from_str::<AudioProfile>(&s).unwrap(), p);
|
||||||
|
let label = p.to_string();
|
||||||
|
assert!(!label.is_empty());
|
||||||
|
labels.push(label);
|
||||||
|
}
|
||||||
|
labels.sort();
|
||||||
|
labels.dedup();
|
||||||
|
assert_eq!(labels.len(), 3);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_unknown_field_tolerance() {
|
fn test_unknown_field_tolerance() {
|
||||||
// Unknown/extra field tolerance: a config JSON containing an extra unrecognized key should still deserialize.
|
// Unknown/extra field tolerance: a config JSON containing an extra unrecognized key should still deserialize.
|
||||||
|
|||||||
+95
-5
@@ -202,11 +202,15 @@ impl JitterBuffer {
|
|||||||
None
|
None
|
||||||
} else {
|
} else {
|
||||||
// Gap with later packets already buffered: a packet was lost
|
// Gap with later packets already buffered: a packet was lost
|
||||||
// or reordered out of window. Conceal this frame via Opus PLC
|
// or reordered out of window. First try Opus in-band FEC from
|
||||||
// and grow the cushion — the jitter beat our current delay.
|
// the next packet; if unavailable, fall back to plain PLC.
|
||||||
self.next_seq = Some(next.wrapping_add(1));
|
self.next_seq = Some(next.wrapping_add(1));
|
||||||
self.note_disruption();
|
self.note_disruption();
|
||||||
self.decoder.decode(None).ok()
|
let next_payload = self.packets.values().next().expect("non-empty");
|
||||||
|
self.decoder
|
||||||
|
.decode_fec(next_payload)
|
||||||
|
.or_else(|_| self.decoder.decode(None))
|
||||||
|
.ok()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -221,8 +225,8 @@ impl JitterBuffer {
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
use crate::codec::AudioEncoder;
|
use crate::codec::opus_impl::{OpusDecoder, OpusEncoder, OpusParams};
|
||||||
use crate::codec::opus_impl::OpusEncoder;
|
use crate::codec::{AudioDecoder, AudioEncoder};
|
||||||
use opus::{Application, Channels};
|
use opus::{Application, Channels};
|
||||||
|
|
||||||
/// A real, decodable Opus packet for one 20ms mono frame at amplitude `amp`.
|
/// A real, decodable Opus packet for one 20ms mono frame at amplitude `amp`.
|
||||||
@@ -233,6 +237,32 @@ mod tests {
|
|||||||
enc.encode(&pcm).unwrap()
|
enc.encode(&pcm).unwrap()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn tone_frame(enc: &mut OpusEncoder, amp: i16, frame_index: usize) -> Vec<u8> {
|
||||||
|
let pcm: Vec<i16> = (0..FRAME_SAMPLES)
|
||||||
|
.map(|i| {
|
||||||
|
let sample_index = frame_index * FRAME_SAMPLES + i;
|
||||||
|
let t = sample_index as f32 / 48_000.0;
|
||||||
|
let fundamental = (t * 220.0 * 2.0 * std::f32::consts::PI).sin();
|
||||||
|
let harmonic = (t * 440.0 * 2.0 * std::f32::consts::PI).sin();
|
||||||
|
((fundamental * 0.7 + harmonic * 0.3) * amp as f32) as i16
|
||||||
|
})
|
||||||
|
.collect();
|
||||||
|
enc.encode(&pcm).unwrap()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn rms_error(a: &[i16], b: &[i16]) -> f64 {
|
||||||
|
assert_eq!(a.len(), b.len());
|
||||||
|
let sum_sq: f64 = a
|
||||||
|
.iter()
|
||||||
|
.zip(b)
|
||||||
|
.map(|(&left, &right)| {
|
||||||
|
let diff = left as f64 - right as f64;
|
||||||
|
diff * diff
|
||||||
|
})
|
||||||
|
.sum();
|
||||||
|
(sum_sq / a.len() as f64).sqrt()
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn buffers_then_plays_in_order() {
|
fn buffers_then_plays_in_order() {
|
||||||
let mut enc = OpusEncoder::new(48000, Channels::Mono, Application::Voip).unwrap();
|
let mut enc = OpusEncoder::new(48000, Channels::Mono, Application::Voip).unwrap();
|
||||||
@@ -289,6 +319,66 @@ mod tests {
|
|||||||
assert!(jb.pop_frame().is_none());
|
assert!(jb.pop_frame().is_none());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn uses_in_band_fec_from_next_packet_for_gap() {
|
||||||
|
let mut enc = OpusEncoder::new(48000, Channels::Mono, Application::Voip).unwrap();
|
||||||
|
enc.apply_params(&OpusParams {
|
||||||
|
bitrate: 20_000,
|
||||||
|
inband_fec: true,
|
||||||
|
packet_loss_perc: 60,
|
||||||
|
dtx: false,
|
||||||
|
})
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let dropped_seq = 5usize;
|
||||||
|
let amps = [1800, 1800, 1800, 1800, 1800, 12_000, 12_000, 12_000];
|
||||||
|
let packets: Vec<Vec<u8>> = amps
|
||||||
|
.into_iter()
|
||||||
|
.enumerate()
|
||||||
|
.map(|(seq, amp)| tone_frame(&mut enc, amp, seq))
|
||||||
|
.collect();
|
||||||
|
|
||||||
|
let mut expected_decoder = OpusDecoder::new(48000, Channels::Mono, FRAME_SAMPLES).unwrap();
|
||||||
|
for packet in packets.iter().take(dropped_seq) {
|
||||||
|
expected_decoder.decode(Some(packet)).unwrap();
|
||||||
|
}
|
||||||
|
let expected_lost = expected_decoder
|
||||||
|
.decode(Some(&packets[dropped_seq]))
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let mut plc_decoder = OpusDecoder::new(48000, Channels::Mono, FRAME_SAMPLES).unwrap();
|
||||||
|
for packet in packets.iter().take(dropped_seq) {
|
||||||
|
plc_decoder.decode(Some(packet)).unwrap();
|
||||||
|
}
|
||||||
|
let pure_plc = plc_decoder.decode(None).unwrap();
|
||||||
|
|
||||||
|
let mut jb = JitterBuffer::new().unwrap();
|
||||||
|
for (seq, packet) in packets.iter().enumerate() {
|
||||||
|
if seq != dropped_seq {
|
||||||
|
jb.insert(seq as u32, packet.clone());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for _ in 0..dropped_seq {
|
||||||
|
assert_eq!(jb.pop_frame().map(|frame| frame.len()), Some(FRAME_SAMPLES));
|
||||||
|
}
|
||||||
|
|
||||||
|
let recovered = jb.pop_frame().expect("gap should be reconstructed");
|
||||||
|
assert_eq!(recovered.len(), FRAME_SAMPLES);
|
||||||
|
assert!(
|
||||||
|
jb.packets.contains_key(&(dropped_seq as u32 + 1)),
|
||||||
|
"FEC source packet must remain buffered for normal decode"
|
||||||
|
);
|
||||||
|
assert_eq!(jb.pop_frame().map(|frame| frame.len()), Some(FRAME_SAMPLES));
|
||||||
|
|
||||||
|
let fec_error = rms_error(&recovered, &expected_lost);
|
||||||
|
let plc_error = rms_error(&pure_plc, &expected_lost);
|
||||||
|
assert!(
|
||||||
|
fec_error < plc_error * 0.75,
|
||||||
|
"FEC reconstruction should be materially closer than PLC (fec_error={fec_error}, plc_error={plc_error})"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn drops_packets_already_played() {
|
fn drops_packets_already_played() {
|
||||||
let mut enc = OpusEncoder::new(48000, Channels::Mono, Application::Voip).unwrap();
|
let mut enc = OpusEncoder::new(48000, Channels::Mono, Application::Voip).unwrap();
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
use crate::config::{NetworkMode, RecordingMode};
|
use crate::config::{AudioProfile, NetworkMode, RecordingMode};
|
||||||
use crate::friends::Friend;
|
use crate::friends::Friend;
|
||||||
use crate::network::PeerState;
|
use crate::network::PeerState;
|
||||||
use crate::presence::{FriendPresence, PresenceMode};
|
use crate::presence::{FriendPresence, PresenceMode};
|
||||||
@@ -56,6 +56,10 @@ pub enum CoreCommand {
|
|||||||
/// Set the relay/discovery posture. Takes effect on the next room join,
|
/// Set the relay/discovery posture. Takes effect on the next room join,
|
||||||
/// since the endpoint is (re)built then.
|
/// since the endpoint is (re)built then.
|
||||||
SetNetworkMode(NetworkMode),
|
SetNetworkMode(NetworkMode),
|
||||||
|
/// Set the Opus encoder / network-resilience profile (W12). Applies live to
|
||||||
|
/// the running capture encoder, and to the next call's encoder. Sent at
|
||||||
|
/// startup from config and whenever the user changes it.
|
||||||
|
SetAudioProfile(AudioProfile),
|
||||||
/// Start/stop recording the call to a local WAV (your mic + the incoming
|
/// Start/stop recording the call to a local WAV (your mic + the incoming
|
||||||
/// mix). No-op start if already recording / not in a call.
|
/// mix). No-op start if already recording / not in a call.
|
||||||
SetRecording(bool),
|
SetRecording(bool),
|
||||||
@@ -212,6 +216,7 @@ pub fn delivery_class(cmd: &CoreCommand) -> DeliveryClass {
|
|||||||
input_device: _,
|
input_device: _,
|
||||||
}
|
}
|
||||||
| CoreCommand::SetNetworkMode(_)
|
| CoreCommand::SetNetworkMode(_)
|
||||||
|
| CoreCommand::SetAudioProfile(_)
|
||||||
| CoreCommand::SetRecording(_)
|
| CoreCommand::SetRecording(_)
|
||||||
| CoreCommand::SetRecordingMode(_)
|
| CoreCommand::SetRecordingMode(_)
|
||||||
| CoreCommand::SendChat(_)
|
| CoreCommand::SendChat(_)
|
||||||
@@ -292,6 +297,7 @@ pub fn coalesce_key(cmd: &CoreCommand) -> Option<CoalesceKey> {
|
|||||||
input_device: _,
|
input_device: _,
|
||||||
}
|
}
|
||||||
| CoreCommand::SetNetworkMode(_)
|
| CoreCommand::SetNetworkMode(_)
|
||||||
|
| CoreCommand::SetAudioProfile(_)
|
||||||
| CoreCommand::SetRecording(_)
|
| CoreCommand::SetRecording(_)
|
||||||
| CoreCommand::SetRecordingMode(_)
|
| CoreCommand::SetRecordingMode(_)
|
||||||
| CoreCommand::SendChat(_)
|
| CoreCommand::SendChat(_)
|
||||||
@@ -574,6 +580,7 @@ mod tests {
|
|||||||
},
|
},
|
||||||
CoreCommand::SetPeerMuted(peer, true),
|
CoreCommand::SetPeerMuted(peer, true),
|
||||||
CoreCommand::SetPresenceMode(PresenceMode::Normal),
|
CoreCommand::SetPresenceMode(PresenceMode::Normal),
|
||||||
|
CoreCommand::SetAudioProfile(crate::config::AudioProfile::BadNetwork),
|
||||||
CoreCommand::SendChat("hello".to_string()),
|
CoreCommand::SendChat("hello".to_string()),
|
||||||
];
|
];
|
||||||
|
|
||||||
|
|||||||
+88
-6
@@ -17,7 +17,7 @@ use crate::network::{
|
|||||||
};
|
};
|
||||||
|
|
||||||
use crate::audio::multitrack::MultitrackRecorder;
|
use crate::audio::multitrack::MultitrackRecorder;
|
||||||
use crate::config::{NetworkMode, RecordingMode};
|
use crate::config::{AudioProfile, NetworkMode, RecordingMode};
|
||||||
use crate::presence::PresenceMode;
|
use crate::presence::PresenceMode;
|
||||||
use iroh::{
|
use iroh::{
|
||||||
Endpoint, EndpointAddr, EndpointId, RelayMode, SecretKey, endpoint::presets, protocol::Router,
|
Endpoint, EndpointAddr, EndpointId, RelayMode, SecretKey, endpoint::presets, protocol::Router,
|
||||||
@@ -31,6 +31,12 @@ use tokio::sync::{Mutex, mpsc};
|
|||||||
|
|
||||||
type CoalesceStore = Arc<StdMutex<HashMap<CoalesceKey, CoreCommand>>>;
|
type CoalesceStore = Arc<StdMutex<HashMap<CoalesceKey, CoreCommand>>>;
|
||||||
|
|
||||||
|
// Mixer -> playback-worker handoff. The playback ring itself targets three
|
||||||
|
// 20ms frames; allow at most two more in flight so worker lag applies
|
||||||
|
// backpressure before the ring can overshoot to its 200ms cap (A6).
|
||||||
|
const PLAYBACK_HANDOFF_QUEUE_FRAMES: usize = 2;
|
||||||
|
const PLAYBACK_HANDOFF_RETRY: Duration = Duration::from_millis(1);
|
||||||
|
|
||||||
pub struct CoreController {
|
pub struct CoreController {
|
||||||
reliable_tx: mpsc::UnboundedSender<CoreCommand>,
|
reliable_tx: mpsc::UnboundedSender<CoreCommand>,
|
||||||
coalesce: CoalesceStore,
|
coalesce: CoalesceStore,
|
||||||
@@ -156,6 +162,22 @@ fn audio_datagram_len_ok(len: usize) -> bool {
|
|||||||
(4..=4 + MAX_OPUS_PAYLOAD).contains(&len)
|
(4..=4 + MAX_OPUS_PAYLOAD).contains(&len)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn send_playback_frame(
|
||||||
|
tx: &std::sync::mpsc::SyncSender<Vec<i16>>,
|
||||||
|
mut frame: Vec<i16>,
|
||||||
|
) -> bool {
|
||||||
|
loop {
|
||||||
|
match tx.try_send(frame) {
|
||||||
|
Ok(()) => return true,
|
||||||
|
Err(std::sync::mpsc::TrySendError::Full(returned)) => {
|
||||||
|
frame = returned;
|
||||||
|
tokio::time::sleep(PLAYBACK_HANDOFF_RETRY).await;
|
||||||
|
}
|
||||||
|
Err(std::sync::mpsc::TrySendError::Disconnected(_)) => return false,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// The presence label to broadcast for a detected game: its display name,
|
/// The presence label to broadcast for a detected game: its display name,
|
||||||
/// sanitized + length-capped, or `None` when there's no game or no broadcastable
|
/// sanitized + length-capped, or `None` when there's no game or no broadcastable
|
||||||
/// name (a Steam appid without a manifest name, or a label that sanitizes empty).
|
/// name (a Steam appid without a manifest name, or a label that sanitizes empty).
|
||||||
@@ -1162,6 +1184,12 @@ async fn run_core_loop(
|
|||||||
// App-internal capture/playback gains (f32 bits), live-read by the audio loops.
|
// App-internal capture/playback gains (f32 bits), live-read by the audio loops.
|
||||||
let input_gain = Arc::new(std::sync::atomic::AtomicU32::new(1.0f32.to_bits()));
|
let input_gain = Arc::new(std::sync::atomic::AtomicU32::new(1.0f32.to_bits()));
|
||||||
let output_gain = Arc::new(std::sync::atomic::AtomicU32::new(1.0f32.to_bits()));
|
let output_gain = Arc::new(std::sync::atomic::AtomicU32::new(1.0f32.to_bits()));
|
||||||
|
// Opus encoder profile (W12) as a discriminant, live-read by the capture
|
||||||
|
// thread so a mid-call profile switch re-tunes the running encoder. Set from
|
||||||
|
// config via the GUI's startup `SetAudioProfile`; defaults to Balanced.
|
||||||
|
let audio_profile = Arc::new(std::sync::atomic::AtomicU8::new(
|
||||||
|
AudioProfile::default().as_u8(),
|
||||||
|
));
|
||||||
// Call recording: an optional live recorder (mic FIFO + WAV writer), shared
|
// Call recording: an optional live recorder (mic FIFO + WAV writer), shared
|
||||||
// by the capture thread (pushes mic) and the mixer task (writes mix frames).
|
// by the capture thread (pushes mic) and the mixer task (writes mix frames).
|
||||||
// `is_recording` is a fast-path gate so the audio loops only take the lock
|
// `is_recording` is a fast-path gate so the audio loops only take the lock
|
||||||
@@ -1648,7 +1676,8 @@ async fn run_core_loop(
|
|||||||
|
|
||||||
// Setup raw audio channels
|
// Setup raw audio channels
|
||||||
let (capture_tx, capture_rx) = std::sync::mpsc::channel();
|
let (capture_tx, capture_rx) = std::sync::mpsc::channel();
|
||||||
let (playback_tx, playback_rx) = std::sync::mpsc::channel();
|
let (playback_tx, playback_rx) =
|
||||||
|
std::sync::mpsc::sync_channel(PLAYBACK_HANDOFF_QUEUE_FRAMES);
|
||||||
|
|
||||||
// Echo cancellation: if enabled, load PipeWire's echo-cancel module
|
// Echo cancellation: if enabled, load PipeWire's echo-cancel module
|
||||||
// bound to the chosen real devices and route capture/playback
|
// bound to the chosen real devices and route capture/playback
|
||||||
@@ -1740,6 +1769,7 @@ async fn run_core_loop(
|
|||||||
let is_recording_capture = is_recording.clone();
|
let is_recording_capture = is_recording.clone();
|
||||||
let multitrack_capture = multitrack.clone();
|
let multitrack_capture = multitrack.clone();
|
||||||
let is_multitrack_capture = is_multitrack.clone();
|
let is_multitrack_capture = is_multitrack.clone();
|
||||||
|
let audio_profile_capture = audio_profile.clone();
|
||||||
|
|
||||||
let capture_thread = std::thread::spawn(move || {
|
let capture_thread = std::thread::spawn(move || {
|
||||||
use opus::{Application, Channels};
|
use opus::{Application, Channels};
|
||||||
@@ -1751,6 +1781,13 @@ async fn run_core_loop(
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
// Tune the encoder to the configured profile (W12), then track
|
||||||
|
// the live discriminant so a mid-call switch re-applies it.
|
||||||
|
let mut current_profile =
|
||||||
|
AudioProfile::from_u8(audio_profile_capture.load(Ordering::Relaxed));
|
||||||
|
if let Err(e) = encoder.apply_profile(current_profile) {
|
||||||
|
crate::log_msg(&format!("Opus profile apply failed: {:?}", e));
|
||||||
|
}
|
||||||
// Per-sender packet sequence number, prepended to every frame so
|
// Per-sender packet sequence number, prepended to every frame so
|
||||||
// receivers can reorder and conceal loss. Wraps after ~years.
|
// receivers can reorder and conceal loss. Wraps after ~years.
|
||||||
let mut seq: u32 = 0;
|
let mut seq: u32 = 0;
|
||||||
@@ -1763,6 +1800,14 @@ async fn run_core_loop(
|
|||||||
let mut mic_meter = MicLevelMeter::new();
|
let mut mic_meter = MicLevelMeter::new();
|
||||||
|
|
||||||
while let Ok(mut pcm) = capture_rx.recv() {
|
while let Ok(mut pcm) = capture_rx.recv() {
|
||||||
|
// Re-tune the encoder if the user switched profile mid-call.
|
||||||
|
// Cheap atomic load per frame; only reconfigures on change.
|
||||||
|
let want =
|
||||||
|
AudioProfile::from_u8(audio_profile_capture.load(Ordering::Relaxed));
|
||||||
|
if want != current_profile && encoder.apply_profile(want).is_ok() {
|
||||||
|
current_profile = want;
|
||||||
|
}
|
||||||
|
|
||||||
// Apply the input gain first so the meter, gate, and what we
|
// Apply the input gain first so the meter, gate, and what we
|
||||||
// transmit all reflect the same (gained) signal.
|
// transmit all reflect the same (gained) signal.
|
||||||
apply_volume(
|
apply_volume(
|
||||||
@@ -2092,7 +2137,7 @@ async fn run_core_loop(
|
|||||||
mixed
|
mixed
|
||||||
};
|
};
|
||||||
|
|
||||||
if playback_tx.send(frame_to_send).is_err() {
|
if !send_playback_frame(&playback_tx, frame_to_send).await {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -2682,6 +2727,13 @@ async fn run_core_loop(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
CoreCommand::SetAudioProfile(profile) => {
|
||||||
|
// Publish the new profile to the capture thread (W12). It picks up
|
||||||
|
// the change on its next frame and re-tunes the live encoder; a
|
||||||
|
// call that starts later reads the same atomic at encoder creation.
|
||||||
|
audio_profile.store(profile.as_u8(), Ordering::Relaxed);
|
||||||
|
}
|
||||||
|
|
||||||
CoreCommand::RegenerateIdentity => {
|
CoreCommand::RegenerateIdentity => {
|
||||||
// Mint + persist a fresh identity, discarding the old one. The
|
// Mint + persist a fresh identity, discarding the old one. The
|
||||||
// persistent endpoint is rebuilt with the new key (now if idle, else
|
// persistent endpoint is rebuilt with the new key (now if idle, else
|
||||||
@@ -3193,12 +3245,15 @@ async fn run_core_loop(
|
|||||||
mod tests {
|
mod tests {
|
||||||
use super::{
|
use super::{
|
||||||
KnownPeers, MAX_OPUS_PAYLOAD, MAX_RETAINED_PEERS, MIC_LEVEL_REPORT_SAMPLES, MicLevelMeter,
|
KnownPeers, MAX_OPUS_PAYLOAD, MAX_RETAINED_PEERS, MIC_LEVEL_REPORT_SAMPLES, MicLevelMeter,
|
||||||
PeerSpeakTicket, admit_retained, apply_peer_volume, apply_volume, audio_datagram_len_ok,
|
PLAYBACK_HANDOFF_QUEUE_FRAMES, PeerSpeakTicket, admit_retained, apply_peer_volume,
|
||||||
coalesce_insert, coalesce_pop, frame_level, mix_frames, mix_stereo_frames,
|
apply_volume, audio_datagram_len_ok, coalesce_insert, coalesce_pop, frame_level,
|
||||||
next_game_change, should_auto_fetch, stereo_to_mono,
|
mix_frames, mix_stereo_frames, next_game_change, send_playback_frame, should_auto_fetch,
|
||||||
|
stereo_to_mono,
|
||||||
};
|
};
|
||||||
use crate::core::messages::{CoalesceKey, CoreCommand, coalesce_key};
|
use crate::core::messages::{CoalesceKey, CoreCommand, coalesce_key};
|
||||||
use std::collections::{HashMap, HashSet};
|
use std::collections::{HashMap, HashSet};
|
||||||
|
use std::sync::mpsc::sync_channel;
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
fn endpoint_id() -> iroh::EndpointId {
|
fn endpoint_id() -> iroh::EndpointId {
|
||||||
iroh::SecretKey::generate().public()
|
iroh::SecretKey::generate().public()
|
||||||
@@ -3432,6 +3487,33 @@ mod tests {
|
|||||||
assert!(!audio_datagram_len_ok(5 + MAX_OPUS_PAYLOAD));
|
assert!(!audio_datagram_len_ok(5 + MAX_OPUS_PAYLOAD));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn playback_handoff_waits_for_bounded_queue_space() {
|
||||||
|
let (tx, rx) = sync_channel::<Vec<i16>>(PLAYBACK_HANDOFF_QUEUE_FRAMES);
|
||||||
|
for n in 0..PLAYBACK_HANDOFF_QUEUE_FRAMES {
|
||||||
|
tx.try_send(vec![n as i16]).unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
let worker = std::thread::spawn(move || {
|
||||||
|
std::thread::sleep(Duration::from_millis(20));
|
||||||
|
for n in 0..PLAYBACK_HANDOFF_QUEUE_FRAMES {
|
||||||
|
assert_eq!(rx.recv().unwrap(), vec![n as i16]);
|
||||||
|
}
|
||||||
|
assert_eq!(rx.recv().unwrap(), vec![99, 100]);
|
||||||
|
});
|
||||||
|
|
||||||
|
let sent = tokio::time::timeout(
|
||||||
|
Duration::from_secs(1),
|
||||||
|
send_playback_frame(&tx, vec![99, 100]),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("bounded handoff should unblock after the worker drains a frame");
|
||||||
|
|
||||||
|
assert!(sent);
|
||||||
|
drop(tx);
|
||||||
|
worker.join().unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn mic_meter_holds_the_peak_across_the_window() {
|
fn mic_meter_holds_the_peak_across_the_window() {
|
||||||
let mut m = MicLevelMeter::new();
|
let mut m = MicLevelMeter::new();
|
||||||
|
|||||||
Reference in New Issue
Block a user