//! The local stack's ports, held in one place. Every default port of every program, and every //! port one program's example configuration names to reach another, agrees with the program that //! serves it, and with the port table of wai/quickstart/README.md: //! //! - wai-relay: relay.example.toml's endpoints are `wai-relay --dev`'s (`RelayConfig::dev`); //! - wai-room: room.example.toml's `relay_url` is the relay example's WebSocket endpoint and path; //! - wai-ingest: gateway.example.toml, read by the gateway's own reader (`GatewayConfig`): its //! `[relay] addr` is the relay example's QUIC endpoint, and each listener it names is a row, //! those it leaves commented out (off until set) said to be, each of those read as the gateway //! reads it once its line is uncommented; //! - `wai dev`'s origin, which the remote desktop template's host.sh reaches by default; //! - every other daemon whose default is in its source: the audit collector, the ledger, the //! fleet service, the quantum job service, the worlds' relay, the stitched presentation //! verifier and `wai http origin`; //! //! no two defaults are one port; the room service started from its example file opens its //! clock's session on the relay started from the relay's, each moved to a free port alike; and //! the gateway started from its example, its ports moved to free ones and its files made, opens //! the listeners the table says are on and none of those it says are off. use joule_moq_ingest::{GatewayConfig, GatewayServer}; use joule_moq_relay::{RelayConfig, RelayServer}; use joule_moq_room::keys::Keys; use joule_moq_room::{Config, Service}; use std::collections::BTreeMap; use std::net::SocketAddr; use std::path::{Path, PathBuf}; use std::sync::atomic::Ordering; use std::time::Duration; fn repo() -> PathBuf { Path::new(env!("CARGO_MANIFEST_DIR")).join("../../../..") } fn read(rel: &str) -> String { std::fs::read_to_string(repo().join(rel)).unwrap_or_else(|e| panic!("{rel}: {e}")) } const ROOM_EXAMPLE: &str = "joule-code/joule-code-rs/crates/joule-moq-room/room.example.toml"; const RELAY_EXAMPLE: &str = "joule-code/joule-code-rs/crates/joule-moq-relay/relay.example.toml"; const GATEWAY_EXAMPLE: &str = "joule-code/joule-code-rs/crates/joule-moq-ingest/gateway.example.toml"; const TABLE: &str = "wai/quickstart/README.md"; /// `ws://host:port/path` as (scheme, authority, path). fn split_url(u: &str) -> (&str, &str, &str) { let (scheme, rest) = u.split_once("://").unwrap_or_else(|| panic!("{u}: a URL")); let (authority, path) = rest.find('/').map_or((rest, ""), |i| (&rest[..i], &rest[i..])); (scheme, authority, path) } /// A row of the port table: its program, what listens there, and the setting that moves it. struct Row { program: String, what: String, setting: String, } /// What a row says of a listener that is off until its setting is given. const OFF: &str = "off until set"; /// The port table: each row's port (with `/udp` for a UDP one) and the row. fn table() -> BTreeMap { let text = read(TABLE); let heading = "\n## Ports\n"; let at = text.find(heading).expect("the README has a port table") + heading.len(); let mut rows = BTreeMap::new(); for line in text[at..].lines().take_while(|l| !l.starts_with("## ")) { let cells: Vec<&str> = line.split('|').map(str::trim).collect(); if cells.len() < 6 || !cells[1].starts_with(|c: char| c.is_ascii_digit()) { continue; } let unquote = |s: &str| s.trim_matches('`').to_string(); let row = Row { program: unquote(cells[2]), what: cells[3].to_string(), setting: unquote(cells[4]) }; assert!(rows.insert(cells[1].to_string(), row).is_none(), "{}: twice in the table", cells[1]); } rows } /// `text` with the line that gives `key` in its table `[section]` uncommented, when the example /// leaves it commented out (a listener the program opens only once it is set); `None` when the /// line is not commented out. fn uncommented(text: &str, section: &str, key: &str) -> Option { let header = format!("[{section}]"); let lines: Vec<&str> = text.lines().collect(); let at = lines.iter().position(|l| l.trim() == header).unwrap_or_else(|| panic!("no {header}")); for (i, line) in lines.iter().enumerate().skip(at + 1).take_while(|(_, l)| !l.trim_start().starts_with('[')) { let (off, body) = match line.trim_start().strip_prefix('#') { Some(rest) => (true, rest.trim_start()), None => (false, line.trim_start()), }; if body.strip_prefix(key).is_some_and(|r| r.trim_start().starts_with('=')) { if !off { return None; } let mut out: Vec = lines.iter().map(|l| l.to_string()).collect(); out[i] = body.to_string(); return Some(out.join("\n")); } } panic!("{header} names no {key}") } /// The gateway's listeners as its reader reads them, each with its section, its key, whether it /// is UDP, and its address in a configuration. type Listener = (&'static str, &'static str, bool, fn(&GatewayConfig) -> Option); const GATEWAY_LISTENERS: &[Listener] = &[ ("http", "listen", false, |c| Some(c.http.listen)), ("rtmp", "listen", false, |c| c.rtmp.listen), ("rtmp", "tls_listen", false, |c| c.rtmp.tls_listen), ("injector", "listen", false, |c| c.injector.listen), ("contrib.srt", "listen", true, |c| c.contrib.srt.listen), ]; /// The numbers of four or five digits on a line of source: the ports it names. fn ports_on(line: &str) -> Vec { line.split(|c: char| !c.is_ascii_digit()).filter(|n| (4..=5).contains(&n.len())).filter_map(|n| n.parse().ok()).collect() } #[test] fn every_default_port_agrees_with_the_program_that_serves_it() { let t = table(); let mut held = Vec::new(); let mut row = |port: String, program: &str, setting: &str, off: bool| { let r = t.get(&port).unwrap_or_else(|| panic!("the table has no row for {port} ({program})")); assert_eq!((r.program.as_str(), r.setting.as_str()), (program, setting), "the table's row for {port}"); assert_eq!(r.what.contains(OFF), off, "the table's row for {port} says {OFF:?} exactly when the listener is off until set"); held.push(port); }; // The relay: its example's endpoints are its development defaults, and the table's. let relay = RelayConfig::from_toml(&read(RELAY_EXAMPLE)).expect("the relay's example reads"); let dev = RelayConfig::dev(); assert_eq!((relay.quic_listen, relay.websocket_listen, relay.path.as_str()), (dev.quic_listen, dev.websocket_listen, dev.path.as_str()), "{RELAY_EXAMPLE} is `wai-relay --dev`"); let quic = relay.quic_listen.expect("a QUIC endpoint"); let ws = relay.websocket_listen.expect("a WebSocket endpoint"); row(format!("{}/udp", quic.port()), "wai-relay", "quic_listen", false); row(ws.port().to_string(), "wai-relay", "websocket_listen", false); // The room service: its relay is the relay's WebSocket endpoint, at the relay's path. let room = Config::from_toml(&read(ROOM_EXAMPLE)).expect("the room's example reads"); let (scheme, authority, path) = split_url(&room.relay_url); let at: SocketAddr = authority.parse().unwrap_or_else(|e| panic!("{authority}: {e}")); assert_eq!((scheme, at, path), ("ws", ws, relay.path.as_str()), "{ROOM_EXAMPLE}'s relay_url is the relay's WebSocket endpoint"); row(room.listen.port().to_string(), "wai-room", "listen", false); // The ingest gateway, its example read by its own reader: its relay is the relay's QUIC // endpoint, and each listener its example names is a row. One the example leaves commented // out is off (the reader reads none) and its row says so; uncommented, the reader reads it at // the row's port. let text = read(GATEWAY_EXAMPLE); let gw = GatewayConfig::from_toml(&text).unwrap_or_else(|e| panic!("{GATEWAY_EXAMPLE} reads as the gateway reads it: {e}")); assert_eq!(gw.relay.addr, quic, "{GATEWAY_EXAMPLE}'s [relay] addr is the relay's QUIC endpoint"); for (section, key, udp, at) in GATEWAY_LISTENERS { let (a, off) = match uncommented(&text, section, key) { None => (at(&gw).unwrap_or_else(|| panic!("{GATEWAY_EXAMPLE}: [{section}] {key} is set, and the gateway reads none")), false), Some(on) => { assert_eq!(at(&gw), None, "{GATEWAY_EXAMPLE}: [{section}] {key} is commented out, and the gateway reads one"); let c = GatewayConfig::from_toml(&on).unwrap_or_else(|e| panic!("{GATEWAY_EXAMPLE} with [{section}] {key} set: {e}")); (at(&c).unwrap_or_else(|| panic!("{GATEWAY_EXAMPLE}: [{section}] {key}, uncommented, is not what the gateway reads")), true) } }; let port = format!("{}{}", a.port(), if *udp { "/udp" } else { "" }); row(port, "wai-ingest", &format!("[{section}] {key}"), off); } // `wai dev`'s origin: its default, and the one the remote desktop template's host reaches. let dev_rs = read("wai/wai-rs/src/bin/wai_cli/dx/dev.rs"); let origin = t.iter().find(|(_, r)| r.program == "wai dev").map(|(port, _)| port.clone()).expect("a row for wai dev"); assert!(dev_rs.contains(&format!("None => {origin},")), "wai dev's default port is {origin}"); row(origin.clone(), "wai dev", "--port", false); assert!(read("wai/examples/remote-desktop/host.sh").contains(&format!("WAI_DEV_ORIGIN:-http://127.0.0.1:{origin}}}")), "host.sh reaches wai dev's origin"); // The daemons whose defaults are in their sources: each source, the text on the line that // reads the setting, the program and the setting. That line names one port, the table's. let sources = [ ("wai/wai-audit/src/main.rs", "env_or(\"WAI_AUDIT_ADDR\"", "wai-audit", "WAI_AUDIT_ADDR"), ("wai/wai-ledger/src/main.rs", "var(\"WAI_LEDGER_ADDR\")", "wai-ledger", "WAI_LEDGER_ADDR"), ("wai/wai-fleet/src/main.rs", "var(\"WAI_FLEET_ADDR\")", "wai-fleet", "WAI_FLEET_ADDR"), ("wai/wai-quantum-service/src/main.rs", "var(\"WAI_QSVC_ADDR\")", "wai-quantum-service", "WAI_QSVC_ADDR"), ("wai/wai-relay/src/main.rs", "var(\"WAI_RELAY_BIND\")", "wai-world-relay", "WAI_RELAY_BIND"), ("wai/wai-rs/src/bin/wai_stitched_presentation.rs", "opt(a, \"--listen\")", "wai_stitched_presentation serve", "--listen"), ("wai/wai-rs/src/bin/wai_cli/http.rs", "opt(args, \"--port\").map_or(", "wai http origin", "--port"), ]; for (src, reads, program, setting) in sources { let text = read(src); let lines: Vec<&str> = text.lines().filter(|l| l.contains(reads) && !l.trim_start().starts_with("//")).collect(); assert_eq!(lines.len(), 1, "{src}: one line reads {setting}: {lines:?}"); let ports = ports_on(lines[0]); assert_eq!(ports.len(), 1, "{src}: the line that reads {setting} names one port: {}", lines[0]); row(ports[0].to_string(), program, setting, false); } // Every row is one held above; the table holds each port once, so no two programs' defaults // are one port on one protocol. held.sort(); assert_eq!(held, t.keys().cloned().collect::>(), "the rows held and the table's"); } /// The room service, started from room.example.toml, opens its clock's session on the relay /// started from relay.example.toml. Both move to free ports alike (the room's `relay_url` keeps /// its scheme, host and path, its port the relay's), and the relay keeps nothing on disk. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn the_room_example_reaches_the_relay_example() { let mut rc = RelayConfig::from_toml(&read(RELAY_EXAMPLE)).unwrap(); rc.quic_listen = None; rc.websocket_listen = Some("127.0.0.1:0".parse().unwrap()); rc.dvr = None; let relay = RelayServer::start(rc).await.expect("the relay starts from its example"); let ws = relay.websocket_addr.expect("its WebSocket endpoint"); let text = read(ROOM_EXAMPLE); let example = Config::from_toml(&text).unwrap(); let (scheme, authority, path) = split_url(&example.relay_url); let moved = format!("{scheme}://{}:{}{path}", authority.rsplit_once(':').unwrap().0, ws.port()); let text = text .replace(&format!("relay_url = \"{}\"", example.relay_url), &format!("relay_url = \"{moved}\"")) .replace(&format!("listen = \"{}\"", example.listen), "listen = \"127.0.0.1:0\""); let cfg = Config::from_toml(&text).unwrap(); assert_eq!(cfg.relay_url, moved, "the example's relay_url, moved"); let svc = Service::new(Keys::generate(), &cfg); tokio::spawn(svc.clone().run_clock()); let up = tokio::time::timeout(Duration::from_secs(30), async { while !svc.connected.load(Ordering::Acquire) { tokio::time::sleep(Duration::from_millis(50)).await; } }) .await; assert!(up.is_ok(), "the clock's session opens on the relay at {moved}"); } /// The gateway, started from gateway.example.toml with its ports moved to free ones and the /// files it names made (its signing key, the relay's certificate and token, a room's signed /// descriptor): it serves HTTP, opens RTMP, which the example sets, and none of the listeners /// the example leaves off (RTMPS, the SCTE 104 injector, SRT). #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn the_gateway_starts_from_its_example() { use tokio::io::{AsyncReadExt, AsyncWriteExt}; let dir = std::env::temp_dir().join(format!("ports-gateway-{}", std::process::id())); std::fs::create_dir_all(&dir).unwrap(); let keys = Keys::generate(); let now_us = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_micros() as u64; let room = joule_moq_room::epoch::descriptor(&keys, joule_moq_room::epoch::roles(false, 0, false, false), 8, 4, now_us); let mut text = read(GATEWAY_EXAMPLE); let example = GatewayConfig::from_toml(&text).unwrap(); for (name, bytes) in [("gateway.key", vec![7u8; 32]), ("relay-cert.pem", Vec::new()), ("relay-token", b"token".to_vec()), ("room.wrd1", room.to_bytes())] { std::fs::write(dir.join(name), bytes).unwrap(); let from = format!("\"{name}\""); assert_eq!(text.matches(&from).count(), 1, "{GATEWAY_EXAMPLE} names {name} once"); text = text.replace(&from, &format!("{:?}", dir.join(name).to_str().unwrap())); } for at in [Some(example.http.listen), example.rtmp.listen] { let from = format!("listen = \"{}\"", at.unwrap()); assert_eq!(text.matches(&from).count(), 1, "{GATEWAY_EXAMPLE}: {from} once"); text = text.replace(&from, "listen = \"127.0.0.1:0\""); } let cfg = GatewayConfig::from_toml(&text).unwrap(); let gw = GatewayServer::start(cfg).await.unwrap_or_else(|e| panic!("the gateway starts from its example: {e}")); assert!(gw.addr.ip().is_loopback() && gw.rtmp_addr.is_some_and(|a| a.ip().is_loopback()), "HTTP and RTMP, on loopback"); assert_eq!((gw.rtmps_addr, gw.injector_addr, gw.srt_addr), (None, None, None), "RTMPS, the injector and SRT, off until set"); assert!(gw.gateway.rooms.contains_key(&room.room_id()), "the room its example names"); let mut s = tokio::net::TcpStream::connect(gw.addr).await.unwrap(); s.write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n").await.unwrap(); let mut head = [0u8; 9]; s.read_exact(&mut head).await.unwrap(); assert_eq!(&head, b"HTTP/1.1 ", "the gateway answers HTTP at its listener"); let _ = std::fs::remove_dir_all(&dir); }