molluskandClaude Opus 4.8
a6aca73c67
Add room text chat riding the existing iroh-gossip topic (same layer as the
presence roster). New GossipMessage::Chat { name, text, ts }; the gossip loop
forwards it as RoomEvent::ChatMessage, core relays it to the UI as
UiEvent::ChatMessage, and RoomState::send_chat broadcasts an authored line
(display name from self-state, ms timestamp). CoreCommand::SendChat sends; our
own author is suppressed by the existing self-echo guard, so the UI echoes our
sent line locally instead.
UI: a full-width chat dock along the bottom of the room (the chosen layout) —
bottom-anchored scrollback with per-sender name colouring (green = you), an
input with Enter-to-send + a Send button, history capped at 300 lines. The room
window default grows to 900x760 so the dock doesn't squeeze the controls column.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1254 lines
57 KiB
Rust
1254 lines
57 KiB
Rust
pub mod messages;
|
|
pub mod jitter;
|
|
|
|
use crate::audio::{AudioBackend, pipewire_impl::PipeWireBackend};
|
|
use crate::codec::{AudioEncoder, opus_impl::OpusEncoder};
|
|
use crate::core::jitter::{JitterBuffer, FRAME_SAMPLES};
|
|
use crate::network::{
|
|
NetworkTransport, RoomState, PeerState, RoomEvent, ConnEvent, PeerSpeakTicket,
|
|
iroh_impl::IrohTransport,
|
|
gossip::IrohGossipState,
|
|
};
|
|
use crate::core::messages::{CoreCommand, UiEvent};
|
|
|
|
use crate::config::NetworkMode;
|
|
use iroh::{Endpoint, EndpointId, RelayMode, endpoint::presets, protocol::Router};
|
|
use iroh_gossip::net::Gossip;
|
|
use tokio::sync::{mpsc, Mutex};
|
|
use std::collections::{HashMap, HashSet};
|
|
use std::sync::Arc;
|
|
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
|
|
use std::time::Duration;
|
|
|
|
pub struct CoreController {
|
|
cmd_tx: mpsc::Sender<CoreCommand>,
|
|
}
|
|
|
|
impl CoreController {
|
|
pub fn new(ui_tx: mpsc::Sender<UiEvent>) -> Self {
|
|
let (cmd_tx, cmd_rx) = mpsc::channel(100);
|
|
|
|
std::thread::spawn(move || {
|
|
let rt = tokio::runtime::Runtime::new().expect("Failed to create Tokio runtime");
|
|
rt.block_on(async move {
|
|
crate::log_msg("Starting core network loop in dedicated Tokio runtime");
|
|
if let Err(e) = run_core_loop(cmd_rx, ui_tx).await {
|
|
crate::log_msg(&format!("App core loop failed: {:?}", e));
|
|
}
|
|
});
|
|
});
|
|
|
|
Self { cmd_tx }
|
|
}
|
|
|
|
pub fn send(&self, cmd: CoreCommand) -> Result<(), mpsc::error::TrySendError<CoreCommand>> {
|
|
self.cmd_tx.try_send(cmd)
|
|
}
|
|
}
|
|
|
|
/// How long a peer may stay "reconnecting" after a transient drop before we give
|
|
/// up and evict it. Comfortably past the QUIC idle timeout (~30s) so a genuine
|
|
/// reconnect has time to complete, but short enough that a crashed/departed peer
|
|
/// clears from the room promptly.
|
|
const RECONNECT_GRACE: Duration = Duration::from_secs(45);
|
|
|
|
/// Per-peer reconnect grace timers (see [`RECONNECT_GRACE`]). Shared between the
|
|
/// room-event task (which arms one on a transient drop and cancels it on a
|
|
/// gossip rejoin) and the conn-event task (which cancels it when the audio link
|
|
/// actually comes back).
|
|
type GraceTimers = Arc<std::sync::Mutex<HashMap<EndpointId, tokio::task::JoinHandle<()>>>>;
|
|
|
|
/// Peers we've completed at least one audio link with. Lets the conn-event task
|
|
/// tell a genuine reconnect (arm an eviction timer) from a first-ever dial (don't).
|
|
/// Scrubbed whenever a peer is evicted or leaves so a later rejoin starts clean.
|
|
type SeenConnected = Arc<std::sync::Mutex<HashSet<EndpointId>>>;
|
|
|
|
/// Cancel and forget a peer's pending grace timer, if any. No-op if none is armed.
|
|
fn cancel_grace_timer(timers: &GraceTimers, peer_id: &EndpointId) {
|
|
if let Some(handle) = timers.lock().unwrap().remove(peer_id) {
|
|
handle.abort();
|
|
}
|
|
}
|
|
|
|
/// Arm a per-peer reconnect grace timer that evicts the peer if its link hasn't
|
|
/// recovered within [`RECONNECT_GRACE`]. No-op if a timer is already pending for
|
|
/// the peer, so the earliest drop notice — whether the gossip `PeerConnectionLost`
|
|
/// or the transport `Connecting` — sets one hard deadline, rather than a flapping
|
|
/// link repeatedly resetting the clock and dodging eviction forever. On firing it
|
|
/// also scrubs the peer from `seen_connected` so a later rejoin isn't treated as a
|
|
/// reconnect on its initial dial.
|
|
fn arm_grace_timer(
|
|
timers: &GraceTimers,
|
|
seen_connected: &SeenConnected,
|
|
transport: &Arc<IrohTransport>,
|
|
jitter: &Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
|
|
ui_tx: &mpsc::Sender<UiEvent>,
|
|
grace: Duration,
|
|
peer_id: EndpointId,
|
|
) {
|
|
let mut timers_guard = timers.lock().unwrap();
|
|
if timers_guard.contains_key(&peer_id) {
|
|
return;
|
|
}
|
|
let transport_evict = transport.clone();
|
|
let jitter_evict = jitter.clone();
|
|
let ui_evict = ui_tx.clone();
|
|
let timers_evict = timers.clone();
|
|
let seen_evict = seen_connected.clone();
|
|
let handle = tokio::spawn(async move {
|
|
tokio::time::sleep(grace).await;
|
|
crate::log_msg(&format!("Reconnect grace expired; evicting peer {:?}", peer_id));
|
|
transport_evict.disconnect_peer(peer_id).await;
|
|
jitter_evict.lock().await.remove(&peer_id);
|
|
// Scrub our internal state *before* announcing the eviction, so anything
|
|
// that observes `PeerConnectionFailed` (or a rejoin racing it) sees a clean
|
|
// slate — a later dial for this identity is then a fresh first-dial, not a
|
|
// reconnect.
|
|
timers_evict.lock().unwrap().remove(&peer_id);
|
|
seen_evict.lock().unwrap().remove(&peer_id);
|
|
let _ = ui_evict.send(UiEvent::PeerConnectionFailed { id: peer_id }).await;
|
|
});
|
|
timers_guard.insert(peer_id, handle);
|
|
}
|
|
|
|
/// Scale a frame in place by a per-peer volume factor, saturating to the i16
|
|
/// range. A volume within `f32::EPSILON` of 1.0 is treated as unity and skipped,
|
|
/// matching the mixer hot path that avoids touching unmodified frames.
|
|
fn apply_volume(frame: &mut [i16], vol: f32) {
|
|
if (vol - 1.0).abs() <= f32::EPSILON {
|
|
return;
|
|
}
|
|
for sample in frame.iter_mut() {
|
|
*sample = (*sample as f32 * vol).clamp(i16::MIN as f32, i16::MAX as f32) as i16;
|
|
}
|
|
}
|
|
|
|
/// Normalized RMS level of a frame in `[0.0, 1.0]` (32768 = full scale), for the
|
|
/// UI level meter. An empty frame reads as 0.0.
|
|
fn frame_level(frame: &[i16]) -> f32 {
|
|
let sum_sq: f32 = frame.iter().map(|&x| (x as f32).powi(2)).sum();
|
|
let rms = (sum_sq / frame.len().max(1) as f32).sqrt();
|
|
(rms / 32768.0).clamp(0.0, 1.0)
|
|
}
|
|
|
|
/// Peak-hold every this many captured samples (~100ms @ 48kHz) before emitting a
|
|
/// [`UiEvent::MicLevel`], so the meter doesn't flood the UI runtime at frame rate.
|
|
const MIC_LEVEL_REPORT_SAMPLES: usize = 4800;
|
|
|
|
/// Peak-holds the raw mic level across captured frames and yields a value to
|
|
/// report roughly every [`MIC_LEVEL_REPORT_SAMPLES`] samples. Shared by the
|
|
/// in-call capture thread and the standalone monitor so both throttle and
|
|
/// peak-hold identically.
|
|
struct MicLevelMeter {
|
|
peak: f32,
|
|
acc: usize,
|
|
}
|
|
|
|
impl MicLevelMeter {
|
|
fn new() -> Self {
|
|
Self { peak: 0.0, acc: 0 }
|
|
}
|
|
|
|
/// Folds one frame into the running peak. Returns `Some(peak)` (and resets)
|
|
/// once enough samples have accumulated to emit a reading, else `None`.
|
|
fn push(&mut self, frame: &[i16]) -> Option<f32> {
|
|
self.peak = self.peak.max(frame_level(frame));
|
|
self.acc += frame.len();
|
|
if self.acc >= MIC_LEVEL_REPORT_SAMPLES {
|
|
let peak = self.peak;
|
|
self.peak = 0.0;
|
|
self.acc = 0;
|
|
Some(peak)
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
}
|
|
|
|
/// A standalone, capture-only mic monitor for gate calibration outside a call.
|
|
/// Owns the worker thread that reads raw PCM and reports its level; the PipeWire
|
|
/// capture stream itself lives in the shared backend. Tear down by stopping the
|
|
/// backend's capture (which closes the channel) and joining this thread.
|
|
struct MicMonitor {
|
|
thread: std::thread::JoinHandle<()>,
|
|
}
|
|
|
|
/// Drains a capture channel, reporting the raw (un-gated) mic level to the UI.
|
|
/// Returns when the channel closes (i.e. the backend's capture stream stopped).
|
|
fn run_mic_monitor(
|
|
rx: std::sync::mpsc::Receiver<Vec<i16>>,
|
|
ui_tx: mpsc::Sender<UiEvent>,
|
|
input_gain: Arc<std::sync::atomic::AtomicU32>,
|
|
) {
|
|
let mut meter = MicLevelMeter::new();
|
|
while let Ok(mut pcm) = rx.recv() {
|
|
// Mirror the in-call path: apply the input gain before metering so the
|
|
// test meter reflects the gained signal (and the input slider moves it).
|
|
apply_volume(&mut pcm, f32::from_bits(input_gain.load(Ordering::Relaxed)));
|
|
if let Some(peak) = meter.push(&pcm) {
|
|
// Drop on a full channel — a stale meter reading is harmless.
|
|
let _ = ui_tx.try_send(UiEvent::MicLevel(peak));
|
|
}
|
|
}
|
|
// Channel closed: the monitor was stopped. Snap the meter back to zero.
|
|
let _ = ui_tx.try_send(UiEvent::MicLevel(0.0));
|
|
}
|
|
|
|
/// Stops a standalone mic monitor if one is running. MUST NOT be called while a
|
|
/// room session is active — `backend.stop()` would also tear down the call's
|
|
/// capture/playback. Monitor and session are mutually exclusive by construction.
|
|
fn stop_mic_monitor(backend: &PipeWireBackend, monitor: Option<MicMonitor>) {
|
|
if let Some(m) = monitor {
|
|
let _ = backend.stop();
|
|
let _ = m.thread.join();
|
|
}
|
|
}
|
|
|
|
/// Sum per-peer frames sample-by-sample into one `frame_len`-sample bus, **without**
|
|
/// clamping — the lossless `i32` sum preserves the true peak so the mix-bus soft
|
|
/// limiter (see [`crate::audio::limiter`]) can ride it down to the ceiling instead
|
|
/// of the old hard clip shattering loud moments. Peers shorter than `frame_len`
|
|
/// contribute 0 past their end; an empty peer set yields a silent bus.
|
|
fn mix_frames(peer_frames: &[Vec<i16>], frame_len: usize) -> Vec<i32> {
|
|
let mut mixed = vec![0i32; frame_len];
|
|
for frame in peer_frames {
|
|
for (out, &sample) in mixed.iter_mut().zip(frame.iter()) {
|
|
*out += sample as i32;
|
|
}
|
|
}
|
|
mixed
|
|
}
|
|
|
|
/// Handles the transport's per-peer link-state stream (`ConnEvent`): arms/cancels
|
|
/// reconnect grace timers, tracks which peers we've linked with, and forwards
|
|
/// link state to the UI. Pulled out of the conn-event task as a unit so the
|
|
/// reconnect-eviction behavior can be tested without standing up a full session.
|
|
/// Exposed (with a `grace` override) for that purpose; not part of the public API.
|
|
pub struct ConnEventHandler {
|
|
ui_tx: mpsc::Sender<UiEvent>,
|
|
grace_timers: GraceTimers,
|
|
seen_connected: SeenConnected,
|
|
transport: Arc<IrohTransport>,
|
|
jitter: Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
|
|
grace: Duration,
|
|
}
|
|
|
|
impl ConnEventHandler {
|
|
pub fn new(
|
|
ui_tx: mpsc::Sender<UiEvent>,
|
|
grace_timers: GraceTimers,
|
|
seen_connected: SeenConnected,
|
|
transport: Arc<IrohTransport>,
|
|
jitter: Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
|
|
) -> Self {
|
|
Self {
|
|
ui_tx,
|
|
grace_timers,
|
|
seen_connected,
|
|
transport,
|
|
jitter,
|
|
grace: RECONNECT_GRACE,
|
|
}
|
|
}
|
|
|
|
/// Override the eviction grace window. For tests that can't wait 45s.
|
|
pub fn with_grace(mut self, grace: Duration) -> Self {
|
|
self.grace = grace;
|
|
self
|
|
}
|
|
|
|
pub async fn handle(&self, event: ConnEvent) {
|
|
match event {
|
|
ConnEvent::Connecting(id) => {
|
|
// A reconnect (we've linked with this peer before): arm an eviction
|
|
// timer so a peer that never comes back is cleared even when gossip
|
|
// doesn't re-report the drop — the transport reliably re-emits this
|
|
// on every outage, gossip's NeighborDown does not. A first-ever dial
|
|
// (not yet in seen_connected) gets no timer; Connected cancels it on
|
|
// recovery.
|
|
if self.seen_connected.lock().unwrap().contains(&id) {
|
|
arm_grace_timer(
|
|
&self.grace_timers,
|
|
&self.seen_connected,
|
|
&self.transport,
|
|
&self.jitter,
|
|
&self.ui_tx,
|
|
self.grace,
|
|
id,
|
|
);
|
|
}
|
|
let _ = self.ui_tx.send(UiEvent::PeerConnecting { id }).await;
|
|
}
|
|
ConnEvent::Connected(id) => {
|
|
// The audio link came back — the peer recovered within the grace
|
|
// window, so cancel its eviction.
|
|
cancel_grace_timer(&self.grace_timers, &id);
|
|
self.seen_connected.lock().unwrap().insert(id);
|
|
let _ = self.ui_tx.send(UiEvent::PeerConnected { id }).await;
|
|
}
|
|
ConnEvent::Left(id) => {
|
|
// The peer closed its link gracefully (intentional leave) — evict
|
|
// immediately, like a PeerLeft, instead of leaving it "reconnecting"
|
|
// until the grace timer or the slow gossip Leave.
|
|
cancel_grace_timer(&self.grace_timers, &id);
|
|
self.seen_connected.lock().unwrap().remove(&id);
|
|
self.transport.disconnect_peer(id).await;
|
|
self.jitter.lock().await.remove(&id);
|
|
let _ = self.ui_tx.send(UiEvent::PeerLeft { id }).await;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
struct ActiveSession {
|
|
endpoint: Endpoint,
|
|
router: Router,
|
|
room_state: Arc<IrohGossipState>,
|
|
capture_thread: std::thread::JoinHandle<()>,
|
|
datagram_task: tokio::task::JoinHandle<()>,
|
|
mixer_task: tokio::task::JoinHandle<()>,
|
|
event_task: tokio::task::JoinHandle<()>,
|
|
conn_event_task: tokio::task::JoinHandle<()>,
|
|
grace_timers: GraceTimers,
|
|
transport: Arc<IrohTransport>,
|
|
/// Loaded PipeWire echo-cancel module (if enabled); unloads on drop.
|
|
echo_cancel: Option<crate::audio::echo_cancel::EchoCancelGuard>,
|
|
}
|
|
|
|
impl ActiveSession {
|
|
async fn shutdown(self, audio_backend: Arc<PipeWireBackend>) {
|
|
crate::log_msg("ActiveSession::shutdown started");
|
|
self.datagram_task.abort();
|
|
self.mixer_task.abort();
|
|
self.event_task.abort();
|
|
self.conn_event_task.abort();
|
|
// Abort any pending reconnect grace timers so they can't fire a stray
|
|
// eviction (or touch a torn-down transport) after the session is gone.
|
|
for (_, handle) in self.grace_timers.lock().unwrap().drain() {
|
|
handle.abort();
|
|
}
|
|
crate::log_msg("Aborted tasks");
|
|
|
|
let audio_backend_clone = audio_backend.clone();
|
|
let _ = tokio::task::spawn_blocking(move || {
|
|
crate::log_msg("Stopping audio backend...");
|
|
let _ = audio_backend_clone.stop();
|
|
crate::log_msg("Audio backend stopped");
|
|
}).await;
|
|
|
|
// Unload the echo-cancel module now that the audio streams releasing its
|
|
// virtual nodes have stopped. (Dropping the guard runs `pactl unload`.)
|
|
drop(self.echo_cancel);
|
|
|
|
crate::log_msg("Leaving room...");
|
|
let _ = self.room_state.leave().await;
|
|
// Close peer links with the graceful goodbye code so remotes evict us
|
|
// promptly (not after the reconnect grace / slow gossip Leave), and stop
|
|
// our supervisors so none redial the about-to-close endpoint.
|
|
self.transport.leave().await;
|
|
crate::log_msg("Room left");
|
|
|
|
crate::log_msg("Shutting down router...");
|
|
let _ = tokio::time::timeout(std::time::Duration::from_secs(1), self.router.shutdown()).await;
|
|
crate::log_msg("Router shut down");
|
|
|
|
crate::log_msg("Joining capture thread...");
|
|
let _ = self.capture_thread.join();
|
|
crate::log_msg("ActiveSession::shutdown complete");
|
|
}
|
|
}
|
|
|
|
/// Finalize and clear the active recording, if any, emitting `RecordingStopped`.
|
|
/// No-op when not recording. Called on stop, room leave, and room switch so a
|
|
/// recording is always closed cleanly (its WAV size fields patched).
|
|
async fn stop_recording(
|
|
recorder: &Arc<std::sync::Mutex<Option<crate::audio::recorder::Recorder>>>,
|
|
is_recording: &Arc<AtomicBool>,
|
|
ui_tx: &mpsc::Sender<UiEvent>,
|
|
) {
|
|
is_recording.store(false, Ordering::Relaxed);
|
|
let rec = recorder.lock().unwrap().take();
|
|
if let Some(rec) = rec {
|
|
let path = rec.path().to_string_lossy().to_string();
|
|
if let Err(e) = rec.finalize() {
|
|
crate::log_msg(&format!("Failed to finalize recording: {e}"));
|
|
}
|
|
crate::log_msg(&format!("Recording saved: {path}"));
|
|
let _ = ui_tx.send(UiEvent::RecordingStopped { path }).await;
|
|
}
|
|
}
|
|
|
|
async fn run_core_loop(
|
|
mut cmd_rx: mpsc::Receiver<CoreCommand>,
|
|
ui_tx: mpsc::Sender<UiEvent>,
|
|
) -> Result<(), anyhow::Error> {
|
|
let memory_lookup = iroh::address_lookup::memory::MemoryLookup::new();
|
|
let secret_key = iroh::SecretKey::generate();
|
|
|
|
let audio_backend = Arc::new(PipeWireBackend::new());
|
|
|
|
let is_muted = Arc::new(AtomicBool::new(false));
|
|
let is_deafened = Arc::new(AtomicBool::new(false));
|
|
let ptt_mode = Arc::new(AtomicBool::new(false));
|
|
let ptt_active = Arc::new(AtomicBool::new(false));
|
|
let noise_gate_threshold = Arc::new(std::sync::atomic::AtomicU32::new(0.01f32.to_bits()));
|
|
// 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 output_gain = Arc::new(std::sync::atomic::AtomicU32::new(1.0f32.to_bits()));
|
|
// Call recording: an optional live recorder (mic FIFO + WAV writer), shared
|
|
// 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
|
|
// while a recording is actually running.
|
|
let recorder: Arc<std::sync::Mutex<Option<crate::audio::recorder::Recorder>>> =
|
|
Arc::new(std::sync::Mutex::new(None));
|
|
let is_recording = Arc::new(AtomicBool::new(false));
|
|
let peer_volumes = Arc::new(Mutex::new(HashMap::<EndpointId, f32>::new()));
|
|
// Peers locally muted by us: decoded for level metering but not mixed.
|
|
let locally_muted = Arc::new(Mutex::new(HashSet::<EndpointId>::new()));
|
|
let mut current_name = "Anonymous".to_string();
|
|
let mut network_mode = NetworkMode::default();
|
|
|
|
let mut active_session: Option<ActiveSession> = None;
|
|
// Standalone capture-only mic meter, live only when no session exists.
|
|
let mut mic_monitor: Option<MicMonitor> = None;
|
|
|
|
while let Some(cmd) = cmd_rx.recv().await {
|
|
match cmd {
|
|
CoreCommand::Join { name, ticket, input_device, output_device, echo_cancellation } => {
|
|
current_name = name.clone();
|
|
|
|
// Finalize any recording before tearing down the old session — its
|
|
// capture/mixer feeders are about to stop.
|
|
stop_recording(&recorder, &is_recording, &ui_tx).await;
|
|
|
|
// Clean up any existing session
|
|
if let Some(session) = active_session.take() {
|
|
crate::log_msg("Shutting down existing active session");
|
|
session.shutdown(audio_backend.clone()).await;
|
|
}
|
|
|
|
// Release a standalone mic monitor if running — it shares the
|
|
// backend's single capture stream, so it must stop before the
|
|
// call claims it. (Safe here: any session was just shut down.)
|
|
stop_mic_monitor(&audio_backend, mic_monitor.take());
|
|
|
|
// Build the endpoint per the configured relay/discovery posture.
|
|
// All postures keep the in-memory address lookup (fed by tickets
|
|
// and gossip); they differ in whether n0's relay and DNS presence
|
|
// beacon are used. `Minimal` sets only the mandatory crypto
|
|
// provider and deliberately omits the n0 DNS publish/resolve.
|
|
let bind_result = match network_mode {
|
|
NetworkMode::N0Full => {
|
|
Endpoint::builder(presets::N0)
|
|
.secret_key(secret_key.clone())
|
|
.address_lookup(memory_lookup.clone())
|
|
.bind()
|
|
.await
|
|
}
|
|
NetworkMode::RelayNoDiscovery => {
|
|
Endpoint::builder(presets::Minimal)
|
|
.secret_key(secret_key.clone())
|
|
.relay_mode(RelayMode::Default)
|
|
.address_lookup(memory_lookup.clone())
|
|
.bind()
|
|
.await
|
|
}
|
|
NetworkMode::DirectOnly => {
|
|
Endpoint::builder(presets::Minimal)
|
|
.secret_key(secret_key.clone())
|
|
.relay_mode(RelayMode::Disabled)
|
|
.address_lookup(memory_lookup.clone())
|
|
.bind()
|
|
.await
|
|
}
|
|
};
|
|
let endpoint = match bind_result {
|
|
Ok(ep) => ep,
|
|
Err(e) => {
|
|
let _ = ui_tx.send(UiEvent::Error(format!("Failed to bind endpoint: {}", e))).await;
|
|
continue;
|
|
}
|
|
};
|
|
endpoint.online().await;
|
|
|
|
// Determine target ticket
|
|
let ticket_str = if ticket.trim().is_empty() || ticket == "create" {
|
|
let topic_id: [u8; 32] = rand::random();
|
|
let host_addr = endpoint.addr();
|
|
crate::log_msg(&format!("Creating room. host_addr={:?}, topic_id={:?}", host_addr, topic_id));
|
|
let ticket = PeerSpeakTicket { host_addr, topic_id };
|
|
ticket.to_string()
|
|
} else {
|
|
let ticket_str = ticket.trim().to_string();
|
|
crate::log_msg(&format!("Joining room with existing ticket={}", ticket_str));
|
|
ticket_str
|
|
};
|
|
|
|
// Initialize Gossip and Transport
|
|
let gossip = Gossip::builder().spawn(endpoint.clone());
|
|
let (transport, audio_proto) = IrohTransport::new(endpoint.clone());
|
|
let transport = Arc::new(transport);
|
|
|
|
// Start Router
|
|
let router = iroh::protocol::Router::builder(endpoint.clone())
|
|
.accept(iroh_gossip::net::GOSSIP_ALPN, gossip.clone())
|
|
.accept(b"peerspeak-audio", audio_proto)
|
|
.spawn();
|
|
|
|
let room_state = Arc::new(IrohGossipState::new(
|
|
endpoint.clone(),
|
|
gossip.clone(),
|
|
memory_lookup.clone(),
|
|
));
|
|
|
|
let self_state = PeerState {
|
|
name: current_name.clone(),
|
|
is_muted: is_muted.load(Ordering::Relaxed),
|
|
addr: endpoint.addr(),
|
|
};
|
|
|
|
crate::log_msg(&format!("Attempting room_state.join with self_state={:?}", self_state));
|
|
if let Err(e) = room_state.join(&ticket_str, self_state.clone()).await {
|
|
crate::log_msg(&format!("Error room_state.join failed: {:?}", e));
|
|
let _ = ui_tx.send(UiEvent::Error(format!("Failed to join room: {}", e))).await;
|
|
let _ = router.shutdown().await;
|
|
continue;
|
|
}
|
|
crate::log_msg("Joined room successfully via room_state");
|
|
|
|
// Setup raw audio channels
|
|
let (capture_tx, capture_rx) = std::sync::mpsc::channel();
|
|
let (playback_tx, playback_rx) = std::sync::mpsc::channel();
|
|
|
|
// Echo cancellation: if enabled, load PipeWire's echo-cancel module
|
|
// bound to the chosen real devices and route capture/playback
|
|
// through its virtual nodes (the sink doubles as the AEC reference).
|
|
// The guard unloads the module on drop — including the early-return
|
|
// paths below, since it's a local until moved into the session. On
|
|
// any failure, warn and fall back to the direct devices.
|
|
let mut echo_cancel_guard = None;
|
|
let (capture_target, playback_target) = if echo_cancellation {
|
|
match crate::audio::echo_cancel::enable(
|
|
input_device.as_deref(),
|
|
output_device.as_deref(),
|
|
) {
|
|
Ok(guard) => {
|
|
echo_cancel_guard = Some(guard);
|
|
crate::log_msg("Echo cancellation enabled");
|
|
(
|
|
Some(crate::audio::echo_cancel::EC_SOURCE.to_string()),
|
|
Some(crate::audio::echo_cancel::EC_SINK.to_string()),
|
|
)
|
|
}
|
|
Err(e) => {
|
|
crate::log_msg(&format!(
|
|
"Echo cancellation unavailable, using direct devices: {e}"
|
|
));
|
|
let _ = ui_tx
|
|
.send(UiEvent::Error(format!("Echo cancellation unavailable: {e}")))
|
|
.await;
|
|
(input_device.clone(), output_device.clone())
|
|
}
|
|
}
|
|
} else {
|
|
(input_device.clone(), output_device.clone())
|
|
};
|
|
|
|
if let Err(e) = audio_backend.start_capture(capture_tx, capture_target) {
|
|
let _ = ui_tx.send(UiEvent::Error(format!("Failed to start capture: {}", e))).await;
|
|
let _ = room_state.leave().await;
|
|
let _ = router.shutdown().await;
|
|
continue;
|
|
}
|
|
|
|
// Shared gauge: PipeWire publishes the playback ring's live depth
|
|
// here (drain side + fill side); the mixer reads it to pace
|
|
// production to the hardware clock instead of a fixed timer.
|
|
let ring_fill = Arc::new(AtomicUsize::new(0));
|
|
if let Err(e) = audio_backend.start_playback(playback_rx, playback_target, ring_fill.clone()) {
|
|
let _ = ui_tx.send(UiEvent::Error(format!("Failed to start playback: {}", e))).await;
|
|
let _ = audio_backend.stop();
|
|
let _ = room_state.leave().await;
|
|
let _ = router.shutdown().await;
|
|
continue;
|
|
}
|
|
|
|
let jitter: Arc<Mutex<HashMap<EndpointId, JitterBuffer>>> = Arc::new(Mutex::new(HashMap::new()));
|
|
|
|
// 1. Capture & encoding thread
|
|
let is_muted_clone = is_muted.clone();
|
|
let ptt_mode_clone = ptt_mode.clone();
|
|
let ptt_active_clone = ptt_active.clone();
|
|
let noise_gate_threshold_clone = noise_gate_threshold.clone();
|
|
let input_gain_clone = input_gain.clone();
|
|
let transport_clone = transport.clone();
|
|
let ui_tx_capture = ui_tx.clone();
|
|
let recorder_capture = recorder.clone();
|
|
let is_recording_capture = is_recording.clone();
|
|
|
|
let capture_thread = std::thread::spawn(move || {
|
|
use opus::{Channels, Application};
|
|
let mut encoder = match OpusEncoder::new(48000, Channels::Mono, Application::Voip) {
|
|
Ok(enc) => enc,
|
|
Err(e) => {
|
|
crate::log_msg(&format!("Capture thread error: {:?}", e));
|
|
return;
|
|
}
|
|
};
|
|
// Per-sender packet sequence number, prepended to every frame so
|
|
// receivers can reorder and conceal loss. Wraps after ~years.
|
|
let mut seq: u32 = 0;
|
|
// Smooth noise gate (hysteresis + attack/release + hangover),
|
|
// carrying envelope state across frames. The live slider value
|
|
// is read per frame so changes apply immediately.
|
|
let mut gate = crate::audio::gate::NoiseGate::new(48000);
|
|
// Peak-held raw mic level for the settings meter, reported
|
|
// pre-gate/pre-mute so calibration reflects the true input.
|
|
let mut mic_meter = MicLevelMeter::new();
|
|
|
|
while let Ok(mut pcm) = capture_rx.recv() {
|
|
// Apply the input gain first so the meter, gate, and what we
|
|
// transmit all reflect the same (gained) signal.
|
|
apply_volume(&mut pcm, f32::from_bits(input_gain_clone.load(Ordering::Relaxed)));
|
|
|
|
if let Some(peak) = mic_meter.push(&pcm) {
|
|
let _ = ui_tx_capture.try_send(UiEvent::MicLevel(peak));
|
|
}
|
|
|
|
if is_muted_clone.load(Ordering::Relaxed) {
|
|
continue;
|
|
}
|
|
if ptt_mode_clone.load(Ordering::Relaxed) && !ptt_active_clone.load(Ordering::Relaxed) {
|
|
continue;
|
|
}
|
|
|
|
let ng_bits = noise_gate_threshold_clone.load(Ordering::Relaxed);
|
|
let ng_thresh = f32::from_bits(ng_bits);
|
|
// Apply the gate in place; skip transmitting a fully-closed
|
|
// frame so we don't send pure silence (the receiver's jitter
|
|
// buffer conceals the gap).
|
|
if !gate.process(&mut pcm, ng_thresh) {
|
|
continue;
|
|
}
|
|
|
|
// Record what we transmit (post-gain, post-gate, post-mute):
|
|
// this is exactly the mic audio peers receive from us. The
|
|
// mixer task pairs it with the incoming mix.
|
|
if is_recording_capture.load(Ordering::Relaxed)
|
|
&& let Some(rec) = recorder_capture.lock().unwrap().as_mut()
|
|
{
|
|
rec.push_mic(&pcm);
|
|
}
|
|
|
|
if let Ok(encoded) = encoder.encode(&pcm) {
|
|
// Frame on the wire: [seq: u32 LE][opus payload].
|
|
let mut packet = Vec::with_capacity(4 + encoded.len());
|
|
packet.extend_from_slice(&seq.to_le_bytes());
|
|
packet.extend_from_slice(&encoded);
|
|
seq = seq.wrapping_add(1);
|
|
transport_clone.broadcast(bytes::Bytes::from(packet));
|
|
}
|
|
}
|
|
});
|
|
|
|
// 2. Receiver task: parse the sequence header and hand each packet
|
|
// to that peer's jitter buffer. Decoding happens later, on the
|
|
// playout side, so loss can be concealed at the right moment.
|
|
let transport_recv = transport.clone();
|
|
let jitter_recv = jitter.clone();
|
|
let datagram_task = tokio::spawn(async move {
|
|
let mut datagram_rx = match transport_recv.receive_datagrams().await {
|
|
Ok(rx) => rx,
|
|
Err(e) => {
|
|
crate::log_msg(&format!("Receiver task error: {:?}", e));
|
|
return;
|
|
}
|
|
};
|
|
|
|
while let Some((from_peer, bytes)) = datagram_rx.recv().await {
|
|
if bytes.len() < 4 {
|
|
continue; // malformed: missing sequence header
|
|
}
|
|
let seq = u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
|
|
let payload = bytes[4..].to_vec();
|
|
|
|
let mut guard = jitter_recv.lock().await;
|
|
let buffer = match guard.entry(from_peer) {
|
|
std::collections::hash_map::Entry::Occupied(entry) => entry.into_mut(),
|
|
std::collections::hash_map::Entry::Vacant(entry) => {
|
|
match JitterBuffer::new() {
|
|
Ok(jb) => entry.insert(jb),
|
|
Err(e) => {
|
|
crate::log_msg(&format!("Failed to init jitter buffer for {:?}: {:?}", from_peer, e));
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
};
|
|
buffer.insert(seq, payload);
|
|
}
|
|
});
|
|
|
|
// 3. Mixing & level extraction loop task. Production is paced by
|
|
// the playback ring's fill level (the PipeWire hardware clock),
|
|
// NOT a fixed software timer: we produce a 20ms frame only when
|
|
// the ring is below its target depth, so the long-run mix rate
|
|
// auto-matches the device drain rate and the producer/consumer
|
|
// beat (which otherwise churns ~20% of audio) disappears. Each
|
|
// produced frame pulls one concealed frame per peer from its
|
|
// jitter buffer, applies per-peer volume, and sums.
|
|
let jitter_mixer = jitter.clone();
|
|
let is_deafened_clone = is_deafened.clone();
|
|
let peer_volumes_mixer = peer_volumes.clone();
|
|
let locally_muted_mixer = locally_muted.clone();
|
|
let output_gain_mixer = output_gain.clone();
|
|
let ui_tx_mixer = ui_tx.clone();
|
|
let ring_fill_mixer = ring_fill.clone();
|
|
let recorder_mixer = recorder.clone();
|
|
let is_recording_mixer = is_recording.clone();
|
|
let mixer_task = tokio::spawn(async move {
|
|
// Mix-bus soft limiter: rides loud multi-peer moments down to
|
|
// the ceiling instead of hard-clipping. State carries across
|
|
// frames (see audio::limiter).
|
|
let mut limiter = crate::audio::limiter::SoftLimiter::new(48_000);
|
|
// When the ring is at/above target we have nothing to do; nap
|
|
// briefly and re-check. Short enough (relative to the ~60ms
|
|
// target and ~21ms device quantum) that we always refill well
|
|
// before the ring can run dry.
|
|
const IDLE_NAP: Duration = Duration::from_millis(2);
|
|
|
|
// Pushing a level event per frame floods the UI runtime at
|
|
// ~50/sec. We peak-hold per-peer levels across this many
|
|
// produced frames and emit once per window (~10/sec) —
|
|
// peak-hold so a brief transient still lights the indicator.
|
|
const LEVEL_EMIT_FRAMES: u32 = 5;
|
|
let mut level_peaks: HashMap<EndpointId, f32> = HashMap::new();
|
|
let mut frames_since_emit: u32 = 0;
|
|
|
|
loop {
|
|
// Pace to the hardware clock: only produce while the ring
|
|
// is draining below target. Otherwise yield and re-check.
|
|
if ring_fill_mixer.load(Ordering::Relaxed) >= crate::audio::PLAYBACK_TARGET_SAMPLES {
|
|
tokio::time::sleep(IDLE_NAP).await;
|
|
continue;
|
|
}
|
|
|
|
let current_volumes = peer_volumes_mixer.lock().await.clone();
|
|
let muted_peers = locally_muted_mixer.lock().await.clone();
|
|
let mut peer_frames = Vec::new();
|
|
|
|
{
|
|
let mut guard = jitter_mixer.lock().await;
|
|
for (&peer_id, buffer) in guard.iter_mut() {
|
|
// `None` means idle/buffering: contribute nothing,
|
|
// but keep a (zero) entry so the UI sees it idle.
|
|
let Some(mut frame) = buffer.pop_frame() else {
|
|
level_peaks.entry(peer_id).or_insert(0.0);
|
|
continue;
|
|
};
|
|
|
|
let vol = current_volumes.get(&peer_id).copied().unwrap_or(1.0);
|
|
apply_volume(&mut frame, vol);
|
|
|
|
// Level is recorded even for locally-muted peers so
|
|
// the UI still shows that they're speaking.
|
|
let peak = level_peaks.entry(peer_id).or_insert(0.0);
|
|
*peak = peak.max(frame_level(&frame));
|
|
|
|
// Locally muted: decoded above (jitter buffer advances,
|
|
// level shown) but not mixed into our output.
|
|
if muted_peers.contains(&peer_id) {
|
|
continue;
|
|
}
|
|
|
|
peer_frames.push(frame);
|
|
}
|
|
}
|
|
|
|
// Lossless i32 sum, then the limiter applies the master
|
|
// output gain (in f32, so a boost past the ceiling is
|
|
// limited too) and rides peaks down to the ceiling.
|
|
let mixed_sum = mix_frames(&peer_frames, FRAME_SAMPLES);
|
|
let out_gain = f32::from_bits(output_gain_mixer.load(Ordering::Relaxed));
|
|
let mixed = limiter.process(&mixed_sum, out_gain);
|
|
|
|
// Record the true call audio (incoming mix + our mic),
|
|
// independent of local deafen — deafen only silences our
|
|
// own monitor, not what the call actually carried.
|
|
if is_recording_mixer.load(Ordering::Relaxed)
|
|
&& let Some(rec) = recorder_mixer.lock().unwrap().as_mut()
|
|
&& let Err(e) = rec.write_frame(&mixed)
|
|
{
|
|
crate::log_msg(&format!("Recording write failed: {e}"));
|
|
}
|
|
|
|
let frame_to_send = if is_deafened_clone.load(Ordering::Relaxed) {
|
|
vec![0i16; FRAME_SAMPLES]
|
|
} else {
|
|
mixed
|
|
};
|
|
|
|
if playback_tx.send(frame_to_send).is_err() {
|
|
break;
|
|
}
|
|
|
|
// Emit coalesced peaks once per window, then reset.
|
|
frames_since_emit += 1;
|
|
if frames_since_emit >= LEVEL_EMIT_FRAMES {
|
|
let levels: Vec<(EndpointId, f32)> = level_peaks.drain().collect();
|
|
let _ = ui_tx_mixer.send(UiEvent::AudioLevels(levels)).await;
|
|
frames_since_emit = 0;
|
|
}
|
|
}
|
|
});
|
|
|
|
// 4. Room event subscriber task
|
|
let mut room_events = match room_state.subscribe_events().await {
|
|
Ok(rx) => rx,
|
|
Err(e) => {
|
|
let _ = ui_tx.send(UiEvent::Error(format!("Failed to subscribe events: {}", e))).await;
|
|
continue;
|
|
}
|
|
};
|
|
let ui_tx_events = ui_tx.clone();
|
|
let jitter_events = jitter.clone();
|
|
let transport_events = transport.clone();
|
|
let grace_timers: GraceTimers = Arc::new(std::sync::Mutex::new(HashMap::new()));
|
|
let grace_timers_events = grace_timers.clone();
|
|
let seen_connected: SeenConnected = Arc::new(std::sync::Mutex::new(HashSet::new()));
|
|
let seen_connected_events = seen_connected.clone();
|
|
let event_task = tokio::spawn(async move {
|
|
while let Some(event) = room_events.recv().await {
|
|
match event {
|
|
RoomEvent::PeerJoined(peer_id, state) => {
|
|
// A (re)join means the peer is back — cancel any
|
|
// pending reconnect grace timer before re-adding it.
|
|
cancel_grace_timer(&grace_timers_events, &peer_id);
|
|
// Establish the audio connection as soon as the peer
|
|
// is known (the transport dedupes the full-mesh race).
|
|
// Hand over the full address so reconnects can dial
|
|
// it directly rather than via the gossip lookup.
|
|
transport_events.connect_peer(state.addr.clone()).await;
|
|
let _ = ui_tx_events.send(UiEvent::PeerJoined { id: peer_id, state }).await;
|
|
}
|
|
RoomEvent::PeerLeft(peer_id) => {
|
|
// Graceful leave — evict immediately.
|
|
cancel_grace_timer(&grace_timers_events, &peer_id);
|
|
seen_connected_events.lock().unwrap().remove(&peer_id);
|
|
transport_events.disconnect_peer(peer_id).await;
|
|
jitter_events.lock().await.remove(&peer_id);
|
|
let _ = ui_tx_events.send(UiEvent::PeerLeft { id: peer_id }).await;
|
|
}
|
|
RoomEvent::PeerUpdated(peer_id, state) => {
|
|
// A re-announce means the peer is alive — cancel any
|
|
// pending grace timer. It may also carry a fresh
|
|
// address (peer back on a new network); refresh the
|
|
// retained dial target so a later reconnect re-reaches
|
|
// it. Idempotent: an ordinary mute/unmute update just
|
|
// re-records the same address.
|
|
cancel_grace_timer(&grace_timers_events, &peer_id);
|
|
transport_events.connect_peer(state.addr.clone()).await;
|
|
let _ = ui_tx_events.send(UiEvent::PeerUpdated { id: peer_id, state }).await;
|
|
}
|
|
RoomEvent::ChatMessage { name, text, .. } => {
|
|
let _ = ui_tx_events.send(UiEvent::ChatMessage { name, text }).await;
|
|
}
|
|
RoomEvent::PeerConnectionLost(peer_id) => {
|
|
// Transient drop: do NOT tear down the peer. Its audio
|
|
// supervisor stays alive and keeps redialing the
|
|
// retained address, so show "reconnecting" and arm a
|
|
// grace timer that evicts the peer only if the link
|
|
// hasn't recovered within RECONNECT_GRACE. A gossip
|
|
// rejoin (PeerJoined/PeerUpdated) or a transport
|
|
// reconnect (ConnEvent::Connected) cancels it first.
|
|
let _ = ui_tx_events.send(UiEvent::PeerConnecting { id: peer_id }).await;
|
|
arm_grace_timer(
|
|
&grace_timers_events,
|
|
&seen_connected_events,
|
|
&transport_events,
|
|
&jitter_events,
|
|
&ui_tx_events,
|
|
RECONNECT_GRACE,
|
|
peer_id,
|
|
);
|
|
}
|
|
}
|
|
}
|
|
});
|
|
|
|
// 5. Connection-state forwarder: turns transport link state into
|
|
// per-peer UI indicators (connecting / reconnecting vs. live).
|
|
let mut conn_events = match transport.subscribe_conn_events().await {
|
|
Ok(rx) => rx,
|
|
Err(e) => {
|
|
let _ = ui_tx.send(UiEvent::Error(format!("Failed to subscribe conn events: {}", e))).await;
|
|
continue;
|
|
}
|
|
};
|
|
let conn_handler = ConnEventHandler::new(
|
|
ui_tx.clone(),
|
|
grace_timers.clone(),
|
|
seen_connected.clone(),
|
|
transport.clone(),
|
|
jitter.clone(),
|
|
);
|
|
let conn_event_task = tokio::spawn(async move {
|
|
while let Some(event) = conn_events.recv().await {
|
|
conn_handler.handle(event).await;
|
|
}
|
|
});
|
|
|
|
let session = ActiveSession {
|
|
endpoint: endpoint.clone(),
|
|
router,
|
|
room_state: room_state.clone(),
|
|
capture_thread,
|
|
datagram_task,
|
|
mixer_task,
|
|
event_task,
|
|
conn_event_task,
|
|
grace_timers,
|
|
transport: transport.clone(),
|
|
echo_cancel: echo_cancel_guard,
|
|
};
|
|
|
|
let self_id = endpoint.id().to_string();
|
|
let _ = ui_tx.send(UiEvent::RoomJoined { ticket: ticket_str, self_id }).await;
|
|
active_session = Some(session);
|
|
}
|
|
|
|
CoreCommand::Leave => {
|
|
// Finalize any recording first, while the audio feeders are alive.
|
|
stop_recording(&recorder, &is_recording, &ui_tx).await;
|
|
if let Some(session) = active_session.take() {
|
|
session.shutdown(audio_backend.clone()).await;
|
|
let _ = ui_tx.send(UiEvent::RoomLeft).await;
|
|
}
|
|
}
|
|
|
|
CoreCommand::ToggleMute => {
|
|
let current = is_muted.load(Ordering::Relaxed);
|
|
let new_state = !current;
|
|
is_muted.store(new_state, Ordering::Relaxed);
|
|
|
|
if let Some(session) = &active_session {
|
|
let self_state = PeerState {
|
|
name: current_name.clone(),
|
|
is_muted: new_state,
|
|
addr: session.endpoint.addr(),
|
|
};
|
|
let _ = session.room_state.update_self_state(self_state).await;
|
|
}
|
|
}
|
|
|
|
CoreCommand::ToggleDeafen => {
|
|
let current = is_deafened.load(Ordering::Relaxed);
|
|
is_deafened.store(!current, Ordering::Relaxed);
|
|
}
|
|
|
|
CoreCommand::SetPttMode(enabled) => {
|
|
ptt_mode.store(enabled, Ordering::Relaxed);
|
|
}
|
|
|
|
CoreCommand::SetPttActive(active) => {
|
|
ptt_active.store(active, Ordering::Relaxed);
|
|
}
|
|
|
|
CoreCommand::SetPeerVolume(peer_id, vol) => {
|
|
let mut guard = peer_volumes.lock().await;
|
|
guard.insert(peer_id, vol);
|
|
}
|
|
|
|
CoreCommand::SetPeerMuted(peer_id, muted) => {
|
|
let mut guard = locally_muted.lock().await;
|
|
if muted {
|
|
guard.insert(peer_id);
|
|
} else {
|
|
guard.remove(&peer_id);
|
|
}
|
|
}
|
|
|
|
CoreCommand::SetNoiseGateThreshold(threshold) => {
|
|
noise_gate_threshold.store(threshold.to_bits(), Ordering::Relaxed);
|
|
}
|
|
|
|
CoreCommand::SetInputVolume(vol) => {
|
|
input_gain.store(vol.to_bits(), Ordering::Relaxed);
|
|
}
|
|
|
|
CoreCommand::SetOutputVolume(vol) => {
|
|
output_gain.store(vol.to_bits(), Ordering::Relaxed);
|
|
}
|
|
|
|
CoreCommand::SetMicMonitor { enabled, input_device } => {
|
|
// During a call the in-call capture thread already reports the
|
|
// mic level, and it owns the backend's capture stream — leave it be.
|
|
if active_session.is_some() {
|
|
continue;
|
|
}
|
|
if enabled {
|
|
if mic_monitor.is_none() {
|
|
let (tx, rx) = std::sync::mpsc::channel();
|
|
match audio_backend.start_capture(tx, input_device) {
|
|
Ok(()) => {
|
|
let ui = ui_tx.clone();
|
|
let gain = input_gain.clone();
|
|
let thread = std::thread::spawn(move || run_mic_monitor(rx, ui, gain));
|
|
mic_monitor = Some(MicMonitor { thread });
|
|
}
|
|
Err(e) => {
|
|
let _ = ui_tx
|
|
.send(UiEvent::Error(format!("Mic test unavailable: {e}")))
|
|
.await;
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
stop_mic_monitor(&audio_backend, mic_monitor.take());
|
|
}
|
|
}
|
|
|
|
CoreCommand::SetNetworkMode(mode) => {
|
|
network_mode = mode;
|
|
}
|
|
|
|
CoreCommand::SetRecording(enabled) => {
|
|
if enabled {
|
|
// Only record while in a call, and not already recording.
|
|
if active_session.is_none() {
|
|
let _ = ui_tx
|
|
.send(UiEvent::Error("Join a call before recording".into()))
|
|
.await;
|
|
} else if !is_recording.load(Ordering::Relaxed) {
|
|
match dirs::home_dir() {
|
|
Some(home) => {
|
|
let dir = home.join("peerspeak-recordings");
|
|
let now = std::time::SystemTime::now()
|
|
.duration_since(std::time::UNIX_EPOCH)
|
|
.map(|d| d.as_secs())
|
|
.unwrap_or(0);
|
|
let started = std::fs::create_dir_all(&dir)
|
|
.map_err(|e| e.to_string())
|
|
.and_then(|_| {
|
|
crate::audio::recorder::Recorder::create(&dir, now)
|
|
.map_err(|e| e.to_string())
|
|
});
|
|
match started {
|
|
Ok(rec) => {
|
|
let path = rec.path().to_string_lossy().to_string();
|
|
*recorder.lock().unwrap() = Some(rec);
|
|
is_recording.store(true, Ordering::Relaxed);
|
|
crate::log_msg(&format!("Recording started: {path}"));
|
|
let _ = ui_tx
|
|
.send(UiEvent::RecordingStarted { path })
|
|
.await;
|
|
}
|
|
Err(e) => {
|
|
let _ = ui_tx
|
|
.send(UiEvent::Error(format!("Recording failed: {e}")))
|
|
.await;
|
|
}
|
|
}
|
|
}
|
|
None => {
|
|
let _ = ui_tx
|
|
.send(UiEvent::Error("No home directory for recordings".into()))
|
|
.await;
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
stop_recording(&recorder, &is_recording, &ui_tx).await;
|
|
}
|
|
}
|
|
|
|
CoreCommand::SendChat(text) => {
|
|
if let Some(session) = &active_session
|
|
&& let Err(e) = session.room_state.send_chat(text).await
|
|
{
|
|
crate::log_msg(&format!("Failed to send chat: {e}"));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::{apply_volume, frame_level, mix_frames, MicLevelMeter, MIC_LEVEL_REPORT_SAMPLES};
|
|
|
|
/// A frame of constant amplitude with the given sample count.
|
|
fn frame(amp: i16, len: usize) -> Vec<i16> {
|
|
vec![amp; len]
|
|
}
|
|
|
|
#[test]
|
|
fn mic_meter_reports_only_after_enough_samples() {
|
|
let mut m = MicLevelMeter::new();
|
|
// One short frame well under the report window yields nothing yet.
|
|
assert_eq!(m.push(&frame(1000, 480)), None);
|
|
// A frame that crosses the window boundary triggers a report.
|
|
assert!(m.push(&frame(1000, MIC_LEVEL_REPORT_SAMPLES)).is_some());
|
|
}
|
|
|
|
#[test]
|
|
fn mic_meter_holds_the_peak_across_the_window() {
|
|
let mut m = MicLevelMeter::new();
|
|
let chunk = MIC_LEVEL_REPORT_SAMPLES / 4;
|
|
// Loud frame first, then quiet ones — the reported value is the loud peak.
|
|
assert_eq!(m.push(&frame(8000, chunk)), None);
|
|
assert_eq!(m.push(&frame(100, chunk)), None);
|
|
assert_eq!(m.push(&frame(100, chunk)), None);
|
|
let reported = m.push(&frame(100, chunk)).expect("window complete");
|
|
let loud = frame_level(&frame(8000, chunk));
|
|
assert!((reported - loud).abs() < 1e-6, "peak should hold the loud frame");
|
|
}
|
|
|
|
#[test]
|
|
fn mic_meter_resets_after_reporting() {
|
|
let mut m = MicLevelMeter::new();
|
|
// Fill and report a loud window.
|
|
assert!(m.push(&frame(8000, MIC_LEVEL_REPORT_SAMPLES)).is_some());
|
|
// The next window of silence must report ~zero, not the stale loud peak.
|
|
let reported = m.push(&frame(0, MIC_LEVEL_REPORT_SAMPLES)).expect("second window");
|
|
assert_eq!(reported, 0.0, "peak and accumulator reset between windows");
|
|
}
|
|
|
|
#[test]
|
|
fn mic_meter_silence_reports_zero() {
|
|
let mut m = MicLevelMeter::new();
|
|
let reported = m.push(&frame(0, MIC_LEVEL_REPORT_SAMPLES)).expect("window complete");
|
|
assert_eq!(reported, 0.0);
|
|
}
|
|
|
|
#[test]
|
|
fn mix_of_no_peers_is_silence() {
|
|
let mixed = mix_frames(&[], 4);
|
|
assert_eq!(mixed, vec![0i32; 4]);
|
|
}
|
|
|
|
#[test]
|
|
fn single_peer_passes_through_unchanged() {
|
|
let frame = vec![100, -200, 300, -400];
|
|
let mixed = mix_frames(std::slice::from_ref(&frame), 4);
|
|
assert_eq!(mixed, vec![100i32, -200, 300, -400]);
|
|
}
|
|
|
|
#[test]
|
|
fn two_peers_sum_sample_by_sample() {
|
|
let a = vec![100, -200, 300, -400];
|
|
let b = vec![50, 200, -100, 400];
|
|
let mixed = mix_frames(&[a, b], 4);
|
|
assert_eq!(mixed, vec![150i32, 0, 200, 0]);
|
|
}
|
|
|
|
#[test]
|
|
fn loud_positive_mix_is_lossless_not_clamped() {
|
|
// The bus is a lossless i32 sum now — the true peak (~2x i16::MAX) is
|
|
// preserved so the limiter can ride it down. (The old mixer clamped here.)
|
|
let a = vec![30_000; 4];
|
|
let b = vec![30_000; 4];
|
|
let mixed = mix_frames(&[a, b], 4);
|
|
assert_eq!(mixed, vec![60_000i32; 4]);
|
|
}
|
|
|
|
#[test]
|
|
fn loud_negative_mix_is_lossless_not_clamped() {
|
|
let a = vec![i16::MIN; 4];
|
|
let b = vec![i16::MIN; 4];
|
|
let mixed = mix_frames(&[a, b], 4);
|
|
assert_eq!(mixed, vec![2 * i16::MIN as i32; 4]);
|
|
}
|
|
|
|
#[test]
|
|
fn shorter_peer_frame_contributes_zero_past_its_end() {
|
|
let full = vec![100, 100, 100, 100];
|
|
let short = vec![10, 20]; // only first two samples
|
|
let mixed = mix_frames(&[full, short], 4);
|
|
assert_eq!(mixed, vec![110i32, 120, 100, 100]);
|
|
}
|
|
|
|
#[test]
|
|
fn volume_unity_is_a_noop() {
|
|
let mut frame = vec![100, -200, 300, -400];
|
|
apply_volume(&mut frame, 1.0);
|
|
assert_eq!(frame, vec![100, -200, 300, -400]);
|
|
}
|
|
|
|
#[test]
|
|
fn volume_zero_mutes() {
|
|
let mut frame = vec![100, -200, 300, -400];
|
|
apply_volume(&mut frame, 0.0);
|
|
assert_eq!(frame, vec![0, 0, 0, 0]);
|
|
}
|
|
|
|
#[test]
|
|
fn volume_half_scales_samples() {
|
|
let mut frame = vec![100, -200, 300, -400];
|
|
apply_volume(&mut frame, 0.5);
|
|
// 100*0.5=50, -200*0.5=-100, 300*0.5=150, -400*0.5=-200 (exact in f32 here)
|
|
assert_eq!(frame, vec![50, -100, 150, -200]);
|
|
}
|
|
|
|
#[test]
|
|
fn volume_boost_saturates_not_wraps() {
|
|
// 20000 * 4.0 = 80000, well past i16::MAX — must clamp, not wrap.
|
|
let mut frame = vec![20_000, -20_000, 20_000, -20_000];
|
|
apply_volume(&mut frame, 4.0);
|
|
assert_eq!(frame, vec![i16::MAX, i16::MIN, i16::MAX, i16::MIN]);
|
|
}
|
|
|
|
#[test]
|
|
fn frame_level_of_silence_is_zero() {
|
|
assert_eq!(frame_level(&[0, 0, 0, 0]), 0.0);
|
|
assert_eq!(frame_level(&[]), 0.0);
|
|
}
|
|
|
|
#[test]
|
|
fn frame_level_of_full_scale_is_about_one() {
|
|
let full = vec![i16::MAX; 64];
|
|
let level = frame_level(&full);
|
|
assert!(level > 0.99 && level <= 1.0, "full-scale level was {level}");
|
|
}
|
|
|
|
#[test]
|
|
fn volume_truncates_toward_zero() {
|
|
let mut frame = vec![3, -3, 5, -5];
|
|
apply_volume(&mut frame, 0.5);
|
|
assert_eq!(frame, vec![1, -1, 2, -2]);
|
|
}
|
|
|
|
#[test]
|
|
fn volume_amplifies_without_saturating() {
|
|
let mut frame = vec![1000, -1000];
|
|
apply_volume(&mut frame, 2.0);
|
|
assert_eq!(frame, vec![2000, -2000]);
|
|
}
|
|
|
|
#[test]
|
|
fn three_peers_sum_without_saturation() {
|
|
let a = vec![10, 20];
|
|
let b = vec![3, 4];
|
|
let c = vec![100, -50];
|
|
let mixed = mix_frames(&[a, b, c], 2);
|
|
assert_eq!(mixed, vec![113i32, -26]);
|
|
}
|
|
|
|
#[test]
|
|
fn mix_zero_pads_output_longer_than_peer_frames() {
|
|
let a = vec![100, 200];
|
|
let mixed = mix_frames(&[a], 4);
|
|
assert_eq!(mixed, vec![100i32, 200, 0, 0]);
|
|
}
|
|
|
|
#[test]
|
|
fn frame_level_mid_range() {
|
|
let frame = vec![16384; 64];
|
|
let level = frame_level(&frame);
|
|
assert!((level - 0.5).abs() < 1e-3, "mid-range level was {level}");
|
|
}
|
|
}
|
|
|