flexaudio 0.5.0

Flexible cross-platform audio capture (microphone, system loopback, per-process) for Linux, Windows, and macOS with a unified API.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

//! Per-tap producer clock, delivery gate, flags, and ring publication.
use super::*;

#[derive(Default)]
pub(super) struct TapState {
    pub seq: u64,
    pub recovered: bool,
    pub discontinuity: bool,
    pub resume_generation: u64,
    pub last_pts: i64,
}

pub(super) enum Ring<'a> {
    Capture(&'a mut ChunkProducer),
    Primary(&'a mut ChunkProducer),
    Secondary(&'a mut SecondaryChunkProducer),
}

pub(super) fn drain(
    shared: &SharedState,
    mut pop: impl FnMut() -> Option<NormalizedChunk>,
    format: OutputFormat,
    gain: f32,
    counter: &AtomicU64,
    state: &mut TapState,
    mut ring: Ring<'_>,
) -> Result<bool> {
    let channels = usize::from(format.channels);
    let capture = matches!(ring, Ring::Capture(_));
    let primary = matches!(ring, Ring::Primary(_));
    let mut emitted = false;
    while let Some(chunk) = pop() {
        let mut data = chunk.samples;
        let raw_pts = chunk.pts_ns;
        let mut integrity_flags = chunk.flags;
        let frames = data.len() / channels;
        let frame_index = advance_frame_index(counter, frames as u64, format.sample_rate)?;
        // Popped frames advance even while discarded, retaining flags until delivery.
        if shared.paused.load(Ordering::SeqCst) {
            continue;
        }
        // Preserve the legacy output PTS/metrics preparation before the delivery lock.
        let prepared = if capture {
            None
        } else {
            let pts = apply_epoch(shared, raw_pts).max(state.last_pts);
            state.last_pts = pts;
            if apply_gain(&mut data, gain) {
                integrity_flags |= ChunkFlags::CLIPPED;
            }
            Some((pts, peak_rms(&data)))
        };
        let _delivery = shared.delivery.lock().unwrap_or_else(|e| e.into_inner());
        if shared.paused.load(Ordering::SeqCst) || shared.terminal.is_failed() {
            continue;
        }
        let (pts_ns, (peak, rms)) = prepared.unwrap_or_else(|| {
            let pts = apply_epoch(shared, raw_pts);
            if apply_gain(&mut data, gain) {
                integrity_flags |= ChunkFlags::CLIPPED;
            }
            (pts, peak_rms(&data))
        });
        let resumed = shared.resume_generation.load(Ordering::SeqCst);
        let mut flags = integrity_flags;
        if state.recovered && !capture && !shared.stopping.load(Ordering::SeqCst) {
            flags |= ChunkFlags::RECOVERED | ChunkFlags::DISCONTINUITY;
        } else if state.recovered && !capture {
            flags |= ChunkFlags::DISCONTINUITY;
        }
        if state.discontinuity || resumed != state.resume_generation {
            flags |= ChunkFlags::DISCONTINUITY;
        }
        let chunk = AudioChunk {
            data,
            frames,
            frame_index,
            pts_ns,
            seq: state.seq,
            flags,
            dropped_before: 0,
            peak,
            rms,
        };
        match &mut ring {
            Ring::Capture(producer) => {
                if let Some((_, evicted)) = producer.push_with_evicted(chunk) {
                    // This independent canonical transport is capture-side loss, not a
                    // primary/secondary output drop. Count the evicted PCM, including tails.
                    let samples = u64::try_from(evicted.data.len())
                        .ok()
                        .and_then(NonZeroU64::new);
                    let loss = AudioLoss::raw_overflow(
                        None,
                        samples,
                        format.sample_rate,
                        format.channels,
                    )?;
                    shared.push_event(Event::AudioLoss { loss });
                }
            }
            Ring::Primary(producer) => {
                if let Some(total) = producer.push(chunk) {
                    shared.push_event(Event::ChunkDropped { count: total });
                }
            }
            Ring::Secondary(producer) => {
                let dropped_before = producer.dropped_count();
                let dropped = producer.push(SecondaryChunk {
                    samples: chunk.data,
                    frames: chunk.frames,
                    frame_index: chunk.frame_index,
                    pts_ns: chunk.pts_ns,
                    seq: chunk.seq,
                    flags: chunk.flags,
                    dropped_before: 0,
                    peak: chunk.peak,
                    rms: chunk.rms,
                });
                if let Some(total) = dropped {
                    let samples = (total - dropped_before)
                        .checked_mul(frames as u64)
                        .and_then(|frames| frames.checked_mul(u64::from(format.channels)))
                        .and_then(NonZeroU64::new);
                    let loss = AudioLoss::output_overflow(
                        OutputTap::Secondary,
                        samples,
                        format.sample_rate,
                        format.channels,
                    )?;
                    shared.push_event(Event::AudioLoss { loss });
                }
            }
        }
        if primary && flags.contains(ChunkFlags::RECOVERED) {
            shared.push_event(Event::StreamRecovered);
        }
        state.recovered = false;
        state.discontinuity = false;
        state.resume_generation = resumed;
        state.seq += 1;
        emitted = true;
    }
    Ok(emitted)
}