71fbfe5a85
- auth.authorities: a WRG1 grant signed by a listed authority is honoured as live-rooms §10.2 maps it (its own media, catalog and events namespaces to publish, the room to subscribe), only beside a proof of possession and within its window; joule-moq's GrantHead reads it under the authority's key. - A field "*" in a grant rule's or a door rule's prefix matches any field; a door rule's trust_field names the namespace field whose key it trusts there. - limits.max_session_ms, max_ingress_kbps (with ingress_burst_ms) and max_resident_mb; a session also ends when its authority grant expires. - The binary stops on SIGTERM as on Ctrl-C. Co-Authored-By: Claude Opus 5.5 <[email protected]>
699 lines
30 KiB
Rust
699 lines
30 KiB
Rust
//! What the gateway's end-to-end tests share: a relay and a gateway on loopback, a room
|
|
//! and its grants, a WebRTC client on str0m (the WHIP publisher and the WHEP player), an
|
|
//! HTTP client, synthetic frames, and a relay subscriber that verifies every receipt.
|
|
//!
|
|
//! Every address is loopback and every wait is bounded.
|
|
|
|
#![allow(dead_code)]
|
|
|
|
use base64::Engine;
|
|
use bytes::Bytes;
|
|
use ed25519_dalek::SigningKey;
|
|
use http_body_util::{BodyExt, Full};
|
|
use joule_moq::carriage::ReceiptBundle;
|
|
use joule_moq::draft22::data::FetchEntry;
|
|
use joule_moq::draft22::{ControlMessage, Parameters, StreamHeader, UniEvent};
|
|
use joule_moq::receipt::ReceiptVerifier;
|
|
use joule_moq::{GroupId, ObjectId, TrackName, TrackNamespace};
|
|
use joule_moq_ingest::catalog;
|
|
use joule_moq_ingest::frame::{self, Codec};
|
|
use joule_moq_ingest::room::{build, Descriptor, Grant};
|
|
use joule_moq_ingest::{sdp, GatewayConfig, GatewayServer};
|
|
use joule_moq_relay::client::{Client, ClientOptions};
|
|
use joule_moq_relay::transport::quic;
|
|
use joule_moq_relay::wire::{ReqRx, ReqTx};
|
|
use joule_moq_relay::{RelayConfig, RelayServer};
|
|
use std::collections::{BTreeMap, HashMap};
|
|
use std::sync::{Arc, Mutex};
|
|
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
|
|
use str0m::change::{SdpAnswer, SdpPendingOffer};
|
|
use str0m::media::{Direction, Frequency, MediaKind, MediaTime, Mid};
|
|
use str0m::net::{Protocol, Receive};
|
|
use str0m::{Candidate, Event, IceConnectionState, Input, Output, Rtc, RtcConfig};
|
|
use tokio::net::UdpSocket;
|
|
use tokio::sync::{mpsc, oneshot};
|
|
use tokio::time::timeout;
|
|
|
|
/// The bound on any one wait.
|
|
pub const STEP: Duration = Duration::from_secs(20);
|
|
|
|
pub fn now_us() -> u64 {
|
|
SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros() as u64
|
|
}
|
|
|
|
// ── The room, the relay and the gateway ────────────────────────────────────────────
|
|
|
|
pub struct World {
|
|
pub relay: RelayServer,
|
|
pub gateway: GatewayServer,
|
|
pub authority: SigningKey,
|
|
pub descriptor: Descriptor,
|
|
pub room: [u8; 32],
|
|
pub publisher: SigningKey,
|
|
pub gateway_key: SigningKey,
|
|
}
|
|
|
|
impl World {
|
|
/// A relay whose door checks every receipt of the room's tracks under the gateway's
|
|
/// key, and a gateway that publishes to it, with `f` adjusting the gateway's
|
|
/// configuration.
|
|
pub async fn start(f: impl FnOnce(&mut GatewayConfig)) -> World {
|
|
let authority = SigningKey::from_bytes(&[0x51; 32]);
|
|
let publisher = SigningKey::from_bytes(&[0x52; 32]);
|
|
let gateway_key = SigningKey::from_bytes(&[0x53; 32]);
|
|
let descriptor = build::room(&authority, now_us() - 60_000_000);
|
|
let room = descriptor.room_id();
|
|
let gk = hex::encode(gateway_key.verifying_key().to_bytes());
|
|
let mut rc = RelayConfig::loopback();
|
|
rc.receipts.door = vec![joule_moq_relay::config::DoorRule { prefix: vec![hex::encode(room)], trust: vec![gk], trust_field: None, hold_ms: 10_000 }];
|
|
let relay = RelayServer::start(rc).await.expect("the relay starts");
|
|
let mut gc = GatewayConfig::loopback(relay.quic_addr.expect("a UDP endpoint"));
|
|
gc.relay.trust_der = vec![relay.identity.chain[0].to_vec()];
|
|
gc.signing.key_seed = Some(gateway_key.to_bytes());
|
|
gc.rooms = vec![joule_moq_ingest::config::RoomConfig { descriptor_path: None, revoked: Vec::new(), descriptor: Some(descriptor.to_bytes()) }];
|
|
f(&mut gc);
|
|
let gateway = GatewayServer::start(gc).await.expect("the gateway starts");
|
|
World { relay, gateway, authority, descriptor, room, publisher, gateway_key }
|
|
}
|
|
|
|
/// A grant for `who` under role `role` (0 publisher, 1 viewer, 2 observer), good now.
|
|
pub fn grant(&self, who: &SigningKey, role: u8) -> Grant {
|
|
let s = now_us() / 1_000_000;
|
|
build::grant(&self.descriptor, &self.authority, &who.verifying_key().to_bytes(), role, s - 60, s + 3600)
|
|
}
|
|
|
|
pub fn token(g: &Grant) -> String {
|
|
base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(g.to_bytes())
|
|
}
|
|
|
|
/// `who`'s grant under `role`, with its key, for requests that prove possession.
|
|
pub fn auth(&self, who: &SigningKey, role: u8) -> Auth {
|
|
Auth::new(self.grant(who, role), who)
|
|
}
|
|
|
|
pub fn base(&self) -> String {
|
|
format!("http://{}", self.gateway.addr)
|
|
}
|
|
|
|
pub fn owner(&self) -> String {
|
|
hex::encode(self.publisher.verifying_key().to_bytes())
|
|
}
|
|
|
|
pub fn whep_url(&self) -> String {
|
|
format!("{}/whep/{}/{}", self.base(), hex::encode(self.room), self.owner())
|
|
}
|
|
|
|
pub async fn stop(self) {
|
|
self.gateway.shutdown().await;
|
|
self.relay.shutdown().await;
|
|
}
|
|
}
|
|
|
|
// ── HTTP ───────────────────────────────────────────────────────────────────────────
|
|
|
|
#[derive(Debug)]
|
|
pub struct Resp {
|
|
pub status: u16,
|
|
pub headers: HashMap<String, String>,
|
|
pub body: Vec<u8>,
|
|
}
|
|
|
|
impl Resp {
|
|
pub fn header(&self, k: &str) -> Option<&str> {
|
|
self.headers.get(&k.to_ascii_lowercase()).map(String::as_str)
|
|
}
|
|
pub fn text(&self) -> String {
|
|
String::from_utf8_lossy(&self.body).into_owned()
|
|
}
|
|
}
|
|
|
|
/// One HTTP/1.1 request on a connection of its own.
|
|
pub async fn http(method: &str, url: &str, headers: &[(&str, &str)], body: &[u8]) -> Resp {
|
|
let rest = url.strip_prefix("http://").expect("an http URL");
|
|
let (host, path) = rest.split_at(rest.find('/').unwrap_or(rest.len()));
|
|
let path = if path.is_empty() { "/" } else { path };
|
|
let stream = timeout(STEP, tokio::net::TcpStream::connect(host)).await.expect("connect in time").expect("connect");
|
|
let (mut sender, conn) = hyper::client::conn::http1::handshake(hyper_util::rt::TokioIo::new(stream)).await.expect("handshake");
|
|
tokio::spawn(conn);
|
|
let mut b = hyper::Request::builder().method(method).uri(path).header("host", host);
|
|
for (k, v) in headers {
|
|
b = b.header(*k, *v);
|
|
}
|
|
let req = b.body(Full::new(Bytes::copy_from_slice(body))).unwrap();
|
|
let resp = timeout(STEP, sender.send_request(req)).await.expect("a response in time").expect("a response");
|
|
let status = resp.status().as_u16();
|
|
let headers = resp.headers().iter().map(|(k, v)| (k.as_str().to_ascii_lowercase(), v.to_str().unwrap_or("").to_string())).collect();
|
|
let body = timeout(STEP, resp.into_body().collect()).await.expect("a body in time").expect("a body").to_bytes().to_vec();
|
|
Resp { status, headers, body }
|
|
}
|
|
|
|
pub fn bearer(token: &str) -> String {
|
|
format!("Bearer {token}")
|
|
}
|
|
|
|
/// A principal's grant and key, and the last challenge the gateway gave it: what a client
|
|
/// presents over HTTP (live-rooms §10.5.3).
|
|
pub struct Auth {
|
|
pub grant: Grant,
|
|
pub key: SigningKey,
|
|
challenge: Mutex<Option<[u8; 32]>>,
|
|
}
|
|
|
|
/// `METHOD //authority/path` of a URL, without its query (live-rooms §10.5.1).
|
|
pub fn context_of(method: &str, url: &str) -> String {
|
|
let rest = url.strip_prefix("http://").expect("an http URL");
|
|
let (host, path) = rest.split_at(rest.find('/').unwrap_or(rest.len()));
|
|
joule_moq::grant_proof::context(method, host, if path.is_empty() { "/" } else { path })
|
|
}
|
|
|
|
impl Auth {
|
|
pub fn new(grant: Grant, key: &SigningKey) -> Auth {
|
|
Auth { grant, key: key.clone(), challenge: Mutex::new(None) }
|
|
}
|
|
|
|
pub fn token(&self) -> String {
|
|
World::token(&self.grant)
|
|
}
|
|
|
|
pub fn challenge(&self) -> Option<[u8; 32]> {
|
|
*self.challenge.lock().unwrap()
|
|
}
|
|
|
|
/// A proof by `key` over `challenge` for `method` at `url`, in base64url.
|
|
pub fn proof(&self, key: &SigningKey, challenge: [u8; 32], method: &str, url: &str) -> String {
|
|
let p = joule_moq::grant_proof::GrantProof::sign(self.grant.digest(), key, challenge, &context_of(method, url)).expect("a proof");
|
|
joule_moq::grant_proof::base64url(&p.to_bytes())
|
|
}
|
|
|
|
fn keep(&self, r: &Resp) {
|
|
if let Some(c) = r.header("wai-room-challenge").and_then(joule_moq::grant_proof::from_base64url).and_then(|b| <[u8; 32]>::try_from(b).ok()) {
|
|
*self.challenge.lock().unwrap() = Some(c);
|
|
}
|
|
}
|
|
|
|
/// The request as given: the grant as the Bearer token and `proof`, if any. The
|
|
/// response's challenge is kept.
|
|
pub async fn send_raw(&self, method: &str, url: &str, headers: &[(&str, &str)], body: &[u8], proof: Option<String>) -> Resp {
|
|
let auth = bearer(&self.token());
|
|
let mut h: Vec<(&str, &str)> = headers.to_vec();
|
|
h.push(("authorization", &auth));
|
|
if let Some(p) = &proof {
|
|
h.push(("wai-room-proof", p));
|
|
}
|
|
let r = http(method, url, &h, body).await;
|
|
self.keep(&r);
|
|
r
|
|
}
|
|
|
|
/// The request with a proof over the last challenge; and, where it is answered `401`
|
|
/// with a fresh challenge, once more over that one, as a client does that has none yet.
|
|
pub async fn send(&self, method: &str, url: &str, headers: &[(&str, &str)], body: &[u8]) -> Resp {
|
|
let proof = self.challenge().map(|c| self.proof(&self.key, c, method, url));
|
|
let r = self.send_raw(method, url, headers, body, proof).await;
|
|
if r.status != 401 || r.header("wai-room-challenge").is_none() || !r.text().contains("pop_") {
|
|
return r;
|
|
}
|
|
let c = self.challenge().expect("a challenge");
|
|
let proof = self.proof(&self.key, c, method, url);
|
|
self.send_raw(method, url, headers, body, Some(proof)).await
|
|
}
|
|
}
|
|
|
|
// ── A WebRTC client on str0m ───────────────────────────────────────────────────────
|
|
|
|
#[derive(Debug, Clone)]
|
|
pub enum PeerEvent {
|
|
Connected,
|
|
Ice(IceConnectionState),
|
|
Media { kind: MediaKind, data: Vec<u8>, rtp: u64 },
|
|
}
|
|
|
|
enum PeerCmd {
|
|
Write { kind: MediaKind, codec: Codec, data: Vec<u8>, ticks: u64 },
|
|
/// Start an ICE restart: the new offer, whose ICE attributes go in the PATCH.
|
|
Restart { reply: oneshot::Sender<String> },
|
|
/// Finish it with the answer fragment the gateway returned.
|
|
Restarted { frag: String, reply: oneshot::Sender<Result<(), String>> },
|
|
Close,
|
|
}
|
|
|
|
/// A WebRTC endpoint driven on its own task, on a loopback UDP socket.
|
|
pub struct Peer {
|
|
cmd: mpsc::UnboundedSender<PeerCmd>,
|
|
pub events: Arc<Mutex<Vec<PeerEvent>>>,
|
|
pub offer: String,
|
|
pub answer: String,
|
|
pub location: String,
|
|
pub etag: String,
|
|
}
|
|
|
|
fn client_config(codecs: &[Codec]) -> RtcConfig {
|
|
let mut c = RtcConfig::new().clear_codecs().set_crypto_provider(Arc::new(str0m::crypto::from_feature_flags()));
|
|
for codec in codecs {
|
|
c = match codec {
|
|
Codec::Avc => c.enable_h264(true),
|
|
Codec::Vp8 => c.enable_vp8(true),
|
|
Codec::Vp9 => c.enable_vp9(true),
|
|
Codec::Av1 => c.enable_av1(true),
|
|
Codec::Opus => c.enable_opus(true, false),
|
|
};
|
|
}
|
|
c
|
|
}
|
|
|
|
/// An offer for one audio and one video section in `dir`, on a new endpoint; the
|
|
/// endpoint's run loop starts once its answer is applied.
|
|
pub struct Offering {
|
|
rtc: Rtc,
|
|
socket: UdpSocket,
|
|
pending: SdpPendingOffer,
|
|
/// The sections this end added: str0m raises no MediaAdded for its own.
|
|
mids: HashMap<MediaKind, Mid>,
|
|
pub offer: String,
|
|
}
|
|
|
|
pub async fn offer(codecs: &[Codec], dir: Direction, audio: bool, video: bool) -> Offering {
|
|
let socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
|
|
let local = socket.local_addr().unwrap();
|
|
let mut rtc = client_config(codecs).build(Instant::now());
|
|
rtc.add_local_candidate(Candidate::host(local, "udp").unwrap());
|
|
let mut api = rtc.sdp_api();
|
|
let mut mids = HashMap::new();
|
|
if audio {
|
|
mids.insert(MediaKind::Audio, api.add_media(MediaKind::Audio, dir, Some("stream".into()), Some("audio".into()), None));
|
|
}
|
|
if video {
|
|
mids.insert(MediaKind::Video, api.add_media(MediaKind::Video, dir, Some("stream".into()), Some("video".into()), None));
|
|
}
|
|
let (o, pending) = api.apply().expect("an offer");
|
|
Offering { rtc, socket, pending, mids, offer: o.to_sdp_string() }
|
|
}
|
|
|
|
impl Offering {
|
|
/// Apply the gateway's answer and start the endpoint.
|
|
pub fn start(mut self, resp: &Resp) -> Peer {
|
|
let answer = resp.text();
|
|
let a = SdpAnswer::from_sdp_string(&answer).expect("the answer parses");
|
|
self.rtc.sdp_api().accept_answer(self.pending, a).expect("the answer applies");
|
|
let (tx, rx) = mpsc::unbounded_channel();
|
|
let events = Arc::new(Mutex::new(Vec::new()));
|
|
tokio::spawn(run_peer(self.rtc, self.socket, rx, events.clone(), answer.clone(), self.mids));
|
|
let location = resp.header("location").expect("a Location").to_string();
|
|
let etag = resp.header("etag").expect("an ETag").trim_matches('"').to_string();
|
|
Peer { cmd: tx, events, offer: self.offer, answer, location, etag }
|
|
}
|
|
}
|
|
|
|
async fn run_peer(mut rtc: Rtc, socket: UdpSocket, mut rx: mpsc::UnboundedReceiver<PeerCmd>, events: Arc<Mutex<Vec<PeerEvent>>>, mut answer: String, mut mids: HashMap<MediaKind, Mid>) {
|
|
let local = socket.local_addr().unwrap();
|
|
let mut buf = vec![0u8; 2048];
|
|
let mut restart: Option<SdpPendingOffer> = None;
|
|
loop {
|
|
let next = loop {
|
|
match rtc.poll_output() {
|
|
Ok(Output::Timeout(t)) => break t,
|
|
Ok(Output::Transmit(t)) => {
|
|
let _ = socket.try_send_to(&t.contents, t.destination);
|
|
}
|
|
Ok(Output::Event(e)) => {
|
|
let mut ev = events.lock().unwrap();
|
|
match e {
|
|
Event::Connected => ev.push(PeerEvent::Connected),
|
|
Event::IceConnectionStateChange(s) => ev.push(PeerEvent::Ice(s)),
|
|
Event::MediaAdded(m) => {
|
|
mids.insert(m.kind, m.mid);
|
|
}
|
|
Event::MediaData(d) => {
|
|
let kind = if d.params.spec().codec.is_audio() { MediaKind::Audio } else { MediaKind::Video };
|
|
ev.push(PeerEvent::Media { kind, data: d.data.to_vec(), rtp: d.time.numer() });
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
Err(_) => return,
|
|
}
|
|
};
|
|
let sleep = tokio::time::sleep_until(tokio::time::Instant::from_std(next.max(Instant::now())));
|
|
tokio::select! {
|
|
r = socket.recv_from(&mut buf) => {
|
|
if let Ok((n, from)) = r
|
|
&& let Ok(r) = Receive::new(Protocol::Udp, from, local, &buf[..n])
|
|
{
|
|
let _ = rtc.handle_input(Input::Receive(Instant::now(), r));
|
|
}
|
|
}
|
|
_ = sleep => {
|
|
let _ = rtc.handle_input(Input::Timeout(Instant::now()));
|
|
}
|
|
c = rx.recv() => match c {
|
|
None | Some(PeerCmd::Close) => {
|
|
rtc.disconnect();
|
|
let _ = rtc.poll_output();
|
|
return;
|
|
}
|
|
Some(PeerCmd::Write { kind, codec, data, ticks }) => {
|
|
let Some(&mid) = mids.get(&kind) else { continue };
|
|
let Some(pt) = rtc.writer(mid).and_then(|w| w.payload_params().find(|p| p.spec().codec == codec.to_str0m()).map(|p| p.pt())) else { continue };
|
|
let freq = if kind == MediaKind::Audio { Frequency::FORTY_EIGHT_KHZ } else { Frequency::NINETY_KHZ };
|
|
if let Some(w) = rtc.writer(mid) {
|
|
let _ = w.write(pt, Instant::now(), MediaTime::new(ticks, freq), data);
|
|
}
|
|
}
|
|
Some(PeerCmd::Restart { reply }) => {
|
|
let mut api = rtc.sdp_api();
|
|
api.ice_restart(true);
|
|
let (o, pending) = api.apply().expect("an ICE restart offer");
|
|
restart = Some(pending);
|
|
let _ = reply.send(o.to_sdp_string());
|
|
}
|
|
Some(PeerCmd::Restarted { frag, reply }) => {
|
|
let r = (|| {
|
|
let f = sdp::parse_frag(&frag).ok_or("the fragment reads")?;
|
|
let (u, p) = (f.ufrag.ok_or("an ice-ufrag")?, f.pwd.ok_or("an ice-pwd")?);
|
|
let a = sdp::restart_answer(&answer, &u, &p, &f.candidates);
|
|
let parsed = SdpAnswer::from_sdp_string(&a).map_err(|e| e.to_string())?;
|
|
let pending = restart.take().ok_or("a restart pending")?;
|
|
rtc.sdp_api().accept_answer(pending, parsed).map_err(|e| e.to_string())?;
|
|
answer = a;
|
|
Ok::<(), String>(())
|
|
})();
|
|
let _ = reply.send(r.map_err(|e| e.to_string()));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Peer {
|
|
pub fn write(&self, kind: MediaKind, codec: Codec, data: Vec<u8>, ticks: u64) {
|
|
let _ = self.cmd.send(PeerCmd::Write { kind, codec, data, ticks });
|
|
}
|
|
|
|
pub async fn restart_offer(&self) -> String {
|
|
let (tx, rx) = oneshot::channel();
|
|
self.cmd.send(PeerCmd::Restart { reply: tx }).unwrap();
|
|
timeout(STEP, rx).await.expect("in time").expect("an offer")
|
|
}
|
|
|
|
pub async fn restarted(&self, frag: String) -> Result<(), String> {
|
|
let (tx, rx) = oneshot::channel();
|
|
self.cmd.send(PeerCmd::Restarted { frag, reply: tx }).unwrap();
|
|
timeout(STEP, rx).await.expect("in time").expect("a result")
|
|
}
|
|
|
|
pub fn close(&self) {
|
|
let _ = self.cmd.send(PeerCmd::Close);
|
|
}
|
|
|
|
pub fn count(&self, f: impl Fn(&PeerEvent) -> bool) -> usize {
|
|
self.events.lock().unwrap().iter().filter(|e| f(e)).count()
|
|
}
|
|
|
|
pub async fn connected(&self) {
|
|
assert!(until(STEP, || self.count(|e| matches!(e, PeerEvent::Connected)) > 0).await, "the WebRTC session connects");
|
|
}
|
|
|
|
pub fn media(&self, kind: MediaKind) -> Vec<Vec<u8>> {
|
|
self.events.lock().unwrap().iter().filter_map(|e| if let PeerEvent::Media { kind: k, data, .. } = e { (*k == kind).then(|| data.clone()) } else { None }).collect()
|
|
}
|
|
}
|
|
|
|
/// Wait, bounded, until `f` holds.
|
|
pub async fn until(limit: Duration, mut f: impl FnMut() -> bool) -> bool {
|
|
let deadline = Instant::now() + limit;
|
|
while !f() {
|
|
if Instant::now() >= deadline {
|
|
return false;
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(20)).await;
|
|
}
|
|
true
|
|
}
|
|
|
|
/// POST a WHIP offer with `token` and start the publisher.
|
|
pub async fn publish(w: &World, auth: &Auth) -> Peer {
|
|
publish_with(w, auth, &[Codec::Avc, Codec::Opus], true, true).await
|
|
}
|
|
|
|
/// As [`publish`], offering `codecs`, and the sections asked for.
|
|
pub async fn publish_with(w: &World, auth: &Auth, codecs: &[Codec], audio: bool, video: bool) -> Peer {
|
|
let o = offer(codecs, Direction::SendOnly, audio, video).await;
|
|
let resp = auth.send("POST", &format!("{}/whip", w.base()), &[("content-type", "application/sdp")], o.offer.as_bytes()).await;
|
|
assert_eq!(resp.status, 201, "WHIP POST: {}", resp.text());
|
|
assert_eq!(resp.header("content-type"), Some("application/sdp"));
|
|
assert!(resp.header("wai-room-challenge").is_some(), "a challenge for the session's next request");
|
|
let p = o.start(&resp);
|
|
p.connected().await;
|
|
p
|
|
}
|
|
|
|
// ── Synthetic media ────────────────────────────────────────────────────────────────
|
|
|
|
/// A body of `n` bytes, none zero, so no start code appears inside a NAL unit.
|
|
fn body(seed: u64, n: usize) -> Vec<u8> {
|
|
(0..n).map(|j| ((seed.wrapping_mul(31).wrapping_add(j as u64 * 7)) % 255 + 1) as u8).collect()
|
|
}
|
|
|
|
/// H.264 access unit `i`: an IDR with its SPS and PPS when `key`, a non-IDR slice
|
|
/// otherwise, each NAL unit after a four-byte start code. Large enough to need FU-A.
|
|
pub fn avc(i: u64, key: bool) -> Vec<u8> {
|
|
let mut f = Vec::new();
|
|
if key {
|
|
f.extend_from_slice(&[0, 0, 0, 1, 0x67, 0x42, 0xe0, 0x1f, 0x8c, 0x8d, 0x40]);
|
|
f.extend_from_slice(&[0, 0, 0, 1, 0x68, 0xce, 0x3c, 0x80]);
|
|
f.extend_from_slice(&[0, 0, 0, 1, 0x65]);
|
|
f.extend_from_slice(&body(i, 3000));
|
|
} else {
|
|
f.extend_from_slice(&[0, 0, 0, 1, 0x41]);
|
|
f.extend_from_slice(&body(i, 700 + (i % 5) as usize * 100));
|
|
}
|
|
f
|
|
}
|
|
|
|
/// Opus packet `i`: a fullband CELT 20 ms stereo TOC, then bytes.
|
|
pub fn opus(i: u64) -> Vec<u8> {
|
|
let mut f = vec![0xfc];
|
|
f.extend_from_slice(&body(i + 7_000, 80 + (i % 7) as usize));
|
|
f
|
|
}
|
|
|
|
/// What a publisher sent, in order.
|
|
#[derive(Default, Clone)]
|
|
pub struct Sent {
|
|
pub video: Vec<Vec<u8>>,
|
|
pub audio: Vec<Vec<u8>>,
|
|
}
|
|
|
|
/// Send `seconds` of media in real time: 30 video frames a second, a keyframe every
|
|
/// `gop`, and 50 audio packets a second. Continues from what `sent` holds.
|
|
pub async fn send(p: &Peer, sent: &mut Sent, seconds: u64, gop: u64) {
|
|
let start = Instant::now();
|
|
let v0 = sent.video.len() as u64;
|
|
let a0 = sent.audio.len() as u64;
|
|
let (nv, na) = (30 * seconds, 50 * seconds);
|
|
let (mut v, mut a) = (0, 0);
|
|
while v < nv || a < na {
|
|
let tv = Duration::from_micros(v * 33_333);
|
|
let ta = Duration::from_micros(a * 20_000);
|
|
let next = tv.min(ta);
|
|
tokio::time::sleep_until(tokio::time::Instant::from_std(start + next)).await;
|
|
if v < nv && tv <= ta {
|
|
let i = v0 + v;
|
|
let f = avc(i, i.is_multiple_of(gop));
|
|
p.write(MediaKind::Video, Codec::Avc, f.clone(), 1_000 + i * 3000);
|
|
sent.video.push(f);
|
|
v += 1;
|
|
} else {
|
|
let i = a0 + a;
|
|
let f = opus(i);
|
|
p.write(MediaKind::Audio, Codec::Opus, f.clone(), 5_000 + i * 960);
|
|
sent.audio.push(f);
|
|
a += 1;
|
|
}
|
|
}
|
|
}
|
|
|
|
// ── A subscriber at the relay that verifies every receipt ─────────────────────────
|
|
|
|
pub async fn relay_client(w: &World) -> (Client, quinn::Endpoint) {
|
|
let ep = quic::client_endpoint("127.0.0.1:0".parse().unwrap(), vec![w.relay.identity.chain[0].clone()], vec![joule_moq::draft22::ALPN.to_vec()], None).unwrap();
|
|
let conn = timeout(STEP, quic::connect(&ep, w.relay.quic_addr.unwrap(), "localhost")).await.expect("in time").expect("a QUIC session");
|
|
let c = Client::connect(Arc::new(conn), ClientOptions::default()).await.expect("SETUP");
|
|
(c, ep)
|
|
}
|
|
|
|
/// One subscribed media track and its receipts track, as the subscriber reads them.
|
|
pub struct Watched {
|
|
pub kind: u8,
|
|
pub media_alias: u64,
|
|
pub receipts_alias: u64,
|
|
/// The request streams, held so the subscriptions stand.
|
|
pub held: Vec<ReqTx>,
|
|
pub receipts_rx: ReqRx,
|
|
}
|
|
|
|
/// What a subscriber read: each object by (Track Alias, group, object).
|
|
pub type Objects = Arc<Mutex<BTreeMap<(u64, u64, u64), Vec<u8>>>>;
|
|
|
|
pub struct Watcher {
|
|
pub client: Arc<Client>,
|
|
pub _ep: quinn::Endpoint,
|
|
pub tracks: Vec<Watched>,
|
|
pub objects: Objects,
|
|
pub ended: Arc<Mutex<Vec<u64>>>,
|
|
}
|
|
|
|
/// Subscribe at the relay to the owner's audio and video tracks and their receipts
|
|
/// tracks, from their start, reading every stream concurrently.
|
|
pub async fn watch(w: &World) -> Watcher {
|
|
watch_kinds(w, &[1, 2]).await
|
|
}
|
|
|
|
/// As [`watch`], for the media kinds `kinds` (1 audio, 2 video).
|
|
pub async fn watch_kinds(w: &World, kinds: &[u8]) -> Watcher {
|
|
let (c, ep) = relay_client(w).await;
|
|
let c = Arc::new(c);
|
|
let owner = w.owner();
|
|
let mut tracks = Vec::new();
|
|
let ended = Arc::new(Mutex::new(Vec::new()));
|
|
for &kind in kinds {
|
|
let ns = catalog::media_namespace(&w.room, &owner, kind);
|
|
let (m, mtx, mrx) = c.subscribe(ns.clone(), TrackName::from_str("0"), Parameters::new()).await.expect("SUBSCRIBE the media track");
|
|
let (r, rtx, rrx) = c.subscribe(ns.clone(), catalog::receipts_name("0"), Parameters::new()).await.expect("SUBSCRIBE the receipts track");
|
|
// PUBLISH_DONE on the media track: the track ended.
|
|
let ended = ended.clone();
|
|
let alias = m.track_alias;
|
|
let mut mrx = mrx;
|
|
tokio::spawn(async move {
|
|
while let Ok(Some(msg)) = mrx.next().await {
|
|
if let ControlMessage::PublishDone(_) = msg {
|
|
ended.lock().unwrap().push(alias);
|
|
return;
|
|
}
|
|
}
|
|
});
|
|
tracks.push(Watched { kind, media_alias: m.track_alias, receipts_alias: r.track_alias, held: vec![mtx, rtx], receipts_rx: rrx });
|
|
}
|
|
let objects: Objects = Arc::default();
|
|
let (c2, o2) = (c.clone(), objects.clone());
|
|
tokio::spawn(async move {
|
|
loop {
|
|
let Some(mut r) = c2.next_stream(Duration::from_secs(1)).await else {
|
|
if Arc::strong_count(&o2) == 1 {
|
|
return;
|
|
}
|
|
continue;
|
|
};
|
|
let o2 = o2.clone();
|
|
tokio::spawn(async move {
|
|
let Ok(Ok(h)) = timeout(STEP, r.start()).await else { return };
|
|
while let Ok(Ok(Some(e))) = timeout(STEP, r.next()).await {
|
|
match (&h, e) {
|
|
(StreamHeader::Subgroup(sh), UniEvent::SubgroupObject(o)) if !o.payload.is_empty() => {
|
|
o2.lock().unwrap().insert((sh.track_alias, sh.group_id, o.object_id), o.payload);
|
|
}
|
|
(StreamHeader::Fetch(_), UniEvent::FetchEntry(FetchEntry::Object(_))) => {}
|
|
_ => {}
|
|
}
|
|
}
|
|
});
|
|
}
|
|
});
|
|
Watcher { client: c, _ep: ep, tracks, objects, ended }
|
|
}
|
|
|
|
/// Every media object of `t` verified against its receipts under `key`: returns the
|
|
/// frames, in group and object order, and the number of groups whose receipts verified.
|
|
pub fn verified(wt: &Watcher, t: &Watched, key: &SigningKey, codec: Codec) -> (Vec<Vec<u8>>, usize) {
|
|
let v = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
|
|
let objs = wt.objects.lock().unwrap();
|
|
let mut bundles = BTreeMap::new();
|
|
for ((alias, g, _), p) in objs.iter() {
|
|
if *alias == t.receipts_alias {
|
|
let b = ReceiptBundle::from_payload(p).expect("a receipts bundle");
|
|
let vg = v.verify_group(&b.group).expect("the group receipt verifies under the gateway's key");
|
|
assert_eq!(b.group.group_id, GroupId(*g), "a receipt names its MOQT group");
|
|
bundles.insert(*g, (b, vg));
|
|
}
|
|
}
|
|
let mut frames = Vec::new();
|
|
let mut prev: Option<(u64, [u8; 32])> = None;
|
|
for (g, (b, _)) in &bundles {
|
|
if let Some((pg, link)) = prev
|
|
&& pg + 1 == *g
|
|
{
|
|
assert_eq!(b.group.parent_receipt_hash, Some(link), "group {g} chains to group {pg}");
|
|
}
|
|
prev = Some((*g, b.group.link_hash()));
|
|
}
|
|
for ((alias, g, o), p) in objs.iter() {
|
|
if *alias != t.media_alias {
|
|
continue;
|
|
}
|
|
let (b, vg) = bundles.get(g).unwrap_or_else(|| panic!("the receipts of group {g} arrived"));
|
|
let r = &b.objects[*o as usize];
|
|
r.check_moqt(b.group.track_alias, GroupId(*g), ObjectId(*o)).expect("the receipt names its object");
|
|
v.verify_payload(p, r, vg).expect("every object verifies against its signed root");
|
|
let opened = frame::open(p).expect("an object is one WAI1 envelope");
|
|
assert_eq!(opened.codec, codec);
|
|
frames.push(frame::unframe(codec, &opened.payload).expect("one frame in its framing"));
|
|
}
|
|
// Every object every receipt names arrived.
|
|
let named: usize = bundles.values().map(|(b, _)| b.objects.len()).sum();
|
|
assert_eq!(named, frames.len(), "every receipted object arrived");
|
|
(frames, bundles.len())
|
|
}
|
|
|
|
pub fn ns_of(w: &World, kind: u8) -> TrackNamespace {
|
|
catalog::media_namespace(&w.room, &w.owner(), kind)
|
|
}
|
|
|
|
/// Wait, bounded, until the subscriber has read nothing new for `quiet`.
|
|
pub async fn settle(wt: &Watcher, quiet: Duration) {
|
|
let deadline = Instant::now() + STEP;
|
|
let mut last = (wt.objects.lock().unwrap().len(), Instant::now());
|
|
while Instant::now() < deadline {
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
let n = wt.objects.lock().unwrap().len();
|
|
if n != last.0 {
|
|
last = (n, Instant::now());
|
|
} else if last.1.elapsed() >= quiet {
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Wait, bounded, until the gateway's WHIP session has taken `n` frames: everything sent
|
|
/// has arrived before the session ends.
|
|
pub async fn received(w: &World, p: &Peer, n: usize) -> bool {
|
|
let id = p.location.rsplit('/').next().unwrap().to_string();
|
|
until(STEP, || w.gateway.gateway.session_frames(false, &id).is_some_and(|f| f as usize >= n)).await
|
|
}
|
|
|
|
/// `got` is a run of `sent`, from where its first element is: that index.
|
|
pub fn run_of(sent: &[Vec<u8>], got: &[Vec<u8>]) -> usize {
|
|
assert!(!got.is_empty(), "something arrived");
|
|
let j = sent.iter().position(|f| *f == got[0]).expect("the first frame received is one that was sent");
|
|
assert!(j + got.len() <= sent.len(), "more frames received than were sent after the first: {} from {} of {}", got.len(), j, sent.len());
|
|
same(got, &sent[j..j + got.len()], "the frames received are the frames sent, in order, byte for byte, from the first received");
|
|
j
|
|
}
|
|
|
|
/// The payloads of track `t`'s media objects, in group and object order, as envelopes.
|
|
pub fn envelopes(wt: &Watcher, t: &Watched) -> Vec<frame::Opened> {
|
|
wt.objects.lock().unwrap().iter().filter(|(k, _)| k.0 == t.media_alias).map(|(_, p)| frame::open(p).expect("an envelope")).collect()
|
|
}
|
|
|
|
/// `got` equals `want`, frame for frame; on failure, says where they part, not every byte.
|
|
pub fn same(got: &[Vec<u8>], want: &[Vec<u8>], what: &str) {
|
|
if got == want {
|
|
return;
|
|
}
|
|
let at = got.iter().zip(want).position(|(a, b)| a != b);
|
|
panic!("{what}: {} received, {} sent, first difference at {:?}", got.len(), want.len(), at);
|
|
}
|
|
|
|
/// DELETE a session's resource with `token`.
|
|
pub async fn delete(w: &World, location: &str, auth: &Auth) -> Resp {
|
|
auth.send("DELETE", &format!("{}{}", w.base(), location), &[], b"").await
|
|
}
|