moq_mux/codec/opus/
import.rs

1use super::Config;
2use crate::catalog::hang::CatalogExt;
3use crate::container::Frame;
4
5/// Opus importer.
6///
7/// Publishes raw Opus frames (no Ogg framing) to a single moq track. Build it with
8/// [`new`](Self::new), passing the track producer and the
9/// [`catalog::Reserved`](crate::catalog::Reserved) it reserves its rendition from.
10///
11/// Every Opus packet is independently decodable, so [`decode`](Self::decode) marks only the first
12/// frame of each group a keyframe (the rest extend it): frames accumulate into the current group
13/// until the caller [`cut`](Self::cut)s or [`seek`](Self::seek)s. The
14/// [`import::Track`](crate::import::Track) facade passes that through, so boundaries are the
15/// caller's either way: cut per packet for one group (one QUIC stream) forwarded immediately, or at
16/// a segment cadence to align with video. Opus' packet loss concealment handles drops.
17pub struct Import {
18	track: crate::container::Producer<crate::catalog::hang::Container, hang::catalog::AudioConfig>,
19}
20
21impl Import {
22	/// Publish on an existing track producer with a resolved catalog config.
23	///
24	/// Audio can't derive its config from frames, so the caller passes a complete
25	/// [`AudioConfig`](hang::catalog::AudioConfig) (build one from an OpusHead with [`config`], or
26	/// from an out-of-band [`Config`] via `into()`). The rendition publishes immediately.
27	pub fn new<E: CatalogExt>(
28		track: moq_net::track::Producer,
29		reserved: crate::catalog::Reserved<E>,
30		config: hang::catalog::AudioConfig,
31	) -> crate::Result<Self> {
32		tracing::debug!(name = ?track.name(), ?config, "starting track");
33		// The caller's config names the container; the writer is built from that same value so the
34		// wire cannot disagree with what the rendition advertises.
35		let wire = crate::catalog::hang::Container::try_from(&config)?;
36		let track = reserved.audio(track, wire, config)?;
37		Ok(Self { track })
38	}
39
40	/// The MoQ track name this importer publishes on.
41	pub fn name(&self) -> &str {
42		self.track.track().name()
43	}
44
45	/// The exclusive presentation end earlier groups have reached, if any.
46	pub(crate) fn live_edge(&self) -> Option<moq_net::Timestamp> {
47		self.track.live_edge()
48	}
49
50	/// A watch-only handle to this track's subscriber demand.
51	pub fn demand(&self) -> moq_net::track::Demand {
52		self.track.track().demand()
53	}
54
55	/// Record a locally encoded frame's transport handoff for catalog jitter measurement.
56	pub fn flush(&mut self, timestamp: moq_net::Timestamp, now: std::time::Instant) -> crate::Result<()> {
57		self.track.flush(timestamp, now)
58	}
59
60	/// Finish the track, flushing the current group.
61	pub fn finish(&mut self) -> crate::Result<()> {
62		self.track.finish()?;
63		Ok(())
64	}
65
66	/// Abort the track with `err` instead of finishing it cleanly, so subscribers
67	/// see the real cause rather than [`moq_net::Error::Dropped`]. Consumes this importer.
68	pub fn abort(self, err: moq_net::Error) {
69		self.track.abort(err);
70	}
71
72	/// Publish what the track measured (bitrate, jitter) into the catalog rendition, filling only
73	/// the fields its config didn't supply.
74	/// Cut the current group at `end` without finishing the track.
75	pub fn cut(&mut self, end: Option<moq_net::Timestamp>) -> crate::Result<()> {
76		self.track.cut(end)?;
77		Ok(())
78	}
79
80	/// Mark a break in the timeline by publishing a marker group. To bound the closing
81	/// group's final frame first, [`cut(end)`](Self::cut) before this. See
82	/// [`Producer::discontinuity`](crate::container::Producer::discontinuity).
83	pub fn discontinuity(&mut self) -> crate::Result<()> {
84		self.track.discontinuity()?;
85		Ok(())
86	}
87
88	/// Close the current group and open the next one at `sequence`.
89	pub fn seek(&mut self, sequence: u64) -> crate::Result<()> {
90		self.track.seek(sequence)?;
91		Ok(())
92	}
93
94	/// Publish one Opus packet, stamping `pts` or a wall clock when absent.
95	///
96	/// Opus is independently decodable, so the packet is marked a keyframe only when it starts a group
97	/// (see [`Producer::needs_keyframe`](crate::container::Producer::needs_keyframe)); otherwise it
98	/// extends the current group. The caller bounds groups via [`cut`](Self::cut) / [`seek`](Self::seek).
99	pub fn decode<B: moq_net::IntoBytes>(&mut self, frame: B, pts: Option<moq_net::Timestamp>) -> crate::Result<()> {
100		let timestamp = self.track.timestamp(pts)?;
101		// Only the first frame of each group is a keyframe, so the group spans until the caller cuts
102		// instead of opening one group (one QUIC stream) per packet.
103		let keyframe = self.track.needs_keyframe();
104		self.track.write(Frame {
105			timestamp,
106			payload: frame.into_bytes(),
107			keyframe,
108			duration: None,
109		})?;
110		Ok(())
111	}
112}
113
114/// Build a catalog config from an OpusHead. Errors on a malformed or empty buffer.
115///
116/// The catalog rate is the decoder's 48 kHz output, not the OpusHead's informational input rate.
117pub fn config(init: &[u8]) -> crate::Result<hang::catalog::AudioConfig> {
118	let mut buf = init;
119	let mut config: hang::catalog::AudioConfig = Config::parse(&mut buf)?.into();
120	// Publish the head as given: a re-encode would rewrite its version byte.
121	config.description = Some(bytes::Bytes::copy_from_slice(init));
122	Ok(config)
123}
124
125impl From<Config> for hang::catalog::AudioConfig {
126	/// Build a catalog config from a config resolved out of band (e.g. gstreamer caps).
127	///
128	/// The catalog describes the decoder's output: Opus always decodes at 48 kHz, whatever the
129	/// informational input rate in [`Config::sample_rate`] claims.
130	fn from(config: Config) -> Self {
131		let mut audio = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Opus, 48_000, config.channel_count);
132		audio.description = config.encode().ok();
133		audio.container = hang::catalog::Container::Legacy;
134		audio
135	}
136}
137
138#[cfg(test)]
139mod tests {
140	use std::time::Duration;
141
142	use moq_net::Timestamp;
143
144	#[test]
145	fn catalog_rate_is_decoder_rate() {
146		// A 44.1 kHz input rate is informational; the description keeps it, the catalog does not.
147		let head = super::Config::new(44_100, 2).encode().unwrap();
148		let config = super::config(&head).unwrap();
149		assert_eq!(config.sample_rate, 48_000);
150		assert_eq!(config.description.as_deref(), Some(head.as_ref()));
151
152		let config: hang::catalog::AudioConfig = super::Config::new(44_100, 2).into();
153		assert_eq!(config.sample_rate, 48_000);
154	}
155
156	#[tokio::test(start_paused = true)]
157	async fn a_loc_reservation_reaches_the_wire_and_the_catalog() {
158		let mut broadcast = moq_net::broadcast::Info::new().produce();
159		let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();
160		let track = broadcast
161			.create_track("audio", hang::container::track_info(hang::catalog::PRIORITY.audio))
162			.unwrap();
163		let subscriber = track.subscribe(None);
164		let reserved = catalog.reserve();
165		let config = crate::codec::opus::Config::new(48_000, 2);
166		let mut config: hang::catalog::AudioConfig = config.into();
167		config.container = hang::catalog::Container::Loc;
168		let mut import = super::Import::new(track, reserved, config).unwrap();
169
170		let audio = catalog.snapshot().audio.renditions.get("audio").cloned().unwrap();
171		assert_eq!(audio.container, hang::catalog::Container::Loc);
172
173		let payload = b"opus payload";
174		import
175			.decode(payload, Some(Timestamp::from_micros(1_000).unwrap()))
176			.unwrap();
177		let mut media = crate::container::Consumer::new(
178			subscriber,
179			crate::catalog::hang::Container::Loc(crate::container::Kind::Data),
180		);
181		let frame = tokio::time::timeout(Duration::from_secs(1), media.read())
182			.await
183			.unwrap()
184			.unwrap()
185			.unwrap();
186		assert_eq!(frame.payload.as_ref(), payload);
187		assert_eq!(frame.timestamp, Timestamp::from_micros(1_000).unwrap());
188
189		import.finish().unwrap();
190	}
191}
moq_mux/codec/opus/import.rs

moq_mux/codec/opus/
import.rs
