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::catalog::hang::Container>,
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		// The catalog says how the track is framed, and it is not always the legacy
46		// wire: `moq import fmp4` publishes CMAF. Reading a moof+mdat fragment as a
47		// varint timestamp plus a payload decodes to garbage rather than failing.
48		let container = moq_mux::catalog::hang::Container::try_from(&catalog.container)?;
49		let mut track = moq_mux::container::Consumer::new(track, container);
50		if let Some(latency) = config.latency_max {
51			track = track.with_latency(latency);
52		}
53
54		Ok(Self {
55			decoder,
56			track,
57			pending: VecDeque::new(),
58		})
59	}
60
61	/// The decoder backend name in use, e.g. `"videotoolbox"` or `"openh264"`.
62	pub fn name(&self) -> &str {
63		self.decoder.name()
64	}
65
66	/// Read the next decoded I420 frame, or `None` when the track ends.
67	pub async fn read(&mut self) -> Result<Option<Frame>, Error> {
68		loop {
69			if let Some(frame) = self.pending.pop_front() {
70				return Ok(Some(frame));
71			}
72
73			let Some(mux_frame) = self.track.read().await? else {
74				return Ok(None);
75			};
76
77			self.pending.extend(
78				self.decoder
79					.decode(mux_frame.payload, mux_frame.timestamp, mux_frame.keyframe)
80					.await?,
81			);
82		}
83	}
84}
85
86#[cfg(test)]
87mod tests {
88	use super::*;
89	use crate::decode::Kind;
90	use crate::encode::{Config as EncodeConfig, Encoder, Kind as EncodeKind, Producer as EncodeProducer};
91
92	#[tokio::test]
93	async fn reads_cmaf_container_declared_by_catalog() {
94		let mut source_broadcast = moq_net::broadcast::Info::new().produce();
95		let source_subscriber = source_broadcast.consume();
96		let source_catalog = moq_mux::catalog::Producer::new(&mut source_broadcast).unwrap();
97		let config = EncodeConfig {
98			kind: EncodeKind::Software,
99			..EncodeConfig::new(320, 240, 30)
100		};
101		let rendition = config.probe().await.unwrap();
102		let mut producer = EncodeProducer::new(source_broadcast, source_catalog, rendition).unwrap();
103		let mut encoder = Encoder::new(&config).unwrap();
104		let rgba = vec![0x80u8; 320 * 240 * 4];
105		for index in 0..2 {
106			encoder.keyframe();
107			let surface = crate::Surface::rgba(&rgba, crate::Size::new(320, 240)).unwrap();
108			let frame = crate::Frame::new(surface, moq_net::Timestamp::from_micros(index * 33_333).unwrap());
109			producer.publish(&encoder.encode(&frame).unwrap()).unwrap();
110		}
111
112		let origin = moq_net::Origin::random().produce();
113		let mut requests = origin.dynamic();
114		let served = source_subscriber.clone();
115		tokio::spawn(async move {
116			while let Ok(request) = requests.requested_broadcast().await {
117				request.accept(served.clone());
118			}
119		});
120		let catalog = moq_mux::catalog::Consumer::<()>::new(&source_subscriber, moq_mux::catalog::CatalogFormat::Hang)
121			.await
122			.unwrap();
123		let source = moq_mux::Source::new(origin.consume(), "test");
124		let mut export = moq_mux::container::fmp4::Export::new(source, catalog);
125		let init = export.next().await.unwrap().expect("CMAF init");
126		let fragment = export.next().await.unwrap().expect("CMAF fragment");
127
128		let mut broadcast = moq_net::broadcast::Info::new().produce();
129		let subscriber = broadcast.consume();
130		let catalog = moq_mux::catalog::Producer::new(&mut broadcast).unwrap();
131		let mut import = moq_mux::container::fmp4::Import::new(broadcast, catalog.reserve());
132		import.decode(&init).unwrap();
133		import.decode(&fragment).unwrap();
134
135		let snapshot = catalog.snapshot();
136		let (name, config) = snapshot.video.renditions.iter().next().expect("video rendition");
137		assert!(matches!(config.container, hang::catalog::Container::Cmaf { .. }));
138		let mut consumer = Consumer::new(
139			&subscriber,
140			config,
141			name,
142			Config {
143				kind: Kind::Software,
144				..Config::new()
145			},
146		)
147		.await
148		.unwrap();
149
150		let frame = consumer.read().await.unwrap().expect("decoded frame");
151		assert_eq!(frame.size(), crate::Size::new(320, 240));
152	}
153}
moq_video/decode/consumer.rs

moq_video/decode/
consumer.rs
