hang 0.20.2

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
91
92
93
94
95

// cargo run --example subscribe

use std::time::Duration;

use anyhow::Context;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
	// Optional: Use moq_native to configure a logger.
	moq_native::Log::new(tracing::Level::DEBUG).init()?;

	// Create an origin that the session can publish incoming broadcasts to.
	let origin = moq_net::Origin::random().produce();
	let consumer = origin.consume();

	// Run the subscription and the session in parallel.
	tokio::select! {
		res = run_session(origin) => res,
		res = run_subscribe(consumer) => res,
	}
}

// Connect to the server and subscribe to broadcasts.
// Automatically reconnects if the connection drops.
async fn run_session(origin: moq_net::origin::Producer) -> anyhow::Result<()> {
	// Optional: Use moq_native to make a QUIC client.
	let client = moq_native::ClientConfig::default().init()?;

	// For local development, use: http://localhost:4443/video-example
	// The "anon" path is usually configured to bypass authentication; be careful!
	let url = url::Url::parse("https://cdn.moq.dev/anon/video-example").unwrap();

	// Establish a connection with automatic reconnection.
	// with_subscriber() registers an OriginProducer for incoming data.
	// Use with_publisher() if you also want to publish from the session.
	let reconnect = client.with_subscriber(origin).reconnect(url);

	// Wait until the reconnect loop stops (e.g. timeout exceeded).
	Ok(reconnect.closed().await?)
}

// Subscribe to a broadcast and read media frames.
async fn run_subscribe(consumer: moq_net::origin::Consumer) -> anyhow::Result<()> {
	// Wait for a broadcast to be announced.
	let moq_net::announce::Update { path, broadcast } = consumer.announced().next().await.context("origin closed")?;

	let broadcast = broadcast.with_context(|| format!("broadcast unannounced: {path}"))?;

	tracing::info!(%path, "broadcast announced");

	// Read the catalog to discover available tracks.
	let catalog_track = broadcast
		.track(hang::Catalog::DEFAULT_NAME)?
		.subscribe(hang::Catalog::default_subscription())
		.await?;
	let mut catalog = moq_mux::catalog::hang::Consumer::<()>::new(catalog_track);

	let info = catalog.next().await?.ok_or_else(|| anyhow::anyhow!("no catalog"))?;

	// Find the first video track.
	let (name, config) = info
		.video
		.renditions
		.iter()
		.next()
		.ok_or_else(|| anyhow::anyhow!("no video renditions"))?;

	tracing::info!(
		%name,
		codec = %config.codec,
		width = ?config.coded_width,
		height = ?config.coded_height,
		"subscribing to video track"
	);

	// Subscribe to the video track.
	let track_consumer = broadcast
		.track(name)?
		.subscribe(moq_net::track::Subscription::default().with_priority(1))
		.await?;
	let mut ordered = moq_mux::container::Consumer::new(track_consumer, moq_mux::catalog::hang::Container::Legacy)
		.with_latency(Duration::from_millis(500));

	// Read frames in latency-bounded presentation order.
	while let Some(frame) = ordered.read().await? {
		tracing::info!(
			timestamp = ?frame.timestamp,
			keyframe = frame.keyframe,
			bytes = frame.payload.len(),
			"received frame"
		);
	}

	Ok(())
}