zencodec_media/
session.rs

1//! Codec sessions: packets in, frames out ([`VideoDecoder`]); frames in,
2//! packets out ([`VideoEncoder`]).
3//!
4//! The protocol, for both directions:
5//!
6//! - `send` offers one input. [`SendStatus::Accepted`] means the session
7//!   took it (a packet's payload is taken; a frame handle is cloned).
8//!   [`SendStatus::ReceivePending`] means "receive first": the input is left
9//!   untouched for the retry, and at least one `receive` now returns
10//!   [`Receive::Ready`]. That is the backpressure: a session never queues
11//!   without bound and never blocks.
12//! - `receive` returns one output, [`Receive::NeedInput`], or, after
13//!   `end_input` and once everything is drained, [`Receive::EndOfStream`],
14//!   which is stable.
15//! - `end_input` declares the end; it is idempotent. Sending after it is a
16//!   [`ProtocolViolation::SendAfterEnd`].
17//! - A decoder's `reset` discards queued input, output and references and
18//!   keeps the configuration; frames already returned stay valid. Seeking is
19//!   `reset` and `configure` (to re-inject a sequence header), then packets
20//!   from the new position.
21//!
22//! Sessions do no I/O, need no async runtime and know nothing about
23//! containers. Work calls take an `Option<&dyn zencodec::Pulse>`: `None` adds no
24//! per-call control and never disables a session's retained control or resource
25//! checks. [`FusedDecoder`] makes errors
26//! and cancellation terminal until a successful `reset`. [`FusedEncoder`]
27//! makes encoder errors terminal; a fresh encoder starts a new operation.
28
29use alloc::boxed::Box;
30use alloc::sync::Arc;
31use core::fmt;
32
33use enough::StopReason;
34pub use zencodec::decode::ErasedError;
35
36use crate::frame::{Frame, FrameInterpretation};
37use crate::limits::MediaLimitExceeded;
38use crate::packet::Packet;
39use crate::track::{CodecId, TrackInfo, VideoTrackInfo};
40use zencodec::time::TimeBase;
41
42#[derive(Clone, Copy, Debug, PartialEq, Eq)]
43#[non_exhaustive]
44/// What `send` did with an input.
45pub enum SendStatus {
46    /// Ownership moved (the packet is taken, or the frame handle was cloned).
47    Accepted,
48    /// Receive output first, then retry the unchanged input. At least one
49    /// `receive` after this returns `Ready`.
50    ReceivePending,
51}
52
53#[derive(Debug)]
54#[non_exhaustive]
55/// What `receive` returned.
56pub enum Receive<T> {
57    /// One output.
58    Ready(T),
59    /// Nothing now; more input may be sent.
60    NeedInput,
61    /// Input ended and everything was drained. Stable.
62    EndOfStream,
63}
64
65#[derive(Clone, Copy, Debug, PartialEq, Eq)]
66#[non_exhaustive]
67/// A caller or a session broke the protocol.
68pub enum ProtocolViolation {
69    /// `send` after `end_input`.
70    SendAfterEnd,
71    /// A packet without payload.
72    EmptyPacket,
73    /// `send` before `configure`.
74    NotConfigured,
75    /// A packet from another configuration epoch than the configured one.
76    EpochMismatch,
77    /// An input without the timestamp the session needs.
78    MissingTimestamp,
79    /// A timestamp on another clock than the session's.
80    ClockMismatch,
81    /// A second `begin`.
82    BeginTwice,
83    /// Output requested before `begin`.
84    NoBegin,
85    /// A session answered `NeedInput` where it had promised output or an
86    /// end (after `end_input`).
87    NoProgress,
88    /// `ReceivePending`, then no output from `receive`.
89    ReceivePendingWithoutOutput,
90    /// A second `finish`.
91    FinishTwice,
92    /// The operation needs an index the container did not provide.
93    IndexRequired,
94    /// `EndOfStream` before `end_input`.
95    EndOfStreamBeforeEndInput,
96}
97
98#[derive(Clone, Copy, Debug, PartialEq, Eq)]
99#[non_exhaustive]
100/// A valid request this implementation does not support.
101pub enum UnsupportedMedia {
102    /// Repositioning.
103    Seek,
104    /// A mid-stream configuration change.
105    ConfigChange,
106    /// A layout that needs no patches (forward-only destination).
107    ForwardOnlyLayout,
108    /// This codec.
109    Codec,
110    /// This container feature.
111    Container,
112    /// This sample layout or depth.
113    PixelFormat,
114    /// Exact preservation of packet timestamps and explicit durations.
115    ExactTiming,
116}
117
118/// Every session error is terminal for the session except `Codec` errors a
119/// backend documents as retryable validation.
120#[derive(Debug)]
121#[non_exhaustive]
122pub enum SessionError<E> {
123    /// The codec or container implementation failed (malformed input,
124    /// backend error).
125    Codec(E),
126    /// The pulse tripped.
127    Cancelled(StopReason),
128    /// Called again after a terminal error.
129    Finished,
130    /// The protocol was broken.
131    Protocol(ProtocolViolation),
132    /// A valid request this implementation does not support.
133    Unsupported(UnsupportedMedia),
134    /// A structural limit was exceeded.
135    Limit(MediaLimitExceeded),
136}
137
138impl<E> SessionError<E> {
139    /// Convert the implementation error, keeping every other variant.
140    pub fn map_codec<F>(self, f: impl FnOnce(E) -> F) -> SessionError<F> {
141        match self {
142            Self::Codec(e) => SessionError::Codec(f(e)),
143            Self::Cancelled(r) => SessionError::Cancelled(r),
144            Self::Finished => SessionError::Finished,
145            Self::Protocol(p) => SessionError::Protocol(p),
146            Self::Unsupported(u) => SessionError::Unsupported(u),
147            Self::Limit(l) => SessionError::Limit(l),
148        }
149    }
150}
151
152impl<E> From<StopReason> for SessionError<E> {
153    fn from(r: StopReason) -> Self {
154        Self::Cancelled(r)
155    }
156}
157
158impl<E: fmt::Display> fmt::Display for SessionError<E> {
159    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
160        match self {
161            Self::Codec(e) => write!(f, "{e}"),
162            Self::Cancelled(r) => write!(f, "cancelled: {r:?}"),
163            Self::Finished => f.write_str("session used after a terminal error"),
164            Self::Protocol(p) => write!(f, "protocol violation: {p:?}"),
165            Self::Unsupported(u) => write!(f, "unsupported: {u:?}"),
166            Self::Limit(l) => write!(f, "{l}"),
167        }
168    }
169}
170
171impl<E: core::error::Error + 'static> core::error::Error for SessionError<E> {
172    fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
173        match self {
174            Self::Codec(e) => Some(e),
175            Self::Limit(l) => Some(l),
176            _ => None,
177        }
178    }
179}
180
181/// What a decoder needs before the first packet of a configuration epoch:
182/// derived from a `TrackInfo`, or from a muxer's expectations.
183#[derive(Clone, Debug)]
184#[non_exhaustive]
185pub struct CodecConfig {
186    /// The codec.
187    pub codec: CodecId,
188    /// Codec configuration bytes (`av1C`, a sequence header, `avcC`).
189    pub private: Option<Arc<[u8]>>,
190    /// The epoch these bytes start.
191    pub epoch: u32,
192    /// The clock of the timestamps that follow.
193    pub time_base: TimeBase,
194    /// Declared video shape, when known.
195    pub video: Option<VideoTrackInfo>,
196}
197
198impl CodecConfig {
199    /// A configuration for `codec` on `time_base`, epoch 0.
200    pub fn new(codec: CodecId, time_base: TimeBase) -> Self {
201        Self {
202            codec,
203            private: None,
204            epoch: 0,
205            time_base,
206            video: None,
207        }
208    }
209    /// The declared video shape.
210    pub fn with_video(mut self, video: VideoTrackInfo) -> Self {
211        self.video = Some(video);
212        self
213    }
214
215    /// Codec configuration bytes.
216    pub fn with_private(mut self, private: Arc<[u8]>) -> Self {
217        self.private = Some(private);
218        self
219    }
220
221    /// The epoch.
222    pub fn with_epoch(mut self, epoch: u32) -> Self {
223        self.epoch = epoch;
224        self
225    }
226}
227
228impl From<&TrackInfo> for CodecConfig {
229    fn from(t: &TrackInfo) -> Self {
230        Self {
231            codec: t.codec().clone(),
232            private: t.codec_private().cloned(),
233            epoch: t.config_epoch(),
234            time_base: t.time_base(),
235            video: t.video_info().copied(),
236        }
237    }
238}
239
240/// Per-stream facts, known once the stream has been seen; not static flags.
241#[derive(Clone, Debug)]
242#[non_exhaustive]
243pub struct VideoStreamInfo {
244    /// The interpretation of the latest frame.
245    pub interpretation: Arc<FrameInterpretation>,
246    /// Packets the decoder may hold before their frame comes out, when the
247    /// stream declares it.
248    pub reorder_depth: Option<u32>,
249}
250
251impl VideoStreamInfo {
252    /// Facts with an interpretation and no declared reorder depth.
253    pub fn new(interpretation: Arc<FrameInterpretation>) -> Self {
254        Self {
255            interpretation,
256            reorder_depth: None,
257        }
258    }
259}
260
261/// A video decoder session: packets in, frames out, in presentation order.
262/// See the [module docs](self) for the protocol.
263pub trait VideoDecoder {
264    /// The implementation's error.
265    type Error: core::error::Error + Send + Sync + 'static;
266
267    /// Start a configuration epoch. Called before the first `send` and on
268    /// every epoch change (new sequence header, new `stsd` entry).
269    fn configure(
270        &mut self,
271        config: &CodecConfig,
272        pulse: Option<&dyn zencodec::Pulse>,
273    ) -> Result<(), SessionError<Self::Error>>;
274
275    /// Offer one packet. `Accepted` takes its payload; `ReceivePending`
276    /// leaves it intact for retry after `receive`.
277    fn send(
278        &mut self,
279        packet: &mut Packet,
280        pulse: Option<&dyn zencodec::Pulse>,
281    ) -> Result<SendStatus, SessionError<Self::Error>>;
282
283    /// One frame at a time, in presentation order.
284    fn receive(
285        &mut self,
286        pulse: Option<&dyn zencodec::Pulse>,
287    ) -> Result<Receive<Frame>, SessionError<Self::Error>>;
288
289    /// Declare end of input; idempotent. Then `receive` until `EndOfStream`.
290    fn end_input(&mut self) -> Result<(), SessionError<Self::Error>>;
291
292    /// Discard queued input, output and references; keep configuration.
293    /// Frames already returned stay valid.
294    fn reset(&mut self) -> Result<(), SessionError<Self::Error>>;
295
296    /// Facts about the stream seen so far.
297    fn stream_info(&self) -> Option<&VideoStreamInfo>;
298}
299
300/// A video encoder session: frames in, packets out, in decode order. See
301/// the [module docs](self) for the protocol.
302pub trait VideoEncoder {
303    /// The implementation's error.
304    type Error: core::error::Error + Send + Sync + 'static;
305
306    /// Offer a frame handle. The encoder clones it to queue (no pixel copy
307    /// is required by the contract; a backend may copy into padded planes).
308    fn send(
309        &mut self,
310        frame: &Frame,
311        pulse: Option<&dyn zencodec::Pulse>,
312    ) -> Result<SendStatus, SessionError<Self::Error>>;
313
314    /// One packet at a time, in decode order, carrying the presentation
315    /// timestamp of the frame it shows.
316    fn receive(
317        &mut self,
318        pulse: Option<&dyn zencodec::Pulse>,
319    ) -> Result<Receive<Packet>, SessionError<Self::Error>>;
320
321    /// Declare end of input; idempotent. Then `receive` until `EndOfStream`.
322    fn end_input(&mut self) -> Result<(), SessionError<Self::Error>>;
323
324    /// Codec configuration a muxer needs before the first packet.
325    fn output_config(&self) -> Option<&CodecConfig>;
326
327    /// Fidelity guaranteed by this actual prepared encoder for every accepted
328    /// input frame. None means unqualified, not lossy or lossless. A guarantee
329    /// must not be inferred from codec identity or an unvalidated request.
330    /// This describes encoding only; preceding conversion may lose samples.
331    /// Fixed-configuration operations bind this value with output_config and
332    /// reject a later change before publishing staged output.
333    fn output_fidelity(&self) -> Option<zencodec::encode::Fidelity> {
334        None
335    }
336}
337
338/// A type-erased, sendable decoder session.
339pub type DynVideoDecoder = dyn VideoDecoder<Error = ErasedError> + Send;
340/// A type-erased, sendable encoder session.
341pub type DynVideoEncoder = dyn VideoEncoder<Error = ErasedError> + Send;
342
343fn erase<E: core::error::Error + Send + Sync + 'static>(
344    e: SessionError<E>,
345) -> SessionError<ErasedError> {
346    e.map_codec(|c| ErasedError::new(Box::new(c)))
347}
348
349/// Erases a decoder's error type.
350pub struct VideoDecoderShim<D>(pub D);
351
352impl<D: VideoDecoder> VideoDecoder for VideoDecoderShim<D> {
353    type Error = ErasedError;
354    fn configure(
355        &mut self,
356        config: &CodecConfig,
357        pulse: Option<&dyn zencodec::Pulse>,
358    ) -> Result<(), SessionError<ErasedError>> {
359        self.0.configure(config, pulse).map_err(erase)
360    }
361    fn send(
362        &mut self,
363        packet: &mut Packet,
364        pulse: Option<&dyn zencodec::Pulse>,
365    ) -> Result<SendStatus, SessionError<ErasedError>> {
366        self.0.send(packet, pulse).map_err(erase)
367    }
368    fn receive(
369        &mut self,
370        pulse: Option<&dyn zencodec::Pulse>,
371    ) -> Result<Receive<Frame>, SessionError<ErasedError>> {
372        self.0.receive(pulse).map_err(erase)
373    }
374    fn end_input(&mut self) -> Result<(), SessionError<ErasedError>> {
375        self.0.end_input().map_err(erase)
376    }
377    fn reset(&mut self) -> Result<(), SessionError<ErasedError>> {
378        self.0.reset().map_err(erase)
379    }
380    fn stream_info(&self) -> Option<&VideoStreamInfo> {
381        self.0.stream_info()
382    }
383}
384
385/// Erases an encoder's error type.
386pub struct VideoEncoderShim<E>(pub E);
387
388impl<E: VideoEncoder> VideoEncoder for VideoEncoderShim<E> {
389    type Error = ErasedError;
390    fn send(
391        &mut self,
392        frame: &Frame,
393        pulse: Option<&dyn zencodec::Pulse>,
394    ) -> Result<SendStatus, SessionError<ErasedError>> {
395        self.0.send(frame, pulse).map_err(erase)
396    }
397    fn receive(
398        &mut self,
399        pulse: Option<&dyn zencodec::Pulse>,
400    ) -> Result<Receive<Packet>, SessionError<ErasedError>> {
401        self.0.receive(pulse).map_err(erase)
402    }
403    fn end_input(&mut self) -> Result<(), SessionError<ErasedError>> {
404        self.0.end_input().map_err(erase)
405    }
406    fn output_config(&self) -> Option<&CodecConfig> {
407        self.0.output_config()
408    }
409    fn output_fidelity(&self) -> Option<zencodec::encode::Fidelity> {
410        self.0.output_fidelity()
411    }
412}
413
414/// Makes errors and cancellation terminal for any decoder, as `Fused` does
415/// for `IncrementalDecode`. A successful `reset` reopens the session.
416pub struct FusedDecoder<D> {
417    inner: D,
418    failed: bool,
419}
420
421impl<D> FusedDecoder<D> {
422    /// Wrap `inner`.
423    pub const fn new(inner: D) -> Self {
424        Self {
425            inner,
426            failed: false,
427        }
428    }
429    /// The wrapped decoder.
430    pub fn into_inner(self) -> D {
431        self.inner
432    }
433}
434
435impl<D: VideoDecoder> FusedDecoder<D> {
436    fn guard<T>(
437        &mut self,
438        pulse: &dyn zencodec::Pulse,
439        f: impl FnOnce(&mut D) -> Result<T, SessionError<D::Error>>,
440    ) -> Result<T, SessionError<D::Error>> {
441        if self.failed {
442            return Err(SessionError::Finished);
443        }
444        self.failed = true;
445        if let Err(r) = pulse.check() {
446            return Err(SessionError::Cancelled(r));
447        }
448        let r = f(&mut self.inner);
449        self.failed = r.is_err();
450        r
451    }
452}
453
454impl<D: VideoDecoder> VideoDecoder for FusedDecoder<D> {
455    type Error = D::Error;
456    fn configure(
457        &mut self,
458        config: &CodecConfig,
459        pulse: Option<&dyn zencodec::Pulse>,
460    ) -> Result<(), SessionError<D::Error>> {
461        let local_pulse = pulse.unwrap_or(&zencodec::NoPulse);
462        self.guard(local_pulse, |d| d.configure(config, pulse))
463    }
464    fn send(
465        &mut self,
466        packet: &mut Packet,
467        pulse: Option<&dyn zencodec::Pulse>,
468    ) -> Result<SendStatus, SessionError<D::Error>> {
469        let local_pulse = pulse.unwrap_or(&zencodec::NoPulse);
470        self.guard(local_pulse, |d| d.send(packet, pulse))
471    }
472    fn receive(
473        &mut self,
474        pulse: Option<&dyn zencodec::Pulse>,
475    ) -> Result<Receive<Frame>, SessionError<D::Error>> {
476        let local_pulse = pulse.unwrap_or(&zencodec::NoPulse);
477        self.guard(local_pulse, |d| d.receive(pulse))
478    }
479    fn end_input(&mut self) -> Result<(), SessionError<D::Error>> {
480        if self.failed {
481            return Err(SessionError::Finished);
482        }
483        self.failed = true;
484        let r = self.inner.end_input();
485        self.failed = r.is_err();
486        r
487    }
488    fn reset(&mut self) -> Result<(), SessionError<D::Error>> {
489        self.failed = true;
490        let r = self.inner.reset();
491        self.failed = r.is_err();
492        r
493    }
494    fn stream_info(&self) -> Option<&VideoStreamInfo> {
495        self.inner.stream_info()
496    }
497}
498
499/// Makes every encoder error and cancellation terminal, including native
500/// validation errors that an unwrapped backend might otherwise permit retrying.
501///
502/// Unlike a decoder, an encoder has no reset operation: create a fresh session
503/// after failure. Informational `output_config` remains available. If a backend
504/// unwinds and the application catches the panic, later work calls still refuse.
505pub struct FusedEncoder<E> {
506    inner: E,
507    failed: bool,
508}
509
510impl<E> FusedEncoder<E> {
511    /// Wrap an encoder with strict terminal-error handling.
512    pub const fn new(inner: E) -> Self {
513        Self {
514            inner,
515            failed: false,
516        }
517    }
518
519    /// Recover the backend, explicitly leaving the fused contract.
520    pub fn into_inner(self) -> E {
521        self.inner
522    }
523}
524
525impl<E: VideoEncoder> FusedEncoder<E> {
526    fn guard<T>(
527        &mut self,
528        pulse: &dyn zencodec::Pulse,
529        f: impl FnOnce(&mut E) -> Result<T, SessionError<E::Error>>,
530    ) -> Result<T, SessionError<E::Error>> {
531        if self.failed {
532            return Err(SessionError::Finished);
533        }
534        // Set before entering either user-supplied operation so an unwinding
535        // backend/pulse cannot accidentally reopen partially changed state.
536        self.failed = true;
537        pulse.check().map_err(SessionError::Cancelled)?;
538        let result = f(&mut self.inner);
539        self.failed = result.is_err();
540        result
541    }
542}
543
544impl<E: VideoEncoder> VideoEncoder for FusedEncoder<E> {
545    type Error = E::Error;
546    fn send(
547        &mut self,
548        frame: &Frame,
549        pulse: Option<&dyn zencodec::Pulse>,
550    ) -> Result<SendStatus, SessionError<E::Error>> {
551        let local_pulse = pulse.unwrap_or(&zencodec::NoPulse);
552        self.guard(local_pulse, |encoder| encoder.send(frame, pulse))
553    }
554    fn receive(
555        &mut self,
556        pulse: Option<&dyn zencodec::Pulse>,
557    ) -> Result<Receive<Packet>, SessionError<E::Error>> {
558        let local_pulse = pulse.unwrap_or(&zencodec::NoPulse);
559        self.guard(local_pulse, |encoder| encoder.receive(pulse))
560    }
561    fn end_input(&mut self) -> Result<(), SessionError<E::Error>> {
562        self.guard(&zencodec::how_far::NoPulse, VideoEncoder::end_input)
563    }
564    fn output_config(&self) -> Option<&CodecConfig> {
565        self.inner.output_config()
566    }
567    fn output_fidelity(&self) -> Option<zencodec::encode::Fidelity> {
568        self.inner.output_fidelity()
569    }
570}
571
572/// Erase a decoder: errors become [`ErasedError`], and errors and
573/// cancellation become terminal until `reset` ([`FusedDecoder`]).
574pub fn erase_video_decoder<D: VideoDecoder + Send + 'static>(d: D) -> Box<DynVideoDecoder> {
575    Box::new(VideoDecoderShim(FusedDecoder::new(d)))
576}
577
578/// Erase an encoder's error type and make all errors/cancellation terminal
579/// ([`FusedEncoder`]). Use [`VideoEncoderShim`] directly when preserving a
580/// backend's explicitly documented retryable validation contract is required.
581pub fn erase_video_encoder<E: VideoEncoder + Send + 'static>(e: E) -> Box<DynVideoEncoder> {
582    Box::new(VideoEncoderShim(FusedEncoder::new(e)))
583}
584
585impl<D: VideoDecoder + ?Sized> VideoDecoder for Box<D> {
586    type Error = D::Error;
587    fn configure(
588        &mut self,
589        config: &CodecConfig,
590        pulse: Option<&dyn zencodec::Pulse>,
591    ) -> Result<(), SessionError<D::Error>> {
592        (**self).configure(config, pulse)
593    }
594    fn send(
595        &mut self,
596        packet: &mut Packet,
597        pulse: Option<&dyn zencodec::Pulse>,
598    ) -> Result<SendStatus, SessionError<D::Error>> {
599        (**self).send(packet, pulse)
600    }
601    fn receive(
602        &mut self,
603        pulse: Option<&dyn zencodec::Pulse>,
604    ) -> Result<Receive<Frame>, SessionError<D::Error>> {
605        (**self).receive(pulse)
606    }
607    fn end_input(&mut self) -> Result<(), SessionError<D::Error>> {
608        (**self).end_input()
609    }
610    fn reset(&mut self) -> Result<(), SessionError<D::Error>> {
611        (**self).reset()
612    }
613    fn stream_info(&self) -> Option<&VideoStreamInfo> {
614        (**self).stream_info()
615    }
616}
617
618impl<E: VideoEncoder + ?Sized> VideoEncoder for Box<E> {
619    type Error = E::Error;
620    fn send(
621        &mut self,
622        frame: &Frame,
623        pulse: Option<&dyn zencodec::Pulse>,
624    ) -> Result<SendStatus, SessionError<E::Error>> {
625        (**self).send(frame, pulse)
626    }
627    fn receive(
628        &mut self,
629        pulse: Option<&dyn zencodec::Pulse>,
630    ) -> Result<Receive<Packet>, SessionError<E::Error>> {
631        (**self).receive(pulse)
632    }
633    fn end_input(&mut self) -> Result<(), SessionError<E::Error>> {
634        (**self).end_input()
635    }
636    fn output_config(&self) -> Option<&CodecConfig> {
637        (**self).output_config()
638    }
639    fn output_fidelity(&self) -> Option<zencodec::encode::Fidelity> {
640        (**self).output_fidelity()
641    }
642}
643
644/// The backpressure loop every driver writes once: send, and while the codec
645/// says `ReceivePending`, receive output into `on_output` and retry; then
646/// deliver whatever output is ready.
647///
648/// # Errors
649///
650/// The decoder's error as `Upstream`, `on_output`'s as `Downstream`.
651pub fn send_packet_until_accepted<D, F, E2>(
652    dec: &mut D,
653    packet: &mut Packet,
654    pulse: Option<&dyn zencodec::Pulse>,
655    mut on_output: F,
656) -> Result<(), PumpError<SessionError<D::Error>, E2>>
657where
658    D: VideoDecoder + ?Sized,
659    F: FnMut(Frame) -> Result<(), E2>,
660{
661    loop {
662        match dec.send(packet, pulse).map_err(PumpError::Upstream)? {
663            SendStatus::Accepted => break,
664            SendStatus::ReceivePending => match dec.receive(pulse).map_err(PumpError::Upstream)? {
665                Receive::Ready(f) => on_output(f).map_err(PumpError::Downstream)?,
666                _ => {
667                    return Err(PumpError::Upstream(SessionError::Protocol(
668                        ProtocolViolation::ReceivePendingWithoutOutput,
669                    )));
670                }
671            },
672        }
673    }
674    // Accepted: deliver whatever output is available now, so frames never
675    // sit in the codec until the next backpressure event.
676    loop {
677        match dec.receive(pulse).map_err(PumpError::Upstream)? {
678            Receive::Ready(f) => on_output(f).map_err(PumpError::Downstream)?,
679            Receive::NeedInput => return Ok(()),
680            Receive::EndOfStream => {
681                return Err(PumpError::Upstream(SessionError::Protocol(
682                    ProtocolViolation::EndOfStreamBeforeEndInput,
683                )));
684            }
685        }
686    }
687}
688
689/// The encoder twin of [`send_packet_until_accepted`].
690///
691/// # Errors
692///
693/// The encoder's error as `Upstream`, `on_output`'s as `Downstream`.
694pub fn send_frame_until_accepted<E, F, E2>(
695    enc: &mut E,
696    frame: &Frame,
697    pulse: Option<&dyn zencodec::Pulse>,
698    mut on_output: F,
699) -> Result<(), PumpError<SessionError<E::Error>, E2>>
700where
701    E: VideoEncoder + ?Sized,
702    F: FnMut(Packet) -> Result<(), E2>,
703{
704    loop {
705        match enc.send(frame, pulse).map_err(PumpError::Upstream)? {
706            SendStatus::Accepted => break,
707            SendStatus::ReceivePending => match enc.receive(pulse).map_err(PumpError::Upstream)? {
708                Receive::Ready(p) => on_output(p).map_err(PumpError::Downstream)?,
709                _ => {
710                    return Err(PumpError::Upstream(SessionError::Protocol(
711                        ProtocolViolation::ReceivePendingWithoutOutput,
712                    )));
713                }
714            },
715        }
716    }
717    loop {
718        match enc.receive(pulse).map_err(PumpError::Upstream)? {
719            Receive::Ready(p) => on_output(p).map_err(PumpError::Downstream)?,
720            Receive::NeedInput => return Ok(()),
721            Receive::EndOfStream => {
722                return Err(PumpError::Upstream(SessionError::Protocol(
723                    ProtocolViolation::EndOfStreamBeforeEndInput,
724                )));
725            }
726        }
727    }
728}
729
730/// End input and drain: every `Ready` goes to `on_output` until
731/// `EndOfStream`.
732///
733/// # Errors
734///
735/// The decoder's error as `Upstream`, `on_output`'s as `Downstream`.
736pub fn drain_decoder<D, F, E2>(
737    dec: &mut D,
738    pulse: Option<&dyn zencodec::Pulse>,
739    mut on_output: F,
740) -> Result<(), PumpError<SessionError<D::Error>, E2>>
741where
742    D: VideoDecoder + ?Sized,
743    F: FnMut(Frame) -> Result<(), E2>,
744{
745    dec.end_input().map_err(PumpError::Upstream)?;
746    loop {
747        match dec.receive(pulse).map_err(PumpError::Upstream)? {
748            Receive::Ready(f) => on_output(f).map_err(PumpError::Downstream)?,
749            Receive::EndOfStream => return Ok(()),
750            Receive::NeedInput => {
751                return Err(PumpError::Upstream(SessionError::Protocol(
752                    ProtocolViolation::NoProgress,
753                )));
754            }
755        }
756    }
757}
758
759/// The encoder twin of [`drain_decoder`].
760///
761/// # Errors
762///
763/// The encoder's error as `Upstream`, `on_output`'s as `Downstream`.
764pub fn drain_encoder<E, F, E2>(
765    enc: &mut E,
766    pulse: Option<&dyn zencodec::Pulse>,
767    mut on_output: F,
768) -> Result<(), PumpError<SessionError<E::Error>, E2>>
769where
770    E: VideoEncoder + ?Sized,
771    F: FnMut(Packet) -> Result<(), E2>,
772{
773    enc.end_input().map_err(PumpError::Upstream)?;
774    loop {
775        match enc.receive(pulse).map_err(PumpError::Upstream)? {
776            Receive::Ready(p) => on_output(p).map_err(PumpError::Downstream)?,
777            Receive::EndOfStream => return Ok(()),
778            Receive::NeedInput => {
779                return Err(PumpError::Upstream(SessionError::Protocol(
780                    ProtocolViolation::NoProgress,
781                )));
782            }
783        }
784    }
785}
786
787/// Where a pump failed: the stage being driven, or the stage it feeds.
788#[derive(Debug)]
789pub enum PumpError<U, D> {
790    /// The stage being driven failed.
791    Upstream(U),
792    /// The stage it feeds failed.
793    Downstream(D),
794}
795
796impl<U: fmt::Display, D: fmt::Display> fmt::Display for PumpError<U, D> {
797    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
798        match self {
799            Self::Upstream(e) => write!(f, "upstream: {e}"),
800            Self::Downstream(e) => write!(f, "downstream: {e}"),
801        }
802    }
803}
804impl<U: core::error::Error + 'static, D: core::error::Error + 'static> core::error::Error
805    for PumpError<U, D>
806{
807    fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
808        match self {
809            Self::Upstream(error) => Some(error),
810            Self::Downstream(error) => Some(error),
811        }
812    }
813}
zencodec_media/session.rs

zencodec_media/
session.rs
