moq-net 0.3.18

The networking layer for Media over QUIC: real-time pub/sub with built-in caching, fan-out, and prioritization.
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
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

//! Control counters cross real protocol drivers, including legacy multiplexing.
//!
//! Time is paused, so the waits below cost nothing and never touch the wall clock.

mod support;

use std::time::Duration;

use moq_net::{
	Hop, Timestamp, Version, origin,
	stats::{Control, Registry, Tier},
};
use support::harness::{MockConnectOptions, connect_mock, now, run};
use support::mock::create_mock_session_pair;

const TIMEOUT: Duration = Duration::from_secs(10);

fn origin(hop: u64) -> origin::Producer {
	let (origin, driver) = origin::Producer::new(origin::Config::new(Hop::new(hop).unwrap()));
	tokio::spawn(run(driver));
	origin
}

#[tokio::test(start_paused = true)]
async fn control_classes_cross_lite_and_ietf_sessions_without_counting_payload() {
	for version in [
		"moq-lite-03",
		"moq-lite-06",
		"moq-lite-07-wip",
		"moq-transport-14",
		"moq-transport-15",
		"moq-transport-16",
		"moq-transport-22",
	] {
		let publisher = origin(1);
		let subscriber = origin(2);
		let registry = Registry::new(Default::default());
		let tier = Tier::new("internal");
		let mut options = MockConnectOptions::new(version.parse::<Version>().unwrap());
		options.server_stats = registry.tier(tier.clone()).session("");
		options.client_publish = Some(publisher.consume());
		options.server_subscribe = Some(subscriber.clone());
		let _pair = connect_mock(options).await;
		let broadcast = publisher.create_broadcast("room").unwrap();
		let track = broadcast.create_track("video", None).unwrap();
		broadcast.announce(Default::default()).unwrap();
		let consumer = subscriber.consume();
		tokio::time::timeout(TIMEOUT, consumer.routed("room"))
			.await
			.expect("announce timeout")
			.expect("routed");
		let remote = tokio::time::timeout(TIMEOUT, consumer.request_broadcast("room"))
			.await
			.expect("resolve timeout")
			.expect("broadcast resolves");
		let reader = tokio::spawn(async move {
			let mut sub = remote.track("video").unwrap().subscribe(None).await.unwrap();
			let mut group = sub.recv_group().await.unwrap().unwrap();
			group.read_frame().await.unwrap().unwrap().payload.len()
		});
		tokio::time::timeout(TIMEOUT, track.used())
			.await
			.expect("no subscriber appeared")
			.unwrap();
		let mut group = track.append_group().unwrap();
		group
			.write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![7; 64 * 1024]))
			.unwrap();
		group.finish().unwrap();
		let read = tokio::time::timeout(TIMEOUT, reader)
			.await
			.expect("group timeout")
			.unwrap();
		assert_eq!(read, 64 * 1024, "{version}");
		let rows = registry.snapshot().control();
		for class in [Control::Announce, Control::Subscribe] {
			assert!(
				rows.iter()
					.any(|(row_tier, row_class, bytes)| *row_tier == tier && *row_class == class && bytes.tx > 0),
				"{version}: {class:?}: {rows:?}"
			);
		}
		let rx: u64 = rows.iter().map(|(_, _, bytes)| bytes.rx).sum();
		assert!(
			rx > 0 && rx < 64 * 1024,
			"{version}: media payload entered control counters: {rows:?}"
		);
	}
}

/// The SETUP a server reads before admission counts under the stats the handshake is
/// admitted with, even for a session with neither origin.
#[tokio::test(start_paused = true)]
async fn setup_counts_under_the_admitted_stats() {
	for version in ["moq-lite-07-wip", "moq-transport-14", "moq-transport-22"] {
		let version: Version = version.parse().unwrap();
		let registry = Registry::new(Default::default());
		let tier = Tier::new("internal");
		let (client_transport, server_transport) = create_mock_session_pair(Some(version.alpn()));
		let client = moq_net::Client::new().with_versions(version.into());
		let server = moq_net::Server::new().with_versions(version.into());
		let stats = registry.tier(tier.clone()).session("");
		let client = async {
			let (session, driver) = client.connect(now(), client_transport).await.unwrap();
			tokio::spawn(run(driver));
			session
		};
		let server = async {
			let request = server.accept_request(now(), server_transport).await.unwrap();
			let (session, driver) = request.with_stats(stats).ok().await.unwrap();
			tokio::spawn(run(driver));
			session
		};
		let _sessions = tokio::join!(client, server);
		let counted = || {
			registry
				.snapshot()
				.control()
				.into_iter()
				.any(|(row_tier, class, bytes)| {
					row_tier == tier && class == Control::Session && bytes.tx > 0 && bytes.rx > 0
				})
		};
		tokio::time::timeout(TIMEOUT, async {
			while !counted() {
				tokio::time::sleep(Duration::from_millis(1)).await;
			}
		})
		.await
		.unwrap_or_else(|_| panic!("{version}: {:?}", registry.snapshot().control()));
	}
}