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 cuts after every packet by default (one group per
15/// frame, forwarded immediately); a caller driving its own boundaries cuts less often. Opus' packet
16/// loss concealment handles drops.
17pub struct Import<E: CatalogExt = ()> {
18	track: crate::container::Producer<crate::catalog::hang::Container>,
19	rendition: crate::catalog::AudioTrack<E>,
20}
21
22impl<E: CatalogExt> Import<E> {
23	/// Publish on an existing track producer with a resolved catalog config.
24	///
25	/// Audio can't derive its config from frames, so the caller passes a complete
26	/// [`AudioConfig`](hang::catalog::AudioConfig) (build one from an OpusHead with [`config`], or
27	/// from an out-of-band [`Config`] via `into()`). The rendition publishes immediately.
28	pub fn new(
29		track: moq_net::track::Producer,
30		reserved: crate::catalog::Reserved<E>,
31		mut config: hang::catalog::AudioConfig,
32	) -> crate::Result<Self> {
33		tracing::debug!(name = ?track.name(), ?config, "starting track");
34		// Advertise this rendition's timeline before publishing (the generic set() no longer does).
35		config.timeline = Some(reserved.producer().timeline(track.name())?.section());
36		let mut rendition = reserved.audio(track.name());
37		rendition.set(config);
38		Ok(Self {
39			track: reserved
40				.producer()
41				.media_producer(track, crate::catalog::hang::Container::Legacy)?,
42			rendition,
43		})
44	}
45
46	/// The MoQ track name this importer publishes on.
47	pub fn name(&self) -> &str {
48		self.track.track().name()
49	}
50
51	/// A watch-only handle to this track's subscriber demand.
52	pub fn demand(&self) -> moq_net::track::Demand {
53		self.track.track().demand()
54	}
55
56	/// Finish the track, flushing the current group.
57	pub fn finish(&mut self) -> crate::Result<()> {
58		self.track.finish()?;
59		self.estimate();
60		Ok(())
61	}
62
63	/// Abort the track with `err` instead of finishing it cleanly, so subscribers
64	/// see the real cause rather than [`moq_net::Error::Dropped`]. Consumes this importer.
65	pub fn abort(self, err: moq_net::Error) {
66		self.track.abort(err);
67	}
68
69	/// Publish what the track measured (bitrate, jitter) into the catalog rendition, filling only
70	/// the fields its config didn't supply.
71	fn estimate(&mut self) {
72		self.rendition.estimate(self.track.estimate());
73	}
74
75	/// Cut the current group at `end` without finishing the track.
76	pub fn cut(&mut self, end: Option<moq_net::Timestamp>) -> crate::Result<()> {
77		self.track.cut(end)?;
78		self.estimate();
79		Ok(())
80	}
81
82	/// Mark a break in the timeline by publishing an empty group. To bound the closing
83	/// group's final frame first, [`cut(end)`](Self::cut) before this. See
84	/// [`Producer::discontinuity`](crate::container::Producer::discontinuity).
85	pub fn discontinuity(&mut self) -> crate::Result<()> {
86		self.track.discontinuity()?;
87		self.estimate();
88		Ok(())
89	}
90
91	/// Close the current group and open the next one at `sequence`.
92	pub fn seek(&mut self, sequence: u64) -> crate::Result<()> {
93		self.track.seek(sequence)?;
94		self.estimate();
95		Ok(())
96	}
97
98	/// Publish one Opus packet, stamping `pts` or a wall clock when absent.
99	///
100	/// Opus is independently decodable, so the packet is marked a keyframe only when it starts a group
101	/// (see [`Producer::needs_keyframe`](crate::container::Producer::needs_keyframe)); otherwise it
102	/// extends the current group. The caller bounds groups via [`cut`](Self::cut) / [`seek`](Self::seek).
103	pub fn decode<B: moq_net::IntoBytes>(&mut self, frame: B, pts: Option<moq_net::Timestamp>) -> crate::Result<()> {
104		let timestamp = self.rendition.timestamp(pts)?;
105		// Only the first frame of each group is a keyframe, so the group spans until the caller cuts
106		// instead of opening one group (one QUIC stream) per packet.
107		let keyframe = self.track.needs_keyframe();
108		self.track.write(Frame {
109			timestamp,
110			payload: frame.into_bytes(),
111			keyframe,
112			duration: None,
113		})?;
114		self.estimate();
115		Ok(())
116	}
117}
118
119/// Build a catalog config from an OpusHead. Errors on a malformed or empty buffer.
120pub fn config(init: &[u8]) -> crate::Result<hang::catalog::AudioConfig> {
121	let mut buf = init;
122	Ok(Config::parse(&mut buf)?.into())
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	fn from(config: Config) -> Self {
128		let mut audio = hang::catalog::AudioConfig::new(
129			hang::catalog::AudioCodec::Opus,
130			config.sample_rate,
131			config.channel_count,
132		);
133		audio.description = config.encode().ok();
134		audio.container = hang::catalog::Container::Legacy;
135		audio
136	}
137}
moq_mux/codec/opus/import.rs

moq_mux/codec/opus/
import.rs
