moq-rtc 0.3.3

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

//! `client subscribe`: dial a remote WHEP endpoint, ingest RTP into a
//! local moq-net broadcast.
//!
//! Mints an SDP offer with `recvonly` audio and video, POSTs it to the
//! WHEP resource URL with `Content-Type: application/sdp`, parses the
//! returned answer, then hands the resulting `str0m::Rtc` to the internal
//! session driver in ingest mode (the same sink the WHIP server uses).

use std::time::Instant;

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

use crate::{Error, Result, client::Client, ingest::IngestSink, session};

pub(crate) async fn dial(client: &Client, url: Url, broadcast: moq_net::broadcast::Producer) -> Result<()> {
	let config = moq_mux::catalog::Config::default()
		.with_max_age(client.config().max_age)
		.with_bandwidth(client.config().bandwidth.clone());
	let sink = Box::new(IngestSink::new(broadcast, config)?);

	let (socket, candidates) = session::bind_udp(&client.config().ice_candidates).await?;
	let mut rtc = Rtc::new(Instant::now());
	for addr in &candidates {
		let cand = Candidate::host(*addr, "udp").map_err(Error::rtc)?;
		rtc.add_local_candidate(cand);
	}

	// Ask for both audio and video, recvonly. The remote answer can decline
	// either by signaling inactive on that m-line.
	let mut api = rtc.sdp_api();
	let mids = [
		api.add_media(MediaKind::Audio, Direction::RecvOnly, None, None, None),
		api.add_media(MediaKind::Video, Direction::RecvOnly, 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, "whep 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::ingest(rtc, socket, candidates, inbound, sink);
	for mid in mids {
		session.open_media(mid)?;
	}
	tokio::spawn(async move {
		let result = session.run().await;
		session::log_session_end("whep client", &result);
	});

	Ok(())
}