ez_ffmpeg/core/frame_export/
audio_iter.rs

1//! [`SampleIter`]: a fused iterator over exported audio chunks with the same
2//! load-bearing teardown ordering as [`FrameIter`](super::FrameIter) (S6).
3
4use super::chunk::AudioChunk;
5use crate::core::scheduler::ffmpeg_scheduler::{FfmpegScheduler, Running};
6use crate::error::Error;
7use crossbeam_channel::Receiver;
8
9/// An iterator over exported [`AudioChunk`]s.
10///
11/// Yields `Ok(chunk)` per exported chunk. When the pipeline finishes (or dies),
12/// the iterator joins the scheduler once: a worker error surfaces as a single
13/// terminal `Err`, after which the iterator is fused and yields `None` forever.
14///
15/// Dropping the iterator early tears the run down cleanly. Teardown drops the
16/// receiver *before* aborting the scheduler (S6): the sink can be parked in a
17/// blocking `send()`, and only dropping the receiver unblocks it so the abort
18/// can join the workers. `Drop` may block until in-flight FFmpeg calls return
19/// (no fixed bound on stalled network IO).
20pub struct SampleIter {
21    rx: Option<Receiver<AudioChunk>>,
22    scheduler: Option<FfmpegScheduler<Running>>,
23    terminated: bool,
24}
25
26impl SampleIter {
27    pub(crate) fn new(rx: Receiver<AudioChunk>, scheduler: FfmpegScheduler<Running>) -> Self {
28        Self {
29            rx: Some(rx),
30            scheduler: Some(scheduler),
31            terminated: false,
32        }
33    }
34
35    /// Joins the scheduler exactly once and maps its result to at most one
36    /// terminal error. Called when the channel disconnects.
37    fn finish(&mut self) -> Option<Result<AudioChunk, Error>> {
38        self.terminated = true;
39        // Drop the receiver before joining so a parked sender is released.
40        self.rx = None;
41        match self.scheduler.take() {
42            Some(scheduler) => match scheduler.wait() {
43                Ok(()) => None,
44                Err(e) => Some(Err(e)),
45            },
46            None => None,
47        }
48    }
49}
50
51impl Iterator for SampleIter {
52    type Item = Result<AudioChunk, Error>;
53
54    fn next(&mut self) -> Option<Self::Item> {
55        if self.terminated {
56            return None;
57        }
58        let recv = match self.rx.as_ref() {
59            Some(rx) => rx.recv(),
60            None => return self.finish(),
61        };
62        match recv {
63            Ok(chunk) => Some(Ok(chunk)),
64            // Channel closed: the pipeline finished or died. Join once.
65            Err(_) => self.finish(),
66        }
67    }
68}
69
70impl std::iter::FusedIterator for SampleIter {}
71
72impl Drop for SampleIter {
73    fn drop(&mut self) {
74        // S6: drop the receiver FIRST (unblocks a sink parked in send()), THEN
75        // abort the scheduler (which joins the workers). Reversed order can
76        // deadlock — see FrameIter::drop for the full rationale.
77        self.rx = None;
78        if let Some(scheduler) = self.scheduler.take() {
79            scheduler.abort();
80        }
81    }
82}
ez_ffmpeg/core/frame_export/audio_iter.rs

ez_ffmpeg/core/frame_export/
audio_iter.rs
