moq-rtc 0.2.4

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

//! VP8 bridge.
//!
//! str0m hands us complete VP8 frames, which is exactly the raw shape that
//! [`moq_mux::codec::vp8::Import`] consumes. The shared importer parses keyframes
//! so the catalog carries the encoded dimensions and stays in sync if they change.

use crate::{Result, codec};

/// Bridges str0m VP8 frames into a MoQ VP8 track.
pub struct Bridge {
	import: codec::DeferredVideo<moq_mux::codec::vp8::Import>,
}

impl Bridge {
	/// Publish a `.vp8` track on `broadcast`, adding the catalog rendition once config is known.
	pub fn new(broadcast: moq_net::broadcast::Producer, catalog: moq_mux::catalog::Producer) -> Result<Self> {
		let import = codec::DeferredVideo::new(broadcast, catalog, ".vp8")?;
		Ok(Self { import })
	}
}

impl codec::Bridge for Bridge {
	fn push(&mut self, frame: codec::Frame) -> Result<()> {
		let pts = moq_net::Timestamp::from_micros(frame.timestamp_us)
			.map_err(|err| crate::Error::Other(anyhow::anyhow!("invalid timestamp: {err}")))?;
		self.import.decode(frame.payload, pts)
	}

	fn abort(self: Box<Self>, err: moq_net::Error) {
		self.import.abort(err);
	}
}

#[cfg(test)]
mod tests {
	use bytes::Bytes;

	use crate::codec::{self, Bridge as _};

	#[test]
	fn keyframe_publishes_catalog_dimensions() {
		let mut broadcast = moq_net::broadcast::Info::new().produce();
		let catalog = moq_mux::catalog::Producer::new(&mut broadcast).unwrap();
		let mut bridge = super::Bridge::new(broadcast, catalog.clone()).unwrap();

		assert!(catalog.snapshot().video.renditions.is_empty());

		// VP8 keyframe header for 320x240.
		bridge
			.push(codec::Frame {
				timestamp_us: 0,
				payload: Bytes::from_static(&[0x10, 0x00, 0x00, 0x9d, 0x01, 0x2a, 0x40, 0x01, 0xf0, 0x00]),
			})
			.unwrap();

		let snapshot = catalog.snapshot();
		let config = snapshot.video.renditions.values().next().unwrap();
		assert_eq!(config.coded_width, Some(320));
		assert_eq!(config.coded_height, Some(240));
	}

	#[tokio::test]
	async fn importer_creation_failure_preserves_abort_error() {
		let mut broadcast = moq_net::broadcast::Info::new().produce();
		let catalog = moq_mux::catalog::Producer::new(&mut broadcast).unwrap();
		let _collision = broadcast.create_track("0.vp8.timeline.z", None).unwrap();
		let consumer = broadcast.consume();
		let mut bridge = super::Bridge::new(broadcast, catalog).unwrap();
		let mut track = consumer.track("0.vp8").unwrap().subscribe(None).await.unwrap();

		let result = bridge.push(codec::Frame {
			timestamp_us: 0,
			payload: Bytes::from_static(&[0x10, 0x00, 0x00, 0x9d, 0x01, 0x2a, 0x40, 0x01, 0xf0, 0x00]),
		});
		assert!(result.is_err(), "timeline collision must fail importer creation");

		Box::new(bridge).abort(moq_net::Error::Transport("session failed".into()));
		let Err(error) = track.recv_group().await else {
			panic!("aborted track must fail");
		};
		assert!(matches!(
			error,
			moq_net::Error::Transport(message) if message == "session failed"
		));
	}
}