ez_ffmpeg/core/frame_export/
iter.rs

1//! [`FrameIter`]: a fused iterator over exported frames with load-bearing
2//! teardown ordering (S6).
3
4use super::error::FrameExportError;
5use super::frame::VideoFrame;
6use crate::core::scheduler::ffmpeg_scheduler::{FfmpegScheduler, Running};
7use crate::core::scheduler::owned_run_iter::OwnedRunIter;
8use crate::error::Error;
9use crossbeam_channel::Receiver;
10
11/// Re-surfaces a typed [`FrameExportError`] that crossed the frame-filter
12/// boundary. The module's own filters (color guard, UniformN sampler) return
13/// their typed errors boxed, which the pipeline wraps as
14/// [`Error::FrameFilterProcess`]; unwrapping here keeps the public contract —
15/// runtime failures match `Error::FrameExport(..)` exactly like open-time
16/// ones. Anything else passes through unchanged.
17fn map_terminal_error(e: Error) -> Error {
18    match e {
19        Error::FrameFilterProcess(boxed) => match boxed.downcast::<FrameExportError>() {
20            Ok(typed) => Error::FrameExport(*typed),
21            Err(other) => Error::FrameFilterProcess(other),
22        },
23        other => other,
24    }
25}
26
27/// An iterator over exported [`VideoFrame`]s.
28///
29/// Yields `Ok(frame)` per exported frame. When the pipeline finishes (or dies),
30/// the iterator joins the scheduler once: a worker error surfaces as a single
31/// terminal `Err`, after which the iterator is fused and yields `None` forever.
32///
33/// Dropping the iterator early tears the run down cleanly. Teardown drops the
34/// receiver *before* aborting the scheduler (S6): the sink can be parked in a
35/// blocking `send()`, and only dropping the receiver unblocks it so the abort
36/// can join the workers. `Drop` may block until in-flight FFmpeg calls return
37/// (no fixed bound on stalled network IO).
38pub struct FrameIter {
39    inner: OwnedRunIter<VideoFrame>,
40}
41
42impl FrameIter {
43    pub(crate) fn new(rx: Receiver<VideoFrame>, scheduler: FfmpegScheduler<Running>) -> Self {
44        Self {
45            inner: OwnedRunIter::new(rx, scheduler, map_terminal_error),
46        }
47    }
48}
49
50impl Iterator for FrameIter {
51    type Item = Result<VideoFrame, Error>;
52
53    fn next(&mut self) -> Option<Self::Item> {
54        self.inner.next()
55    }
56}
57
58impl std::iter::FusedIterator for FrameIter {}
59
60#[cfg(test)]
61mod tests {
62    use super::*;
63
64    #[test]
65    fn typed_export_error_is_unwrapped_from_filter_process() {
66        let boxed: Box<dyn std::error::Error + Send + Sync> =
67            Box::new(FrameExportError::HdrRequiresToneMapping);
68        let mapped = map_terminal_error(Error::FrameFilterProcess(boxed));
69        assert!(matches!(
70            mapped,
71            Error::FrameExport(FrameExportError::HdrRequiresToneMapping)
72        ));
73    }
74
75    #[test]
76    fn foreign_filter_process_error_passes_through() {
77        let boxed: Box<dyn std::error::Error + Send + Sync> = "some other failure".into();
78        let mapped = map_terminal_error(Error::FrameFilterProcess(boxed));
79        assert!(matches!(mapped, Error::FrameFilterProcess(_)));
80    }
81
82    #[test]
83    fn non_filter_errors_pass_through() {
84        let mapped = map_terminal_error(Error::EOF);
85        assert!(matches!(mapped, Error::EOF));
86    }
87}
ez_ffmpeg/core/frame_export/iter.rs

ez_ffmpeg/core/frame_export/
iter.rs
