moq-cli 0.8.4

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

//! HLS / LL-HLS endpoints: pull a remote playlist into MoQ (import), or serve
//! HLS over HTTP from MoQ broadcasts (export).

use std::net::SocketAddr;
use std::time::Duration;

use anyhow::Context;
use axum::http::Method;
use hang::moq_net;
use hang::moq_net::AsPath;

use crate::moq::notify_ready;

/// HLS import (pull a remote playlist) args.
#[derive(clap::Args, Clone)]
pub struct ImportArgs {
	/// Playlist URL (http/https) or local file path.
	pub playlist: String,
}

/// HLS export (serve over HTTP) args.
#[derive(clap::Args, Clone)]
pub struct ExportArgs {
	/// HTTP listener for the HLS endpoints.
	#[arg(long, default_value = "[::]:8089")]
	pub listen: SocketAddr,

	/// TLS certificates, keys, self-signed generation, and optional mTLS roots.
	#[command(flatten)]
	pub tls: moq_native::tls::Server,

	/// LL-HLS part target duration.
	#[arg(long, default_value = "500ms", value_parser = humantime::parse_duration)]
	pub part_target: Duration,

	/// Minimum media kept in each rendition's sliding window.
	#[arg(long, default_value = "16s", value_parser = humantime::parse_duration)]
	pub window: Duration,

	/// Maximum latency before skipping a stalled group. Generous by default so
	/// live GOPs aren't skipped while segments build.
	#[arg(long = "latency-max", default_value = "10s", value_parser = humantime::parse_duration)]
	pub latency_max: Duration,

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

/// Pull a remote HLS/LL-HLS playlist (URL or file path) into the Origin under `name`.
pub async fn import(origin: &moq_net::OriginProducer, name: String, playlist: String) -> anyhow::Result<()> {
	let mut producer = moq_net::Broadcast::new().produce();

	// Create the catalog tracks BEFORE announcing the broadcast: a subscriber that
	// sees the announce must be able to subscribe `catalog.json` immediately. Announcing
	// first left a window where a consumer got `NotFound` (the export side used to paper
	// over it with a retry loop).
	let catalog = moq_mux::catalog::Producer::new(&mut producer)?;
	anyhow::ensure!(
		origin.publish_broadcast(&name, producer.consume()),
		"failed to publish broadcast"
	);

	let mut importer = moq_hls::import::Import::new(producer, catalog, moq_hls::import::Config::new(playlist))?;

	tracing::info!(%name, "importing HLS");

	importer.init().await?;
	notify_ready();
	Ok(importer.run().await?)
}

/// Serve HLS/LL-HLS over HTTP for the single broadcast `name` (reached at
/// `/<name>/master.m3u8`); other broadcasts in the Origin are not served.
pub async fn export(origin: moq_net::OriginConsumer, args: ExportArgs, name: String) -> anyhow::Result<()> {
	let scoped = origin
		.scope(&[name.as_path()])
		.with_context(|| format!("failed to scope origin to broadcast `{name}`"))?;

	let config = moq_hls::export::Config {
		part_target: args.part_target,
		window: args.window,
		latency: args.latency_max,
		..Default::default()
	};
	let server = moq_hls::Server::new(scoped, config);
	let app = server.router().layer(args.cors.layer([Method::GET])?);

	let tls = if args.tls.cert.is_empty() && args.tls.generate.is_empty() {
		None
	} else {
		let alpn = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
		Some(args.tls.server_config(alpn)?)
	};

	let listener = moq_native::bind::tcp(args.listen)?;

	tracing::info!(listen = %args.listen, "serving HLS");
	notify_ready();

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