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

moq_video/decode/
consumer.rs
