moq-rtc 0.3.1

WebRTC (WHIP/WHEP) gateway for Media over QUIC
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

//! Per-broadcast [`MediaSink`](crate::session::MediaSink) used by every
//! RTP-in flow (`server publish` / WHIP server, `client subscribe` / WHEP
//! client).
//!
//! Holds the [`moq_net::broadcast::Producer`] and per-track codec bridges.
//! On `MediaAdded`, it inspects the negotiated codec and instantiates the
//! matching bridge; on each `MediaData`, it forwards into the bridge.

use crate::{Error, Result, codec, session};

pub struct IngestSink {
	broadcast: moq_net::broadcast::Producer,
	catalog: moq_mux::catalog::Producer,
	bridges: session::Bridges,
}

impl IngestSink {
	pub fn new(mut broadcast: moq_net::broadcast::Producer, config: moq_mux::catalog::Config) -> Result<Self> {
		let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config)?;
		Ok(Self {
			broadcast,
			catalog,
			bridges: session::Bridges::new(),
		})
	}
}

impl session::MediaSink for IngestSink {
	fn on_track(
		&mut self,
		mid: str0m::media::Mid,
		_kind: str0m::media::MediaKind,
		codec_kind: str0m::format::Codec,
		audio_params: Option<(u32, u32)>,
	) -> Result<()> {
		let bridge: Box<dyn codec::Bridge> = match codec_kind {
			str0m::format::Codec::Opus => {
				let (sample_rate, channels) = audio_params.unwrap_or((48_000, 2));
				Box::new(codec::opus::Bridge::new(
					self.broadcast.clone(),
					self.catalog.clone(),
					sample_rate,
					channels,
				)?)
			}
			str0m::format::Codec::H264 => {
				Box::new(codec::h264::Bridge::new(self.broadcast.clone(), self.catalog.clone())?)
			}
			str0m::format::Codec::H265 => {
				Box::new(codec::h265::Bridge::new(self.broadcast.clone(), self.catalog.clone())?)
			}
			str0m::format::Codec::Vp8 => {
				Box::new(codec::vp8::Bridge::new(self.broadcast.clone(), self.catalog.clone())?)
			}
			str0m::format::Codec::Vp9 => {
				Box::new(codec::vp9::Bridge::new(self.broadcast.clone(), self.catalog.clone())?)
			}
			str0m::format::Codec::Av1 => {
				Box::new(codec::av1::Bridge::new(self.broadcast.clone(), self.catalog.clone())?)
			}
			other => return Err(Error::UnsupportedCodec(format!("{other:?}"))),
		};
		self.bridges.insert(mid, bridge);
		Ok(())
	}

	fn on_frame(&mut self, mid: str0m::media::Mid, frame: codec::Frame) -> Result<()> {
		self.bridges.push(mid, frame)
	}

	fn tick(&mut self) -> Result<()> {
		self.bridges.tick()
	}

	fn abort(&mut self, err: moq_net::Error) {
		self.bridges.abort(err);
	}
}

#[cfg(test)]
mod tests {
	use std::time::Duration;

	use super::*;
	use crate::session::MediaSink;

	/// The retention the caller configured has to land on the media tracks the codec
	/// bridges mint, not just on the catalog producer it was set on.
	#[tokio::test]
	async fn tracks_inherit_the_configured_retention() {
		let broadcast = moq_net::broadcast::Info::new().produce();
		let mut sink = IngestSink::new(
			broadcast.clone(),
			moq_mux::catalog::Config::default().with_max_age(Duration::from_secs(3)),
		)
		.unwrap();

		sink.on_track(
			"0".into(),
			str0m::media::MediaKind::Video,
			str0m::format::Codec::H264,
			None,
		)
		.unwrap();

		let info = broadcast.consume().track("0.avc3").unwrap().query().await.unwrap();
		assert_eq!(info.max_age, Duration::from_secs(3));
	}
}