moq_audio/decode/
consumer.rs

1//! Subscribe to an encoded audio track and emit raw PCM.
2
3use bytes::Bytes;
4
5use super::decoder::{Config, Decoder};
6use crate::resample::Resampler;
7use crate::{Error, Frame};
8
9/// Subscribe to a moq-mux audio track and emit decoded PCM in the layout
10/// declared by [`Config`].
11///
12/// The mirror of [`encode::Producer`](crate::encode::Producer): output format /
13/// sample rate / channel count are fixed at construction, and
14/// [`read`](Self::read) returns plain [`Frame`]s.
15pub struct Consumer {
16	decoder: Decoder,
17	track: moq_mux::container::Consumer<moq_mux::container::legacy::Wire>,
18	resampler: Option<Resampler>,
19	config: Config,
20	resolved_sample_rate: u32,
21	resolved_channels: u32,
22}
23
24impl Consumer {
25	/// Subscribe to `name` in `broadcast`, using the catalog entry to pick the
26	/// codec.
27	pub async fn new(
28		broadcast: &moq_net::broadcast::Consumer,
29		catalog: &hang::catalog::AudioConfig,
30		name: impl Into<String>,
31		config: Config,
32	) -> Result<Self, Error> {
33		let decoder = Decoder::new(catalog)?;
34		let sample_rate = config.sample_rate.unwrap_or_else(|| decoder.sample_rate());
35		let channels = config.channels.unwrap_or_else(|| decoder.channel_count());
36
37		if channels != decoder.channel_count() {
38			return Err(Error::Unsupported(format!(
39				"channel remapping not implemented (decoder {}ch, requested {channels}ch)",
40				decoder.channel_count()
41			)));
42		}
43
44		let resampler = if sample_rate == decoder.sample_rate() {
45			None
46		} else {
47			let chunk_frames = (decoder.sample_rate() as usize * 20) / 1000;
48			Some(Resampler::new(
49				decoder.sample_rate(),
50				sample_rate,
51				decoder.channel_count(),
52				chunk_frames,
53			)?)
54		};
55
56		let name = name.into();
57		let track = broadcast
58			.track(&name)?
59			.subscribe(moq_net::track::Subscription::default().with_priority(hang::catalog::PRIORITY.audio))
60			.await?;
61		let mut track = moq_mux::container::Consumer::new(track, moq_mux::container::legacy::Wire);
62		if let Some(latency) = config.latency_max {
63			track = track.with_latency(latency);
64		}
65
66		Ok(Self {
67			decoder,
68			track,
69			resampler,
70			config,
71			resolved_sample_rate: sample_rate,
72			resolved_channels: channels,
73		})
74	}
75
76	/// The config this consumer was built with.
77	pub fn config(&self) -> &Config {
78		&self.config
79	}
80
81	/// Sample rate samples are actually delivered at, which is
82	/// [`Config::sample_rate`] resolved against the catalog.
83	pub fn sample_rate(&self) -> u32 {
84		self.resolved_sample_rate
85	}
86
87	/// Channel count samples are actually delivered at, which is
88	/// [`Config::channels`] resolved against the catalog.
89	pub fn channels(&self) -> u32 {
90		self.resolved_channels
91	}
92
93	/// Read the next decoded PCM frame, or `None` when the track ends.
94	pub async fn read(&mut self) -> Result<Option<Frame>, Error> {
95		let Some(mux_frame) = self.track.read().await? else {
96			return Ok(None);
97		};
98
99		let decoded = self.decoder.decode(&mux_frame.payload)?;
100		let pcm = match self.resampler.as_mut() {
101			Some(r) => r.process(&decoded)?,
102			None => decoded,
103		};
104
105		let bytes = self.config.format.from_interleaved_f32(&pcm, self.resolved_channels)?;
106		Ok(Some(Frame {
107			timestamp: mux_frame.timestamp,
108			data: Bytes::from(bytes),
109		}))
110	}
111}
moq_audio/decode/consumer.rs

moq_audio/decode/
consumer.rs
