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

//! `client publish`: dial a remote WHIP endpoint and push a MoQ broadcast
//! out as RTP.
//!
//! Mints an SDP offer with `sendonly` audio + video, POSTs it to the WHIP
//! resource URL, parses the returned answer, and hands the resulting
//! `str0m::Rtc` to the internal session driver running in egress mode. The
//! bitstream / RTP packetization is identical to the WHEP server path, so
//! most of the work lives in the crate-private egress source.

use str0m::{
	Candidate,
	change::SdpAnswer,
	format::Codec,
	media::{Direction, MediaKind},
};
use url::Url;

use crate::{Error, Result, client::Client, egress::EgressSource, session};

pub(crate) async fn dial(
	client: &Client,
	url: Url,
	origin: moq_net::origin::Consumer,
	path: impl moq_net::AsPath,
) -> Result<()> {
	let source = EgressSource::new(moq_mux::Source::new(origin, path)).await?;
	let codecs = source.catalog_codecs();
	if codecs.is_empty() {
		return Err(Error::NoRenditions);
	}

	let (socket, candidates) = session::bind_udp(&client.config().ice_candidates).await?;
	// Restrict to codecs the catalog can actually source so the remote
	// answer doesn't pick a codec we have no rendition for.
	let mut rtc = session::rtc_with_codecs(&codecs);
	for addr in &candidates {
		let cand = Candidate::host(*addr, "udp").map_err(Error::rtc)?;
		rtc.add_local_candidate(cand);
	}

	// Only advertise media kinds the catalog can source. str0m emits a rejected
	// m-line with no payload types when a kind has no enabled codecs, which is not
	// valid SDP and cannot be parsed by the WHIP server.
	let mut api = rtc.sdp_api();
	let mut mids = Vec::new();
	if codecs.iter().any(|codec| matches!(codec, Codec::Opus)) {
		mids.push(api.add_media(MediaKind::Audio, Direction::SendOnly, None, None, None));
	}
	if codecs
		.iter()
		.any(|codec| matches!(codec, Codec::H264 | Codec::H265 | Codec::Vp8 | Codec::Vp9 | Codec::Av1))
	{
		mids.push(api.add_media(MediaKind::Video, Direction::SendOnly, None, None, None));
	}
	let (offer, pending) = api.apply().ok_or(Error::NoSdpChanges)?;

	let res = client
		.http()
		.post(url.clone())
		.header(reqwest::header::CONTENT_TYPE, "application/sdp")
		.header(reqwest::header::ACCEPT, "application/sdp")
		.body(offer.to_sdp_string())
		.send()
		.await?;

	if !res.status().is_success() {
		return Err(Error::HttpStatus(res.status().as_u16()));
	}

	let body = res.text().await?;
	let answer = SdpAnswer::from_sdp_string(&body).map_err(|err| Error::InvalidSdp(err.to_string()))?;

	rtc.sdp_api().accept_answer(pending, answer).map_err(Error::rtc)?;
	tracing::info!(%url, "whip client connected");

	// 1:1 socket (no demux on the client): pump its datagrams into the session.
	// The session tags each datagram with the advertised candidate matching its
	// family (str0m matches the destination against a host candidate, not the bind).
	let inbound = session::spawn_socket_reader(socket.clone());
	let mut session = session::Session::egress(rtc, socket, candidates, inbound, source);
	for mid in mids {
		session.open_media(mid)?;
	}
	tokio::spawn(async move {
		let result = session.run().await;
		session::log_session_end("whip client", &result);
	});

	Ok(())
}