moq_video/decode/
consumer.rs

1//! Subscribe to an encoded H.264, H.265, or AV1 track and emit raw I420 frames.
2
3use std::collections::VecDeque;
4
5use hang::catalog::VideoConfig;
6
7use super::decoder::{Config, Decoder};
8use crate::Error;
9use crate::Frame;
10
11/// Subscribe to a moq-mux video track and emit decoded I420.
12///
13/// The codec/backend are fixed at construction; [`read`](Self::read) returns
14/// plain [`Frame`]s. The direct mirror of `moq_audio::decode::Consumer`.
15pub struct Consumer {
16	decoder: Decoder,
17	track: moq_mux::container::Consumer<moq_mux::container::legacy::Wire>,
18	/// Frames a single access unit decoded to but `read` hasn't returned yet.
19	/// One AU yields one frame in the low-delay path, but a backend may hand back
20	/// more, so we buffer to keep `read` one-frame-per-call.
21	pending: VecDeque<Frame>,
22}
23
24impl Consumer {
25	/// Subscribe to `name` in `broadcast`, decoding it per the catalog entry.
26	/// Errors if the rendition's codec is not supported by a native backend.
27	pub async fn new(
28		broadcast: &moq_net::broadcast::Consumer,
29		catalog: &VideoConfig,
30		name: impl Into<String>,
31		config: Config,
32	) -> Result<Self, Error> {
33		let decoder = Decoder::new(catalog, &config)?;
34
35		let name = name.into();
36		let track = broadcast
37			.track(&name)?
38			.subscribe(moq_net::track::Subscription::default().with_priority(hang::catalog::PRIORITY.video))
39			.await?;
40		let mut track = moq_mux::container::Consumer::new(track, moq_mux::container::legacy::Wire);
41		if let Some(latency) = config.latency_max {
42			track = track.with_latency(latency);
43		}
44
45		Ok(Self {
46			decoder,
47			track,
48			pending: VecDeque::new(),
49		})
50	}
51
52	/// The decoder backend name in use, e.g. `"videotoolbox"` or `"openh264"`.
53	pub fn name(&self) -> &str {
54		self.decoder.name()
55	}
56
57	/// Read the next decoded I420 frame, or `None` when the track ends.
58	pub async fn read(&mut self) -> Result<Option<Frame>, Error> {
59		loop {
60			if let Some(frame) = self.pending.pop_front() {
61				return Ok(Some(frame));
62			}
63
64			let Some(mux_frame) = self.track.read().await? else {
65				return Ok(None);
66			};
67
68			self.pending.extend(
69				self.decoder
70					.decode(&mux_frame.payload, mux_frame.timestamp, mux_frame.keyframe)?,
71			);
72		}
73	}
74}
moq_video/decode/consumer.rs

moq_video/decode/
consumer.rs
