moq-cli 0.12.0

Media over QUIC
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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147

//! WebRTC (WHIP/WHEP) endpoints. Direction picks the HTTP role:
//! - import `--listen` = WHIP server (accept publishes); `--connect` = WHEP client (pull).
//! - export `--listen` = WHEP server (serve plays); `--connect` = WHIP client (push).

use std::net::SocketAddr;

use anyhow::Context;
use axum::http::Method;
use hang::moq_net;
use moq_tokio::RedactedUrl;
use url::Url;

use crate::moq::{ImportTarget, notify_ready};

/// WebRTC endpoint args: exactly one of `--connect` (WHIP/WHEP client) /
/// `--listen` (WHIP/WHEP server). The parent direction picks WHIP vs WHEP.
#[derive(usage::Args, Clone)]
#[usage(unknown_flags = "error", args_override_self = false)]
#[usage(group("rtc-mode", required))]
pub struct Args {
	/// Dial a remote WHIP/WHEP endpoint URL.
	#[usage(name = "rtc-connect", long = "connect", value_name = "URL", group = "rtc-mode")]
	pub connect: Option<Url>,

	/// Bind an HTTP listener for WHIP/WHEP, scoped to the single `--broadcast`
	/// (peers reach it at `http://host:port/<broadcast>`).
	#[usage(name = "rtc-listen", long = "listen", value_name = "ADDR", group = "rtc-mode")]
	pub listen: Option<SocketAddr>,

	/// Shared UDP socket for ICE/media (one port for all sessions).
	#[usage(long, requires = "--listen", default = "[::]:0")]
	pub udp_bind: SocketAddr,

	/// Public UDP address(es) advertised as ICE host candidates (repeatable).
	#[usage(long, requires = "rtc-listen")]
	pub public_addr: Vec<SocketAddr>,

	/// Browser CORS policy for the WHIP/WHEP listener.
	#[usage(flatten)]
	pub cors: crate::web::Cors,
}

/// HTTP and ICE settings shared by the WHIP and WHEP listeners.
pub struct Listen {
	/// HTTP address to bind.
	pub addr: SocketAddr,

	/// Shared UDP socket for ICE and media.
	pub udp_bind: SocketAddr,

	/// Public UDP addresses advertised as ICE host candidates.
	pub public_addr: Vec<SocketAddr>,

	/// Browser CORS policy for the HTTP listener.
	pub cors: crate::web::Cors,
}

/// WHIP server: accept incoming WebRTC publishes into the Origin as `target.name` (import).
pub async fn listen_import(target: ImportTarget, listen: Listen) -> anyhow::Result<()> {
	let publisher = scope_producer(&target.origin, &target.name)?;
	let mut config = server_config(&listen);
	config.max_age = target.max_age;
	config.bandwidth = target.bandwidth;
	let server = moq_rtc::Server::new(config);
	serve(server.publish_router(publisher), "WHIP", listen).await
}

/// WHEP server: serve WebRTC plays of `name` from the Origin (export).
pub async fn listen_export(origin: moq_net::origin::Consumer, name: String, listen: Listen) -> anyhow::Result<()> {
	let scope = moq_net::Patterns::from(
		moq_net::Pattern::subtree(&name).with_context(|| format!("invalid broadcast name `{name}`"))?,
	);
	let subscriber = origin
		.scope("", &scope)
		.with_context(|| format!("failed to scope origin to broadcast `{name}`"))?;
	let server = moq_rtc::Server::new(server_config(&listen));
	serve(server.subscribe_router(subscriber), "WHEP", listen).await
}

/// Restrict a producer to the single broadcast `name` so a WHIP peer can only publish it.
fn scope_producer(origin: &moq_net::origin::Producer, name: &str) -> anyhow::Result<moq_net::origin::Producer> {
	let scope = moq_net::Patterns::from(
		moq_net::Pattern::subtree(name).with_context(|| format!("invalid broadcast name `{name}`"))?,
	);
	origin
		.scope("", &scope)
		.with_context(|| format!("failed to scope origin to broadcast `{name}`"))
}

/// WHEP client: pull a remote broadcast into the Origin under `target.name` (import).
pub async fn connect_import(target: ImportTarget, url: Url) -> anyhow::Result<()> {
	let name = &target.name;
	let producer = target
		.origin
		.create_broadcast(name)
		.context("failed to create broadcast")?;
	// The WHEP pull fills the tracks as they arrive; announce up front so viewers
	// can discover the broadcast while it connects.
	producer
		.announce(Default::default())
		.context("failed to announce broadcast")?;

	tracing::info!(url = %RedactedUrl::new(&url), %name, "WHEP client pulling");
	notify_ready();

	let mut config = moq_rtc::client::Config::default();
	config.max_age = target.max_age;
	config.bandwidth = target.bandwidth;
	let client = moq_rtc::Client::new(config);
	Ok(client.subscribe(url, producer).await?)
}

/// WHIP client: push a broadcast from the Origin to a remote (export).
pub async fn connect_export(origin: moq_net::origin::Consumer, url: Url, name: String) -> anyhow::Result<()> {
	// Confirm the broadcast is reachable (and wait for it to be announced) before dialing;
	// the egress re-resolves it (and any referenced sibling broadcast) through the origin.
	origin
		.routed(&name)
		.await
		.with_context(|| format!("origin closed before broadcast `{name}` was announced"))?;

	tracing::info!(url = %RedactedUrl::new(&url), %name, "WHIP client pushing");
	notify_ready();

	let client = moq_rtc::Client::new(moq_rtc::client::Config::default());
	Ok(client.publish(url, origin, &name).await?)
}

fn server_config(listen: &Listen) -> moq_rtc::server::Config {
	let mut config = moq_rtc::server::Config::default();
	config.udp_bind = listen.udp_bind;
	config.ice_candidates.clone_from(&listen.public_addr);
	config
}

async fn serve(router: axum::Router, role: &str, listen: Listen) -> anyhow::Result<()> {
	let cors = listen
		.cors
		.layer([Method::POST, Method::PATCH, Method::DELETE, Method::OPTIONS])?;
	let app = router.layer(cors);
	let listener = moq_tokio::bind::tcp(listen.addr)?;

	tracing::info!(listen = %listen.addr, role, "serving WebRTC");
	notify_ready();

	crate::web::serve(listener, app, None).await
}