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/// Each packet handed to [`decode`](Self::decode) is published in its own group so
12/// the relay can forward it immediately without waiting for a group boundary; Opus'
13/// packet loss concealment handles drops.
14pub struct Import<E: CatalogExt = ()> {
15	track: crate::container::Producer<crate::catalog::hang::Container>,
16	rendition: crate::catalog::AudioTrack<E>,
17}
18
19impl<E: CatalogExt> Import<E> {
20	/// Publish on an existing track producer with a resolved catalog config.
21	///
22	/// Audio can't derive its config from frames, so the caller passes a complete
23	/// [`AudioConfig`](hang::catalog::AudioConfig) (build one from an OpusHead with [`config`], or
24	/// from an out-of-band [`Config`] via `into()`). The rendition publishes immediately.
25	pub fn new(
26		track: moq_net::track::Producer,
27		reserved: crate::catalog::Reserved<E>,
28		mut config: hang::catalog::AudioConfig,
29	) -> crate::Result<Self> {
30		tracing::debug!(name = ?track.name(), ?config, "starting track");
31		// Advertise this rendition's timeline before publishing (the generic set() no longer does).
32		config.timeline = Some(reserved.producer().timeline(track.name())?.section());
33		let mut rendition = reserved.audio(track.name());
34		rendition.set(config);
35		Ok(Self {
36			track: reserved
37				.producer()
38				.media_producer(track, crate::catalog::hang::Container::Legacy)?,
39			rendition,
40		})
41	}
42
43	/// The MoQ track name this importer publishes on.
44	pub fn name(&self) -> &str {
45		self.track.track().name()
46	}
47
48	/// A watch-only handle to this track's subscriber demand.
49	pub fn demand(&self) -> moq_net::track::Demand {
50		self.track.track().demand()
51	}
52
53	/// Finish the track, flushing the current group.
54	pub fn finish(&mut self) -> crate::Result<()> {
55		self.rendition.record_group_end(None);
56		self.track.finish()?;
57		Ok(())
58	}
59
60	/// Abort the track with `err` instead of finishing it cleanly, so subscribers
61	/// see the real cause rather than [`moq_net::Error::Dropped`]. Consumes this importer.
62	pub fn abort(self, err: moq_net::Error) {
63		self.track.abort(err);
64	}
65
66	/// Cut the current group at `end` without finishing the track.
67	pub fn cut(&mut self, end: Option<moq_net::Timestamp>) -> crate::Result<()> {
68		self.rendition.record_group_end(end);
69		self.track.cut(end)?;
70		Ok(())
71	}
72
73	/// Close the current group and open the next one at `sequence`.
74	pub fn seek(&mut self, sequence: u64) -> crate::Result<()> {
75		self.rendition.record_group_end(None);
76		self.track.seek(sequence)?;
77		Ok(())
78	}
79
80	/// Publish one Opus packet as its own group, stamping `pts` or a wall clock when absent.
81	pub fn decode<B: moq_net::IntoBytes>(&mut self, frame: B, pts: Option<moq_net::Timestamp>) -> crate::Result<()> {
82		let timestamp = self.rendition.timestamp(pts)?;
83		self.rendition.record_group_end(Some(timestamp));
84		let bytes = frame.as_ref().len();
85		self.track.write(Frame {
86			timestamp,
87			payload: frame.into_bytes(),
88			keyframe: true,
89			duration: None,
90		})?;
91		self.track.cut(None)?;
92		self.rendition.record_frame(timestamp, bytes);
93		Ok(())
94	}
95}
96
97/// Build a catalog config from an OpusHead. Errors on a malformed or empty buffer.
98pub fn config(init: &[u8]) -> crate::Result<hang::catalog::AudioConfig> {
99	let mut buf = init;
100	Ok(Config::parse(&mut buf)?.into())
101}
102
103impl From<Config> for hang::catalog::AudioConfig {
104	/// Build a catalog config from a config resolved out of band (e.g. gstreamer caps).
105	fn from(config: Config) -> Self {
106		let mut audio = hang::catalog::AudioConfig::new(
107			hang::catalog::AudioCodec::Opus,
108			config.sample_rate,
109			config.channel_count,
110		);
111		audio.container = hang::catalog::Container::Legacy;
112		audio
113	}
114}
moq_mux/codec/opus/import.rs

moq_mux/codec/opus/
import.rs
