zencodec_media/
output.rs

1//! Checked synchronous container output for forward writers and seekable files.
2//!
3//! [`PreparedOutput::new`](crate::output::PreparedOutput::new) requires exact packet timing and owns the actual begun
4//! muxer and destination without publishing its header. [`PreparedOutput::publish`](crate::output::PreparedOutput::publish)
5//! consumes that preparation. [`Output::new`](crate::output::Output::new) combines those steps for callers
6//! that do not need a policy decision between them. Unsupported timing is an
7//! error, never a fallback to container-chosen timing.
8//!
9//! ```
10//! use zencodec_media::{container::Muxer, output::{Failure, Forward, Output, Stats},
11//!     packet::Packet, track::TrackInfo};
12//! fn write<M: Muxer, W: std::io::Write>(
13//!     muxer: M, writer: W, tracks: &[TrackInfo], packets: impl IntoIterator<Item = Packet>,
14//! ) -> Result<Stats, Failure> {
15//!     let mut output = Output::new(muxer, Forward(writer), tracks, None)?;
16//!     for packet in packets { output.send(packet, None)?; }
17//!     output.finish(None)
18//! }
19//! ```
20//!
21//! To ask policy before any write, replace `Output::new` with
22//! [`PreparedOutput::new`](crate::output::PreparedOutput::new), retain the result while deciding, then call
23//! [`PreparedOutput::publish`](crate::output::PreparedOutput::publish). Use [`SeekBack::new`](crate::output::SeekBack::new) instead of [`Forward`](crate::output::Forward) for
24//! a file whose current position is the start of this output.
25//!
26//! These drivers may block on `Write`. Byte availability does not establish safe
27//! encoder backpressure: finish or separately qualify upstream raw-buffer retention
28//! before sending packets to a slow writer. No whole-operation memory bound is
29//! implied. Dropping either state does not implicitly finish or publish.
30//! `None` for a borrowed Pulse adds no per-call control and preserves native
31//! retained controls, timing validation and destination checks.
32
33use crate::{
34    container::{MuxHint, MuxTiming, Muxer},
35    packet::Packet,
36    session::{ErasedError, ProtocolViolation, SessionError, UnsupportedMedia},
37    track::TrackInfo,
38};
39use alloc::{boxed::Box, vec::Vec};
40
41use std::{
42    error::Error,
43    fmt,
44    io::{self, Write},
45};
46use zencodec::operation::DestinationFacts;
47
48type Cause = Box<dyn Error + Send + Sync>;
49
50/// Shared destination wrappers; media and raster routes use the same origin rules.
51pub use zencodec_io::destination::{Destination, Forward, SeekBack};
52
53#[derive(Clone, Copy, Debug)]
54#[non_exhaustive]
55/// Stage at which output stopped.
56pub enum Stage {
57    /// Validate and begin the retained muxer before publication.
58    Prepare,
59    /// Validate or process a packet, including draining its output.
60    Packet,
61    /// Publish the prepared header.
62    Drain,
63    /// Finalize, append the tail or flush.
64    Finish,
65    /// Apply or seek around a final patch.
66    Patch,
67}
68
69#[derive(Debug)]
70#[non_exhaustive]
71/// Terminal failure with original cause and confirmed append progress.
72/// `emitted` excludes overwrites; a failed patch may have partial effects.
73pub struct Failure {
74    /// Stage of the failure.
75    pub stage: Stage,
76    /// Confirmed appended bytes, including any partial failed write.
77    pub emitted: u64,
78    /// Original I/O, native codec, cancellation or protocol error.
79    pub cause: Cause,
80}
81impl fmt::Display for Failure {
82    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
83        write!(
84            f,
85            "mux {:?} failed after {} appended bytes",
86            self.stage, self.emitted
87        )
88    }
89}
90impl Error for Failure {
91    fn source(&self) -> Option<&(dyn Error + 'static)> {
92        Some(&*self.cause)
93    }
94}
95
96#[derive(Default, Debug)]
97#[non_exhaustive]
98/// Successful output progress; byte counts are not memory measurements.
99pub struct Stats {
100    /// Confirmed appended bytes, including any partial failed write.
101    pub emitted: u64,
102    /// Number of completed finish patches.
103    pub patches: usize,
104    /// Total bytes overwritten by finish patches.
105    pub patch_bytes: usize,
106    /// Largest observed pending byte slice, not an allocation or RSS bound.
107    pub max_pending: usize,
108}
109
110/// Active output driver. Any `send` failure or unwind makes further sends fail.
111/// `finish` consumes the driver; dropping it never retries effects.
112pub struct Output<M, D> {
113    mux: Option<M>,
114    destination: D,
115    tracks: Vec<TrackInfo>,
116    timing: MuxTiming,
117    exact_timing: bool,
118    stats: Stats,
119}
120
121/// Owns the actual begun muxer and destination without publishing its header.
122/// Dropping it abandons only private preparation. Publication consumes the token.
123pub struct PreparedOutput<M, D>(Output<M, D>);
124impl<M: Muxer, D: Destination> PreparedOutput<M, D> {
125    /// Require exact packet timing from the actual muxer, or fail before writing.
126    pub fn new(
127        mux: M,
128        destination: D,
129        tracks: &[TrackInfo],
130        pulse: Option<&dyn zencodec::Pulse>,
131    ) -> Result<Self, Failure> {
132        let pulse = pulse.unwrap_or(&zencodec::NoPulse);
133        Output::prepare_unpublished(mux, destination, tracks, pulse, true, M::begin_exact_timing)
134            .map(Self)
135    }
136    /// Timing/layout of the retained muxer. Streaming bytes do not certify safe
137    /// backpressure against an upstream encoder's raw-buffer ownership.
138    pub const fn timing(&self) -> MuxTiming {
139        self.0.timing
140    }
141    /// Facts of this exact preparation for policy and diagnostics. No bytes are
142    /// written, and the prepared muxer/writer remain owned by this object.
143    pub fn facts(&self) -> DestinationFacts {
144        self.0.facts()
145    }
146    pub(crate) fn validate_packet(&self, packet: &Packet) -> Result<(), Cause> {
147        self.0.validate_packet(packet)
148    }
149    /// Publish the retained header; any failure consumes the preparation.
150    pub fn publish(self, pulse: Option<&dyn zencodec::Pulse>) -> Result<Output<M, D>, Failure> {
151        let pulse = pulse.unwrap_or(&zencodec::NoPulse);
152        let mut output = self.0;
153        let mut mux = output.mux.take().expect("prepared muxer is retained");
154        output
155            .drain(&mut mux, pulse)
156            .map_err(|cause| output.failure(Stage::Drain, cause))?;
157        output.mux = Some(mux);
158        Ok(output)
159    }
160}
161
162fn stopped(pulse: &dyn zencodec::Pulse) -> Result<(), Cause> {
163    pulse
164        .check()
165        .map_err(|reason| Box::new(SessionError::<ErasedError>::Cancelled(reason)) as Cause)
166}
167fn protocol(kind: ProtocolViolation) -> Cause {
168    Box::new(SessionError::<ErasedError>::Protocol(kind))
169}
170
171fn write_all(
172    writer: &mut impl Write,
173    mut bytes: &[u8],
174    count: &mut u64,
175    pulse: &dyn zencodec::Pulse,
176) -> Result<(), Cause> {
177    while !bytes.is_empty() {
178        stopped(pulse)?;
179        match writer.write(bytes) {
180            Ok(0) => return Err(io::Error::from(io::ErrorKind::WriteZero).into()),
181            Ok(n) if n <= bytes.len() => {
182                *count = count
183                    .checked_add(n as u64)
184                    .ok_or_else(|| io::Error::from(io::ErrorKind::InvalidData))?;
185                bytes = &bytes[n..];
186            }
187            Ok(_) => return Err(io::Error::from(io::ErrorKind::InvalidData).into()),
188            Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
189            Err(e) => return Err(e.into()),
190        }
191    }
192    stopped(pulse)
193}
194
195impl<M: Muxer, D: Destination> Output<M, D> {
196    /// Explicitly allow the muxer's ordinary timing behavior, then publish.
197    /// This does not guarantee exact packet durations. Prefer [`Output::new`](crate::output::Output::new).
198    pub fn with_container_timing(
199        mux: M,
200        destination: D,
201        tracks: &[TrackInfo],
202        pulse: Option<&dyn zencodec::Pulse>,
203    ) -> Result<Self, Failure> {
204        Self::prepare(mux, destination, tracks, pulse, false, M::begin)
205    }
206
207    /// Require exact packet timing from the actual muxer, or fail before writing.
208    pub fn new(
209        mux: M,
210        destination: D,
211        tracks: &[TrackInfo],
212        pulse: Option<&dyn zencodec::Pulse>,
213    ) -> Result<Self, Failure> {
214        Self::prepare(mux, destination, tracks, pulse, true, M::begin_exact_timing)
215    }
216
217    fn prepare(
218        mux: M,
219        destination: D,
220        tracks: &[TrackInfo],
221        pulse: Option<&dyn zencodec::Pulse>,
222        exact_timing: bool,
223        begin: impl FnOnce(&mut M, &[TrackInfo], MuxHint) -> Result<MuxTiming, SessionError<M::Error>>,
224    ) -> Result<Self, Failure> {
225        PreparedOutput(Self::prepare_unpublished(
226            mux,
227            destination,
228            tracks,
229            pulse.unwrap_or(&zencodec::NoPulse),
230            exact_timing,
231            begin,
232        )?)
233        .publish(pulse)
234    }
235
236    fn prepare_unpublished(
237        mut mux: M,
238        destination: D,
239        tracks: &[TrackInfo],
240        pulse: &dyn zencodec::Pulse,
241        exact_timing: bool,
242        begin: impl FnOnce(&mut M, &[TrackInfo], MuxHint) -> Result<MuxTiming, SessionError<M::Error>>,
243    ) -> Result<Self, Failure> {
244        let prepare = || -> Result<_, Cause> {
245            stopped(pulse)?;
246            let hint = if destination.can_patch() {
247                MuxHint::default()
248            } else {
249                MuxHint::forward_only()
250            };
251            let timing = begin(&mut mux, tracks, hint)?;
252            if timing.patches_at_finish() && !destination.can_patch() {
253                return Err(Box::new(SessionError::<ErasedError>::Unsupported(
254                    UnsupportedMedia::ForwardOnlyLayout,
255                )));
256            }
257            stopped(pulse)?;
258            Ok(timing)
259        };
260        let timing = prepare().map_err(|cause| Failure {
261            stage: Stage::Prepare,
262            emitted: 0,
263            cause,
264        })?;
265        Ok(Self {
266            mux: Some(mux),
267            destination,
268            tracks: tracks.to_vec(),
269            timing,
270            exact_timing,
271            stats: Stats::default(),
272        })
273    }
274    /// Facts of the retained container/writer, unchanged by publication.
275    /// Progress and partial effects are reported separately through stats/errors.
276    pub fn facts(&self) -> DestinationFacts {
277        DestinationFacts::new(
278            self.destination.can_patch(),
279            self.timing.streams(),
280            self.timing.patches_at_finish(),
281            self.exact_timing,
282        )
283    }
284    fn validate_packet(&self, packet: &Packet) -> Result<(), Cause> {
285        let track = self
286            .tracks
287            .iter()
288            .find(|track| track.index() == packet.track())
289            .ok_or_else(|| protocol(ProtocolViolation::NotConfigured))?;
290        if packet.config_epoch() != track.config_epoch() {
291            return Err(protocol(ProtocolViolation::EpochMismatch));
292        }
293        for timestamp in [packet.pts(), packet.dts()].into_iter().flatten() {
294            if timestamp.base() != track.time_base() {
295                return Err(protocol(ProtocolViolation::ClockMismatch));
296            }
297        }
298        Ok(())
299    }
300    fn failure(&self, stage: Stage, cause: Cause) -> Failure {
301        Failure {
302            stage,
303            emitted: self.stats.emitted,
304            cause,
305        }
306    }
307    fn drain(&mut self, mux: &mut M, pulse: &dyn zencodec::Pulse) -> Result<(), Cause> {
308        let bytes = mux.pending_output();
309        self.stats.max_pending = self.stats.max_pending.max(bytes.len());
310        write_all(&mut self.destination, bytes, &mut self.stats.emitted, pulse)?;
311        mux.consume_output(bytes.len());
312        Ok(())
313    }
314    /// Confirmed progress so far, including before a terminal failure.
315    pub const fn stats(&self) -> &Stats {
316        &self.stats
317    }
318    /// Whether a prior failure or unwind made this driver terminal.
319    pub fn is_failed(&self) -> bool {
320        self.mux.is_none()
321    }
322    /// Validate and write one packet. Failure is terminal and preserves its cause.
323    pub fn send(
324        &mut self,
325        packet: Packet,
326        pulse: Option<&dyn zencodec::Pulse>,
327    ) -> Result<(), Failure> {
328        let local_pulse = pulse.unwrap_or(&zencodec::NoPulse);
329        let mut mux = self.mux.take().ok_or_else(|| {
330            self.failure(
331                Stage::Packet,
332                Box::new(SessionError::<ErasedError>::Finished),
333            )
334        })?;
335        let work = || -> Result<(), Cause> {
336            stopped(local_pulse)?;
337            self.validate_packet(&packet)?;
338            mux.write_packet(packet, pulse)?;
339            self.drain(&mut mux, local_pulse)?;
340            Ok(())
341        };
342        work().map_err(|cause| self.failure(Stage::Packet, cause))?;
343        self.mux = Some(mux);
344        Ok(())
345    }
346    /// Preflight final patches, append the tail, apply patches and flush.
347    /// A patch I/O failure can leave partial overwrites; it is not a transaction.
348    pub fn finish(mut self, pulse: Option<&dyn zencodec::Pulse>) -> Result<Stats, Failure> {
349        let local_pulse = pulse.unwrap_or(&zencodec::NoPulse);
350        let mux = self.mux.take().ok_or_else(|| {
351            self.failure(
352                Stage::Finish,
353                Box::new(SessionError::<ErasedError>::Finished),
354            )
355        })?;
356        stopped(local_pulse).map_err(|cause| self.failure(Stage::Finish, cause))?;
357        let finish = mux
358            .finish(pulse)
359            .map_err(|e| self.failure(Stage::Finish, Box::new(e)))?;
360        let total = finish
361            .validate(self.stats.emitted)
362            .map_err(|e| self.failure(Stage::Finish, Box::new(e)))?;
363        if !finish.patches.is_empty()
364            && (!self.destination.can_patch() || !self.timing.patches_at_finish())
365        {
366            return Err(self.failure(
367                Stage::Finish,
368                Box::new(SessionError::<ErasedError>::Unsupported(
369                    UnsupportedMedia::ForwardOnlyLayout,
370                )),
371            ));
372        }
373        self.stats.max_pending = self.stats.max_pending.max(finish.tail.len());
374        write_all(
375            &mut self.destination,
376            &finish.tail,
377            &mut self.stats.emitted,
378            local_pulse,
379        )
380        .map_err(|cause| self.failure(Stage::Finish, cause))?;
381        self.stats.patches = finish.patches.len();
382        for patch in finish.patches {
383            self.destination
384                .position(patch.offset)
385                .map_err(|e| self.failure(Stage::Patch, Box::new(e)))?;
386            let mut written = 0;
387            write_all(
388                &mut self.destination,
389                &patch.bytes,
390                &mut written,
391                local_pulse,
392            )
393            .map_err(|cause| self.failure(Stage::Patch, cause))?;
394            self.stats.patch_bytes += patch.bytes.len();
395        }
396        if self.stats.patches > 0 {
397            self.destination
398                .position(total)
399                .map_err(|e| self.failure(Stage::Patch, Box::new(e)))?;
400        }
401        self.destination
402            .flush()
403            .map_err(|e| self.failure(Stage::Finish, Box::new(e)))?;
404        stopped(local_pulse).map_err(|cause| self.failure(Stage::Finish, cause))?;
405        Ok(self.stats)
406    }
407}
zencodec_media/output.rs

zencodec_media/
output.rs
