//! Uncompressed essence into a room's tracks, end to end on loopback (WAI //! `extensions/essence-contribution.md` §10): a publisher POSTs `/st2110` with its grant, the //! description of a video stream, a sound stream and an ancillary data stream, and the source //! it states; this crate's own sender sends them over UDP, from pictures and sound it //! computes, with one video packet and one sound packet never sent. A subscriber at the relay //! receives every group of both tracks and verifies every receipt under the gateway's key: //! every video object is AV1 within the video floor, and the sound decodes to the samples //! sent, bit for bit, less the packet that was lost. The session's record states each //! stream, counts the loss and the frame not published for it, says where each capture //! instant came from, counts the ancillary data not carried, and holds an origin claim for //! each track; the SCTE 104 message in the ancillary data became the room's signed cues. //! Then the refusals, a multicast group joined on loopback, and an encoder asked for that is //! not here: the gateway does not start, or starts on the next one and says so in the //! session's answer and its record. The tests that hold a session to its frame rate at //! broadcast sizes are `tests/essence_rt.rs`. //! //! The sender here holds no PTP clock. It counts its RTP timestamps from the system clock //! on the PTP timescale, as a sender whose media clock is direct with offset 0 would, and //! its description says so: a fixture for the path that reads a capture instant from an RTP //! timestamp, not a claim about this host's clock. mod common; use common::*; use ed25519_dalek::{SigningKey, Verifier}; use joule_moq::draft22::{Parameters, StreamHeader, UniEvent}; use joule_moq::TrackName; use joule_moq_ingest::catalog; use joule_moq_ingest::frame::Codec; use joule_moq_ingest::room::{build, media, Descriptor}; use joule_moq_ingest::st2110::sdp::{AudioFormat, Sampling}; use joule_moq_ingest::st2110::send::{self, anc_packet, audio_packet, description, scte104_anc, video_format, video_packets, Clocks, Media, Out}; use joule_moq_ingest::st2110::{flac, session}; use serde_json::{json, Value}; use std::sync::{Arc, Mutex}; use std::time::Duration; use tokio::net::UdpSocket; use tokio::time::timeout; const INJECTOR_SEED: [u8; 32] = [0x72; 32]; const LEAP: u64 = 37; async fn world(f: impl FnOnce(&mut joule_moq_ingest::GatewayConfig)) -> World { // The relay's door holds the media tracks to their receipts; an events track carries none. let door = |rc: &mut joule_moq_relay::RelayConfig, room: &[u8; 32], gk: &str| { rc.receipts.door = ["audio", "video"] .iter() .map(|k| joule_moq_relay::config::DoorRule { prefix: vec![hex::encode(room), "*".into(), k.to_string()], trust: vec![gk.to_string()], trust_field: None, hold_ms: 10_000 }) .collect(); }; World::start_with(door, |c| { let authority = SigningKey::from_bytes(&[0x51; 32]); let d = Descriptor::read(c.rooms[0].descriptor.as_ref().unwrap()).unwrap(); let key = SigningKey::from_bytes(&INJECTOR_SEED); let s = now_us() / 1_000_000; let g = build::grant(&d, &authority, &key.verifying_key().to_bytes(), 0, s - 60, s + 3600); c.injector.listen = Some("127.0.0.1:0".parse().unwrap()); c.injector.room = hex::encode(d.room_id()); c.injector.key_seed = Some(INJECTOR_SEED); // The principal's chain, kept across a restart of this gateway (a key given in code needs a // place for it). c.injector.state_path = Some(std::env::temp_dir().join(format!("wai-ingest-injector-st2110-{}-{}.chain.json", std::process::id(), std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()))); c.injector.grant = Some(g.to_bytes()); c.st2110.enabled = true; // The encoder every host has: what a session does is the same on each. The // real-time encoders are held to their rates in `tests/essence_rt.rs`. c.st2110.encoder = "rav1e".into(); c.st2110.bind_ip = "127.0.0.1".parse().unwrap(); c.st2110.interface = "127.0.0.1".parse().unwrap(); c.contrib.idle_ms = 1_500; f(c); }) .await } async fn open(w: &World, auth: &Auth, body: &Value) -> Resp { auth.send("POST", &format!("{}/st2110", w.base()), &[("content-type", "application/json")], &serde_json::to_vec(body).unwrap()).await } /// The count of a `rate` clock on the PTP timescale at `unix_us`. fn ticks(unix_us: u64, rate: u64) -> u32 { ((unix_us as u128 + LEAP as u128 * 1_000_000) * rate as u128 / 1_000_000) as u32 } /// A `multiple_operation_message()` of one immediate splice-out of 1.5 s with auto-return /// (SCTE 104 §7, Table 9-5), as `tests/scte104_injection.rs` builds it. fn splice_message() -> Vec { let mut d = vec![2u8]; d.extend_from_slice(&0x10u32.to_be_bytes()); d.extend_from_slice(&[0, 9]); d.extend_from_slice(&0u16.to_be_bytes()); d.extend_from_slice(&15u16.to_be_bytes()); d.extend_from_slice(&[1, 2, 1, 0]); let mut o = vec![0xff, 0xff, 0, 0, 0, 3, 2, 0, 1, 0, 0, 1]; o.extend_from_slice(&0x0101u16.to_be_bytes()); o.extend_from_slice(&(d.len() as u16).to_be_bytes()); o.extend_from_slice(&d); let len = o.len() as u16; o[2..4].copy_from_slice(&len.to_be_bytes()); o } /// Each object of a track with the capture instant and the hop time its properties state. type Stamps = Arc, Option)>>>; async fn watch_stamps(w: &World, kind: u8) -> (Stamps, Vec) { let (c, ep) = relay_client(w).await; let c = Arc::new(c); let ns = catalog::media_namespace(&w.room, &w.owner(), kind); let (_, tx, _rx) = c.subscribe(ns, TrackName::from_str("0"), Parameters::new()).await.expect("SUBSCRIBE"); let out: Stamps = Arc::default(); let o2 = out.clone(); tokio::spawn(async move { let _ep = ep; while let Some(mut r) = c.next_stream(Duration::from_secs(30)).await { 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 { if let (StreamHeader::Subgroup(sh), UniEvent::SubgroupObject(o)) = (&h, e) { let one = |p: &joule_moq::draft22::kvp::Properties, ty: u64| p.0.iter().find(|k| k.ty == ty).and_then(|k| k.as_bytes()).and_then(|b| joule_moq::draft22::varint::get(b, "a timestamp").ok()).map(|v| v.0); let inner = o.properties.immutable().ok().flatten().unwrap_or_default(); o2.lock().unwrap().push((sh.group_id, o.object_id, one(&inner, joule_moq::tempo::CAPTURE_TIMESTAMP), one(&o.properties, joule_moq::tempo::HOP_TIME))); } } }); } }); (out, vec![tx]) } /// The (group, object) of every object a watcher holds of a track. fn w_objects(wt: &Watcher, t: &Watched) -> Vec<(u64, u64)> { wt.objects.lock().unwrap().keys().filter(|k| k.0 == t.media_alias).map(|k| (k.1, k.2)).collect() } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn uncompressed_essence_reaches_a_relay_subscriber_with_every_receipt() { timeout(Duration::from_secs(240), run()).await.expect("the test finishes in time"); } async fn run() { let w = world(|_| {}).await; let auth = w.auth(&w.publisher, 0); const FRAMES: u32 = 50; let vf = video_format(Sampling::YCbCr422, 10, 128, 72, (25, 1)); let af = AudioFormat { bits: 24, rate: 48_000, channels: 2, ptime_us: Some(1_000), channel_order: Some("SMPTE2110.(ST)".into()) }; let text = description(&[Media::Video(vf.clone()), Media::Audio(af.clone()), Media::Anc(vec![(0x41, 0x07), (0x61, 0x01), (0x60, 0x60), (0x61, 0x02)])], "127.0.0.1", &[30_000, 30_002, 30_004], Clocks::Traceable, Some("SAMP")); let r = open(&w, &auth, &json!({"sdp": text, "source": "digitalCapture", "captions": {"lang": "en", "origin": "human"}})).await; assert_eq!(r.status, 201, "POST /st2110: {}", r.text()); let s: Value = serde_json::from_slice(&r.body).unwrap(); let id = s["session"].as_str().unwrap().to_string(); assert_eq!(r.header("location"), Some(format!("/st2110/s/{id}").as_str())); assert_eq!((s["streams"][0]["clock"].as_str(), s["streams"][0]["tsmode"].as_str(), s["origin_claim"].as_bool()), (Some("ptp_traceable_stated"), Some("SAMP"), Some(true))); // The answer names the encoder: the one asked for, software, and no fallback. assert_eq!((s["encoder"]["backend"].as_str(), s["encoder"]["asked"].as_str(), s["encoder"]["hardware"].as_bool(), s["encoder"]["fallback"].is_null()), (Some("rav1e"), Some("rav1e"), Some(false), true), "{}", s["encoder"]); let ports: Vec = s["streams"].as_array().unwrap().iter().map(|x| x["port"].as_u64().unwrap() as u16).collect(); assert!(contrib_live(&w, &id).await, "the session publishes its tracks"); let watcher = watch(&w).await; let (stamps, _held) = watch_stamps(&w, media::VIDEO).await; let cat = watch_tracks(&w, vec![(0, catalog::catalog_namespace(&w.room, &w.owner()), "catalog".into())]).await; let injector = w.gateway.injector.clone().expect("the injector runs"); // The sender: one socket a stream, a frame every 40 ms with its sound and its ancillary data. let mut socks = Vec::new(); for p in &ports { let sk = UdpSocket::bind("127.0.0.1:0").await.unwrap(); sk.connect(("127.0.0.1", *p)).await.unwrap(); socks.push(sk); } let (mut vo, mut ao, mut no) = (Out::new(96, 0x5a5a_0001, 65_530), Out::new(97, 0x5a5a_0002, 65_530), Out::new(98, 0x5a5a_0003, 0)); // Each frame's time code (VITC at 25 frames a second, 10:00:SS:FF), a CDP stating it, and // a pop-on caption in 608 data on field 1, a pair a frame, shown at frame 19 (its first // End of Caption) and erased at frame 30. let label = |n: u32| [10u8, 0, (n / 25) as u8, (n % 25) as u8]; // CTA-708 in the CDPs (essence-contribution §8.5), the caption service directory of each // CDP stating service 1 in French and service 2 in Spanish, and service 3 not at all. // Service 2 shows a window at frame 12 and deletes it at frame 22. Services 1 and 3 each // show one at frame 32 and delete it at frame 44: service 1, the primary service, is // carried in the publisher's English (the directory's French counted as differing), and // service 3, with no stated language, is not carried. Each a caption channel packet of // a service block a service, its constructs first. let packet = |seq: u8, blocks: &[(u8, &[u8])]| -> Vec<[u8; 3]> { let mut p = vec![0u8]; for (service, data) in blocks { p.push(service << 5 | data.len() as u8); p.extend_from_slice(data); } if p.len() % 2 == 1 { p.push(0); } p[0] = seq << 6 | (p.len() / 2) as u8; p.chunks(2).enumerate().map(|(i, c)| [if i == 0 { 0xff } else { 0xfe }, c[0], c[1]]).collect() }; let window = |text: &[u8]| -> Vec { [&[0x98, 0x38, 74, 104, 0x70, 31, 0x19][..], text, &[0x03][..]].concat() }; let shown = packet(0, &[(2, &window(b"HOLA")[..])]); let deleted = packet(1, &[(2, &[0x8c, 0xff][..])]); let primary = packet(2, &[(1, &window(b"UNO")[..]), (3, &window(b"TRES")[..])]); let gone = packet(3, &[(1, &[0x8c, 0xff][..]), (3, &[0x8c, 0xff][..])]); let dtvcc_constructs = (shown.len() + deleted.len() + primary.len() + gone.len()) as u64; let cdp = |n: u32| { let mut cc = match n { 12 => shown.clone(), 22 => deleted.clone(), 32 => primary.clone(), 44 => gone.clone(), _ => Vec::new(), }; cc.resize(24, [0xfa, 0x00, 0x00]); let directory = vec![[0x81, b'f', b'r', b'e', 0xc1, 0x7f, 0xff], [0x82, b's', b'p', b'a', 0xc2, 0x7f, 0xff]]; send::cdp(&send::CdpParts { frame_rate: 3, sequence: n as u16, time: Some((label(n), 0, false)), cc: Some(cc), active: true, svc: Some((0x50, directory)) }) }; // Two pop-on captions in 608: HELLO, shown at frame 19 and erased at 30, carried; BYE, // shown at frame 42 once service 1 has shown a caption, counted and not carried // (CTA-708-E §9.24), and erased at 46. let mut cc: Vec<[u8; 2]> = vec![[0x14, 0x20], [0x14, 0x20], [0x14, 0x2e], [0x14, 0x2e], [0x14, 0x70], [0x14, 0x70], *b"HE", *b"LL", [b'O', 0], [0x14, 0x2f], [0x14, 0x2f]]; cc.extend([[0, 0]; 9]); cc.extend([[0x14, 0x2c], [0x14, 0x2c]]); cc.extend([[0, 0]; 2]); cc.extend([[0x14, 0x20], [0x14, 0x20], [0x14, 0x2e], [0x14, 0x2e], [0x14, 0x70], [0x14, 0x70], *b"BY", [b'E', 0], [0x14, 0x2f], [0x14, 0x2f]]); cc.extend([[0, 0]; 2]); cc.extend([[0x14, 0x2c], [0x14, 0x2c]]); let pair = |n: u32| -> [u8; 2] { let p = if (10..10 + cc.len() as u32).contains(&n) { cc[(n - 10) as usize] } else { [0, 0] }; [joule_moq_ingest::st2110::captions::odd(p[0]), joule_moq_ingest::st2110::captions::odd(p[1])] }; let start = now_us(); let audio_rtp0 = ticks(start, 48_000); let mut pcm_sent: Vec = Vec::new(); let started = std::time::Instant::now(); for n in 0..FRAMES { let at = start + n as u64 * 40_000; let rtp = ticks(at, 90_000); for (k, p) in video_packets(&vf, &send::picture(&vf, n), rtp, &mut vo, 1_440).into_iter().enumerate() { // One packet of frame 7 is never sent. if n == 7 && k == 1 { continue; } socks[0].send(&p).await.unwrap(); } let mut ancs = vec![send::atc(label(n), &[], [0; 8], 1, 0), cdp(n), send::cc608(1, 12, pair(n))]; if n == 5 { ancs.extend(scte104_anc(&splice_message(), false)); } socks[2].send(&anc_packet(&ancs, 0, rtp, true, &mut no)).await.unwrap(); for k in 0..40u32 { let first = (n * 40 + k) as u64 * 48; let samples = send::sound(&af, first, 48); let p = audio_packet(&af, &samples, audio_rtp0.wrapping_add(first as u32), &mut ao); // One packet of sound is never sent. if n == 9 && k == 3 { continue; } pcm_sent.extend_from_slice(&samples); socks[1].send(&p).await.unwrap(); } tokio::time::sleep_until((started + Duration::from_millis(40 * (n as u64 + 1))).into()).await; } // The session ends when nothing has come for its idle time, and signs its record. assert!(until(Duration::from_secs(120), || w.gateway.gateway.records().iter().any(|r| r.record.protocol == "st2110")).await, "the session's record is signed"); assert!(until(STEP, || watcher.ended.lock().unwrap().len() == 2).await, "PUBLISH_DONE on both media tracks"); settle(&watcher, Duration::from_millis(500)).await; let rec = w.gateway.gateway.records().into_iter().find(|r| r.record.protocol == "st2110").unwrap(); rec.verify().unwrap(); let d = &rec.record.detail; assert_eq!(rec.record.ended, "idle"); // Video: every object verifies, and is AV1 within the video floor. let video = watcher.tracks.iter().find(|t| t.kind == media::VIDEO).unwrap(); let audio = watcher.tracks.iter().find(|t| t.kind == media::AUDIO).unwrap(); let (units, groups) = verified(&watcher, video, &w.gateway_key, Codec::Av1); assert!(groups >= 1 && !units.is_empty(), "the video track carries a group"); let env = envelopes(&watcher, video); for e in &env { // A keyframe's object holds the sequence header: the Main profile, the Main tier, level // 4.1 or below, and a framing within the level. The frames after it are of its sequence. if e.key { joule_moq_ingest::video_floor::check_payload(&e.payload).unwrap_or_else(|o| panic!("an object outside the video floor: {}", o.detail())); } let dim = |i: usize| u32::from_le_bytes(e.payload[4 * i..4 * i + 4].try_into().unwrap()); assert_eq!((dim(1), dim(2)), (72, 128), "the framing states the picture's size"); assert_eq!(e.manifest["model_requirement"]["capability"], "wai.video.av1"); assert_eq!(e.manifest["live"]["rate"], 90_000); } assert!(env[0].key, "a group begins at a keyframe"); let h = joule_moq_ingest::frame::av1_header(&units[0]).expect("a keyframe holds a sequence header"); assert_eq!((h.width, h.height), (128, 72)); let rtps: Vec = env.iter().map(|e| e.manifest["live"]["rtp"].as_u64().unwrap()).collect(); assert!(rtps.windows(2).all(|x| x[1] > x[0] && (x[1] - x[0]).is_multiple_of(3_600)), "each object's RTP timestamp is its frame's: {rtps:?}"); // Every frame sent is accounted for: published, not published for a missing sample, or // not published because the encoder was behind. Nothing is drawn for the lost packet. let sv = &d["streams"][0]; let behind = sv["not_published"]["frames_encoder_behind_not_published"].as_u64().unwrap_or(0); assert_eq!((sv["lost"].as_u64(), sv["read"]["frames"].as_u64(), sv["read"]["frames_incomplete"].as_u64(), sv["not_published"]["frames_incomplete_not_published"].as_u64()), (Some(1), Some(FRAMES as u64), Some(1), Some(1))); assert_eq!(units.len() as u64 + 1 + behind, FRAMES as u64, "every frame is published or counted: {} published, {behind} behind the encoder", units.len()); let frame7 = ticks(start + 7 * 40_000, 90_000) as u64; assert!(rtps.iter().all(|r| *r as u32 as u64 != frame7), "the frame with a missing row is not published"); // The stage's own counts say the same: every whole frame was offered to it, and each // was encoded or counted as finding its queue full. Nothing else is dropped. let enc = &d["encoder"]; let stage = &enc["stage"]; assert_eq!((enc["backend"].as_str(), enc["asked"].as_str(), enc["fallback"].is_null()), (Some("rav1e"), Some("rav1e"), true)); assert_eq!((stage["frames_offered"].as_u64(), stage["frames_not_encoded_queue_full"].as_u64(), stage["frames_encoded"].as_u64()), (Some(FRAMES as u64 - 1), Some(behind), Some(units.len() as u64)), "{stage}"); assert_eq!(stage["drop_policy"], joule_moq_ingest::st2110::encode::DROP_POLICY); assert!(stage["keyframes"].as_u64().unwrap() >= 1 && stage["rungs"]["count"].as_u64().unwrap() >= 1); println!("essence_st2110: {} of {} frames of 128 x 72 at 25 a second published in this build by {} ({} found the encoder's queue full): {}", units.len(), FRAMES, enc["backend"], behind, stage); // Sound: FLAC objects that decode to the samples sent, bit for bit; the lost packet's // 48 sample frames are counted missing and none is made for them. let (streams, _) = verified(&watcher, audio, &w.gateway_key, Codec::Flac); let mut got: Vec = Vec::new(); for st in &streams { let (samples, ch, bits, rate) = flac::read(st).expect("a FLAC stream of the session's writer"); assert_eq!((ch, bits, rate), (2, 24, 48_000)); got.extend(samples); } assert_eq!(got.len(), pcm_sent.len()); assert!(got == pcm_sent, "the sound is the sound sent"); let sa = &d["streams"][1]; assert_eq!((sa["lost"].as_u64(), sa["read"]["timestamp_jumps"].as_u64(), sa["read"]["sample_frames_missing"].as_u64()), (Some(1), Some(1), Some(48))); let aenv = envelopes(&watcher, audio); let lens: Vec = aenv.windows(2).map(|x| x[1].manifest["live"]["rtp"].as_u64().unwrap() - x[0].manifest["live"]["rtp"].as_u64().unwrap()).collect(); assert!(lens.iter().filter(|l| **l != 960).count() <= 2, "an object every 20 ms, but at the break: {lens:?}"); assert_eq!(w.relay.relay().stats().refused_at_door, 0, "the relay's door verified every object"); // Time: each video object states a capture instant read from its RTP timestamp, within // the frame's own time, and a hop time; the record says where the instants came from. let st = stamps.lock().unwrap().clone(); assert_eq!(st.len(), units.len(), "the stamped subscriber saw every object"); for (_, _, capture, hop) in &st { let (c, hp) = (capture.expect("a CaptureTimestamp"), hop.expect("a HopTime")); assert!(c + 1 >= start / 1000 && c <= (start + FRAMES as u64 * 40_000) / 1000 + 1, "a capture instant within the frames sent: {c}"); assert!(hp >= c, "the hop time is the relay's or the gateway's, after the capture"); } assert!(sv["capture_instants"]["from_rtp_timestamp"].as_u64().unwrap() >= FRAMES as u64 - 1); assert_eq!((sv["capture_instants"]["from_arrival"].as_u64(), sv["capture_instants"]["rtp_disagreed_with_gateway_clock"].as_u64()), (Some(0), Some(0))); assert_eq!(d["reference_clock"], "none held by the gateway: no lock is claimed"); assert_eq!((sv["statement"]["clock"].as_str(), sv["statement"]["tsmode"].as_str(), sv["statement"]["sender_type_checked"].as_str()), (Some("ptp_traceable_stated"), Some("SAMP"), Some("no"))); assert_eq!(sv["statement"]["ssrc"], 0x5a5a_0001u32); // The catalog states the tracks' playout timing. settle(&cat, Duration::from_millis(300)).await; let last = { let objs = cat.objects.lock().unwrap(); objs.iter().rfind(|(k, _)| k.0 == cat.tracks[0].media_alias).map(|(_, b)| serde_json::from_slice::(b).unwrap()).expect("a catalog") }; let wai: Vec<&Value> = last["tracks"].as_array().unwrap().iter().filter(|t| t["packaging"] == "wai").collect(); assert_eq!(wai.len(), 2); assert!(wai.iter().all(|t| t["wai_tempo"] == json!({"max_delay_ms": 0, "playout_delay_ms": 0}))); assert!(wai.iter().any(|t| t["wai_capabilities"] == json!(["wai.video.av1"])) && wai.iter().any(|t| t["wai_capabilities"] == json!(["wai.audio.flac"]))); // Ancillary data: every packet is read. The SCTE 104 message became the room's cues, // signed by its injector; the time codes went onto the video objects; the CDPs were read // and their cc_data decoded; the first 608 caption and the windows of 708 services 2 and 1 // each became a partial and a final revision of a caption event, signed by the same // principal. Service 3, which the directory gives no language, is decoded and not // carried; service 1's French in the directory, beside the publisher's English, is // counted; and the 608 caption after service 1 has shown one is not carried (CTA-708-E // §9.24 and §4.5.1, essence-contribution §10.6). assert_eq!(d["ancillary_not_carried"], json!({})); assert_eq!(d["scte104"], json!({"injected": 1})); let a = &d["time_code_and_captions"]; assert_eq!((a["time_codes_read"].as_u64(), a["cdps_read"].as_u64(), a["cdp_constructs_without_data"].as_u64(), a["cc608_packets_field_1"].as_u64()), (Some(FRAMES as u64), Some(FRAMES as u64), Some(24 * FRAMES as u64 - dtvcc_constructs), Some(FRAMES as u64)), "{a}"); assert_eq!((a["cdp_constructs_dtvcc"].as_u64(), a["dtvcc_packets"].as_u64()), (Some(dtvcc_constructs), Some(4)), "{a}"); assert_eq!((a["captions_decoded"].as_u64(), a["captions_carried"].as_u64(), a["captions_finished"].as_u64()), (Some(4), Some(3), Some(3)), "{a}"); assert_eq!((a["captions_not_carried:service_language_unstated"].as_u64(), a["captions_not_carried:cc608_beside_cta708"].as_u64(), a["caption_service_1_language_differs"].as_u64()), (Some(1), Some(1), Some(1)), "{a}"); assert!(a.get("time_codes_discontinuous").is_none() && a.get("cdps_discontinuous").is_none(), "{a}"); assert_eq!(d["captions_stated"], json!({"lang": "en", "origin": 1, "model": null})); let video = watcher.tracks.iter().find(|t| t.kind == media::VIDEO).unwrap(); let env = envelopes(&watcher, video); let rtp0 = ticks(start, 90_000); assert!(!env.is_empty()); for e in &env { let n = (e.manifest["live"]["rtp"].as_u64().unwrap() as u32).wrapping_sub(rtp0) / 3600; let [h, m, s, f] = label(n); assert_eq!(e.manifest["live"]["tc"], json!({"label": format!("{h:02}:{m:02}:{s:02}:{f:02}"), "drop": false, "interpolated": false}), "the time code of frame {n}"); } assert_eq!(a["video_objects_with_time_code"].as_u64(), Some(env.len() as u64), "{a}"); assert!(until(STEP, || injector.events().len() >= 22).await, "an out and its in, each after its injection record, and two revisions of each of three captions, each with its annotation and its window: {}", injector.events().len()); let events = injector.events(); let anchored: Vec<&Vec> = events.iter().filter(|e| e[117] == 1).collect(); assert_eq!(anchored.len(), 18, "the captions' revisions, their annotations and their windows are anchored"); let video_track = catalog::track_digest(&catalog::media_namespace(&w.room, &w.owner(), media::VIDEO), "0"); let published: Vec<(u64, u64)> = w_objects(&watcher, video); for e in &anchored { assert_eq!(&e[118..150], &video_track[..], "anchored to the video track"); let (g, o) = (u64::from_le_bytes(e[150..158].try_into().unwrap()), u64::from_le_bytes(e[158..166].try_into().unwrap())); assert!(published.contains(&(g, o)), "anchored to an object published: ({g}, {o})"); } let payload = |e: &[u8]| e[169..169 + u16::from_le_bytes([e[167], e[168]]) as usize].to_vec(); // Each caption event is followed by its annotation and its window (timed-text §4.7); the // 708 window's first, shown before the 608 caption and gone before it. assert!(anchored.chunks(3).all(|p| (p[0][166], p[1][166], p[2][166]) == (0x0a, 0x01, 0x01))); let of = |text: &[u8]| -> Vec<&Vec> { anchored.chunks(3).filter(|p| payload(p[0]).ends_with(text)).flat_map(|p| [p[0], p[1], p[2]]).collect() }; let (hola, uno, anchored) = (of(b"HOLA"), of(b"UNO"), of(b"HELLO")); assert_eq!((hola.len(), uno.len(), anchored.len()), (6, 6, 6)); assert!(of(b"TRES").is_empty() && of(b"BYE").is_empty(), "service 3 and the 608 caption beside service 1 are not carried"); // A final's duration_ms is how long it is shown from its own t (timed-text §4.1): the // time left to the instant its caption left, the partial's t plus the media time from its // frame to the frame that removed it, and at least 1. So the cue it resolves to (§4.5) // ends where the caption left the screen, or 1 ms after the final's t where the final // was signed later, and is never shown again for the time it was shown. let t_of = |e: &[u8]| u64::from_le_bytes(e[109..117].try_into().unwrap()); let ends_where_it_left = |c: &[&Vec], text: &[u8], lang: &[u8], shown_ms: u64, what: &str| { let (tp, tf) = (t_of(c[0]) / 1000, t_of(c[3]) / 1000); let d = u32::from_le_bytes(payload(c[3])[..4].try_into().unwrap()) as u64; assert_eq!(payload(c[3])[4..], [&[1, lang.len() as u8][..], lang, &(text.len() as u16).to_le_bytes(), text].concat()[..], "{what}: the final's text"); assert!(d >= 1, "{what}: a final states a duration"); assert_eq!(tf + d, (tp + shown_ms).max(tf + 1), "{what}: shown {shown_ms} ms from the partial's t ({tp} ms), the final signed at {tf} ms with duration_ms {d}: the cue ends where the caption left"); }; assert_eq!(payload(uno[0]), [&0u32.to_le_bytes()[..], &[1, 2], b"en", &3u16.to_le_bytes(), b"UNO"].concat(), "service 1 in the publisher's language, not the directory's"); ends_where_it_left(&uno, b"UNO", b"en", 480, "shown from frame 32 to frame 44"); assert_eq!(payload(hola[0]), [&0u32.to_le_bytes()[..], &[1, 9], b"es-x-svc2", &4u16.to_le_bytes(), b"HOLA"].concat(), "service 2 in the directory's language, its number a private-use subtag"); ends_where_it_left(&hola, b"HOLA", b"es-x-svc2", 400, "shown from frame 12 to frame 22"); assert_eq!(&payload(hola[4])[5..37], &joule_moq_ingest::room::sha256(&[&hola[0][..hola[0].len() - 64]])[..], "the window's final supersedes its partial"); // The windows: CTA-708 service 2 and service 1, window 0 of each, and CTA-608 data // channel 1, after both revisions of each caption. for (c, w) in [(&hola, [2u8, 2, 0]), (&uno, [2, 1, 0]), (&anchored, [1, 1, 0])] { assert_eq!((payload(c[2]), payload(c[5])), ([&b"WTW1\x01"[..], &w].concat(), [&b"WTW1\x01"[..], &w].concat()), "a caption's window, timed-text §4.7"); } let (partial, final_) = (payload(anchored[0]), payload(anchored[3])); // caption: duration_ms, origin 1 (no model), "en", the text. assert_eq!(partial, [&0u32.to_le_bytes()[..], &[1, 2], b"en", &5u16.to_le_bytes(), b"HELLO"].concat()); assert_eq!(final_[4..], [&[1, 2][..], b"en", &5u16.to_le_bytes(), b"HELLO"].concat()[..]); ends_where_it_left(&anchored, b"HELLO", b"en", 440, "shown from frame 19 to frame 30"); // The annotations: partial, then final superseding the partial's caption event. let (a0, a1) = (payload(anchored[1]), payload(anchored[4])); assert_eq!((&a0[..5], a0[37], &a1[..5], a1[37]), (&b"WTX1"[..], 0, &b"WTX1"[..], 1)); assert_eq!(&a1[5..37], &joule_moq_ingest::room::sha256(&[&anchored[0][..anchored[0].len() - 64]])[..], "the final supersedes the partial"); // The origin claims: one a track, signed by the gateway, of step capture, each naming // its stream's statement. let claims = d["origin_claims"].as_array().unwrap(); assert_eq!(claims.len(), 2); for c in claims { let mut body = c.clone(); let sig = body.as_object_mut().unwrap().remove("sig").unwrap(); let sig = ed25519_dalek::Signature::from_slice(&hex::decode(sig.as_str().unwrap()).unwrap()).unwrap(); // Over RFC 8785 of the claim without `sig`: its members in key order, no whitespace. let jcs = serde_json::to_string(&joule_moq::json_order::sorted(&body)).unwrap(); w.gateway_key.verifying_key().verify(&[b"wai:synthesis\x01".as_slice(), jcs.as_bytes()].concat(), &sig).expect("the claim verifies under the gateway's key"); assert_eq!((c["class"].as_str(), c["step"].as_str(), c["source"].as_str()), (Some("captured"), Some("capture"), Some("digitalCapture"))); let i = if c["media"] == "video" { 0 } else { 1 }; assert_eq!(c["record"], d["streams"][i]["statement_digest"]); assert_eq!(hex::encode(session::statement_digest(&d["streams"][i]["statement"])), d["streams"][i]["statement_digest"].as_str().unwrap()); let kind = if i == 0 { media::VIDEO } else { media::AUDIO }; assert_eq!(c["subject"]["track"].as_str().unwrap(), hex::encode(catalog::track_digest(&catalog::media_namespace(&w.room, &w.owner(), kind), "0"))); } let first_group = stamps.lock().unwrap().iter().map(|s| s.0).min().unwrap(); assert_eq!(claims.iter().find(|c| c["media"] == "video").unwrap()["subject"]["from_group"], first_group); // The session's resource is gone. assert_eq!(auth.send("DELETE", &format!("{}/st2110/s/{id}", w.base()), &[], b"").await.status, 404); w.stop().await; } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn a_description_that_is_not_carried_is_refused_with_its_reason() { timeout(Duration::from_secs(120), refusals()).await.expect("in time"); } async fn refusals() { let w = world(|_| {}).await; let sdp_of = |m: &[Media], dest: &str, clocks: Clocks| description(m, dest, &[30_000, 30_002, 30_004][..m.len()], clocks, None); let base = video_format(Sampling::YCbCr422, 10, 128, 72, (25, 1)); let good = sdp_of(&[Media::Video(base.clone())], "127.0.0.1", Clocks::Local); let reason = |r: &Resp| serde_json::from_slice::(&r.body).ok().and_then(|v| v["reason"].as_str().map(str::to_string)).unwrap_or_else(|| r.text()); let cases: Vec<(&str, Value, u16, &str)> = vec![ ("no description", json!({}), 400, "body_malformed"), ("a source that is no capture's", json!({"sdp": good, "source": "trainedAlgorithmicMedia"}), 400, "body_malformed"), ("interlaced video", json!({"sdp": good.replace("PM=2110GPM", "PM=2110GPM; interlace")}), 422, "sdp_scan_unsupported"), ("no depth", json!({"sdp": good.replace(" depth=10;", "")}), 422, "sdp_fmtp_missing"), ("RGB", json!({"sdp": sdp_of(&[Media::Video(video_format(Sampling::Rgb, 10, 128, 72, (25, 1)))], "127.0.0.1", Clocks::Local)}), 422, "essence_format_not_carried"), ("12 bits", json!({"sdp": sdp_of(&[Media::Video(video_format(Sampling::YCbCr422, 12, 128, 72, (25, 1)))], "127.0.0.1", Clocks::Local)}), 422, "essence_format_not_carried"), ("2160 lines at 60 a second", json!({"sdp": sdp_of(&[Media::Video(video_format(Sampling::YCbCr422, 10, 3840, 2160, (60, 1)))], "127.0.0.1", Clocks::Local)}), 422, "essence_outside_floor"), ("nine channels", json!({"sdp": sdp_of(&[Media::Audio(AudioFormat { bits: 24, rate: 48_000, channels: 9, ptime_us: Some(125), channel_order: None })], "127.0.0.1", Clocks::Local)}), 422, "essence_format_not_carried"), ("two video streams", json!({"sdp": sdp_of(&[Media::Video(base.clone()), Media::Video(base.clone())], "127.0.0.1", Clocks::Local)}), 422, "essence_streams_unsupported"), ("a multicast group, where the gateway joins none", json!({"sdp": sdp_of(&[Media::Video(base.clone())], "239.255.21.10", Clocks::Local)}), 403, "multicast_not_permitted"), ]; for (i, (what, body, status, why)) in cases.into_iter().enumerate() { // A grant opens one session's request; each case takes its own. let who = SigningKey::from_bytes(&[0x80 + i as u8; 32]); let r = open(&w, &w.auth(&who, 0), &body).await; assert_eq!((r.status, reason(&r).as_str()), (status, why), "{what}: {}", r.text()); } // A stream whose clock is not traceable is carried, and says so; a DELETE ends it, and // its record states the instants as arrivals. let auth = w.auth(&w.publisher, 0); let r = open(&w, &auth, &json!({"sdp": good})).await; assert_eq!(r.status, 201, "{}", r.text()); let s: Value = serde_json::from_slice(&r.body).unwrap(); assert_eq!((s["streams"][0]["clock"].as_str(), s["streams"][0]["tsmode"].as_str(), s["origin_claim"].as_bool()), (Some("not_traceable"), Some("NEW"), Some(false))); let id = s["session"].as_str().unwrap(); assert!(contrib_live(&w, id).await); // The same principal again, while its session runs. let again = open(&w, &w.auth(&w.publisher, 0), &json!({"sdp": good})).await; assert_eq!((again.status, reason(&again).as_str()), (409, "session_conflict")); // One frame, from another address than the first packet's after it: only the first source's is read. let port = s["streams"][0]["port"].as_u64().unwrap() as u16; let (a, b) = (UdpSocket::bind("127.0.0.1:0").await.unwrap(), UdpSocket::bind("127.0.0.1:0").await.unwrap()); let mut out = Out::new(96, 7, 0); let packets = video_packets(&base, &send::picture(&base, 0), 1_000, &mut out, 1_440); for p in &packets { a.send_to(p, ("127.0.0.1", port)).await.unwrap(); } tokio::time::sleep(Duration::from_millis(100)).await; b.send_to(&packets[0], ("127.0.0.1", port)).await.unwrap(); tokio::time::sleep(Duration::from_millis(300)).await; let loc = format!("{}/st2110/s/{id}", w.base()); assert_eq!(auth.send("DELETE", &loc, &[], b"").await.status, 200); assert!(until(Duration::from_secs(60), || w.gateway.gateway.records().iter().any(|r| r.record.protocol == "st2110")).await); let rec = w.gateway.gateway.records().into_iter().find(|r| r.record.protocol == "st2110").unwrap(); let sv = &rec.record.detail["streams"][0]; assert_eq!(rec.record.ended, "deleted"); assert_eq!((sv["statement"]["clock"].as_str(), sv["from_other_sources"].as_u64(), sv["read"]["frames"].as_u64(), sv["capture_instants"]["from_rtp_timestamp"].as_u64(), sv["capture_instants"]["from_arrival"].as_u64()), (Some("not_traceable"), Some(1), Some(1), Some(0), Some(1))); assert_eq!(rec.record.detail["origin_claims"], json!([]), "no source was stated, so no claim is signed"); w.stop().await; // A gateway that does not take essence. let w = World::start(|_| {}).await; let r = open(&w, &w.auth(&w.publisher, 0), &json!({"sdp": good})).await; assert_eq!(r.status, 404); w.stop().await; } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn a_multicast_group_is_joined_on_loopback() { timeout(Duration::from_secs(60), multicast()).await.expect("in time"); } /// As far as a loopback group allows: with the configuration's line, the description's group /// is joined on the loopback interface at the description's own port, and the session is /// opened; whether a datagram sent to the group comes back on this host's loopback is the /// host's routing, which this test does not hold. async fn multicast() { let w = world(|c| c.st2110.multicast = true).await; let port = low_udp(); let f = video_format(Sampling::YCbCr422, 10, 128, 72, (25, 1)); let text = description(&[Media::Video(f)], "239.255.21.10", &[port], Clocks::Local, None); let auth = w.auth(&w.publisher, 0); let r = open(&w, &auth, &json!({"sdp": text})).await; assert_eq!(r.status, 201, "{}", r.text()); let s: Value = serde_json::from_slice(&r.body).unwrap(); assert_eq!(s["streams"][0]["port"], port, "a group's stream is received at the description's port"); let loc = format!("{}/st2110/s/{}", w.base(), s["session"].as_str().unwrap()); assert!(contrib_live(&w, s["session"].as_str().unwrap()).await); assert_eq!(auth.send("DELETE", &loc, &[], b"").await.status, 200); w.stop().await; } /// An encoder asked for by name that is not on the host: the gateway does not start, and /// says why; with `encoder_fallback` it starts on the next one, and the session's answer and /// its record say which encoder was asked for, which encodes, and why. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn an_encoder_that_is_not_here_is_a_refusal_or_a_stated_fallback() { timeout(Duration::from_secs(180), absent_encoder()).await.expect("in time"); } async fn absent_encoder() { // A hardware encoder through a device that does not exist, and a software encoder // through a library that does not: neither is on any host. fn absent(c: &mut joule_moq_ingest::GatewayConfig, name: &str) { c.st2110.enabled = true; c.st2110.encoder = name.into(); c.st2110.va_device = "/dev/dri/no-such-render-node".into(); c.st2110.svt_av1_library = "/no/such/libSvtAv1Enc.so".into(); } for name in ["va", "svt-av1"] { let relay = joule_moq_relay::RelayServer::start(joule_moq_relay::RelayConfig::loopback()).await.unwrap(); let mut gc = joule_moq_ingest::GatewayConfig::loopback(relay.quic_addr.unwrap()); absent(&mut gc, name); let refused = joule_moq_ingest::GatewayServer::start(gc).await.err().expect("the gateway does not start on an encoder that is not here"); assert!(refused.contains(&format!("encoder = \"{name}\" is not here")) && refused.contains("Nothing is used in its place"), "{refused}"); relay.shutdown().await; let w = world(|c| { absent(c, name); c.st2110.encoder_fallback = true; }) .await; let auth = w.auth(&w.publisher, 0); let f = video_format(Sampling::YCbCr422, 10, 128, 72, (25, 1)); let text = description(&[Media::Video(f.clone())], "127.0.0.1", &[30_000], Clocks::Local, None); let r = open(&w, &auth, &json!({"sdp": text})).await; assert_eq!(r.status, 201, "{}", r.text()); let s: Value = serde_json::from_slice(&r.body).unwrap(); let said = |e: &Value| { assert_eq!((e["asked"].as_str(), e["hardware"].as_bool()), (Some(name), Some(false)), "{e}"); assert_ne!(e["backend"].as_str(), Some(name), "it is not used in the name of the encoder asked for"); assert!(e["fallback"].as_str().is_some_and(|x| x.starts_with(&format!("{name} was asked for and is not here: "))), "{e}"); }; said(&s["encoder"]); let id = s["session"].as_str().unwrap(); assert!(contrib_live(&w, id).await); let sock = UdpSocket::bind("127.0.0.1:0").await.unwrap(); let mut out = Out::new(96, 7, 0); for n in 0..3u32 { for p in video_packets(&f, &send::picture(&f, n), 1_000 + n * 3_600, &mut out, 1_440) { sock.send_to(&p, ("127.0.0.1", s["streams"][0]["port"].as_u64().unwrap() as u16)).await.unwrap(); } tokio::time::sleep(Duration::from_millis(40)).await; } tokio::time::sleep(Duration::from_millis(300)).await; assert_eq!(auth.send("DELETE", &format!("{}/st2110/s/{id}", w.base()), &[], b"").await.status, 200); assert!(until(Duration::from_secs(90), || w.gateway.gateway.records().iter().any(|r| r.record.protocol == "st2110")).await); let rec = w.gateway.gateway.records().into_iter().find(|r| r.record.protocol == "st2110").unwrap(); rec.verify().unwrap(); said(&rec.record.detail["encoder"]); let stage = &rec.record.detail["encoder"]["stage"]; assert_eq!(stage["frames_offered"], 3); assert_eq!(stage["frames_encoded"].as_u64().unwrap() + stage["frames_not_encoded_queue_full"].as_u64().unwrap(), 3, "{stage}"); w.stop().await; } } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn a_sender_that_bursts_whole_frames_is_read_outside_a_narrow_senders_model() { timeout(Duration::from_secs(120), narrow()).await.expect("in time"); } /// Planted: a sender that sends each whole frame at once, against a description that says /// its sender is narrow (`TP=2110TPN`, ST 2110-21 §7.1.2). The session reads its arrivals /// against that type's Network Compatibility Model and its record counts every frame read /// outside it. The control: the same frames on the linear schedule, four packets at an /// instant (`st2110::pace`), whose bucket held a small part of what the burst's did. Both /// runs' frames are read whole: the model is read, not enforced. async fn narrow() { let w = world(|_| {}).await; // A small picture in small payloads: about fifty packets a frame, and an encoder that // finishes its frames in a moment on any host. let f = video_format(Sampling::YCbCr422, 10, 128, 72, (25, 1)); let text = description(&[Media::Video(f.clone())], "127.0.0.1", &[30_000], Clocks::Local, None).replace("TP=2110TPW", "TP=2110TPN"); let mut cinst = Vec::new(); for burst in [0usize, 4] { let auth = w.auth(&w.publisher, 0); let r = open(&w, &auth, &json!({"sdp": text})).await; assert_eq!(r.status, 201, "{}", r.text()); let s: Value = serde_json::from_slice(&r.body).unwrap(); let id = s["session"].as_str().unwrap().to_string(); assert!(contrib_live(&w, &id).await); let sock = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); sock.connect(("127.0.0.1", s["streams"][0]["port"].as_u64().unwrap() as u16)).unwrap(); let f2 = f.clone(); let sent = tokio::task::spawn_blocking(move || { let mut out = Out::new(96, 0x6e61_0001, 0); joule_moq_ingest::st2110::pace::Pacer::new(sock, burst) .run(12, (25, 1), std::time::Instant::now() + Duration::from_millis(100), |n, data, lens| { for p in video_packets(&f2, &send::picture(&f2, n), 1_000 + n * 3_600, &mut out, 500) { data.extend_from_slice(&p); lens.push(p.len()); } }) .unwrap() }) .await .unwrap(); tokio::time::sleep(Duration::from_millis(300)).await; assert_eq!(auth.send("DELETE", &format!("{}/st2110/s/{id}", w.base()), &[], b"").await.status, 200); assert!(until(Duration::from_secs(60), || w.gateway.gateway.records().iter().any(|r| r.record.session == id)).await); let rec = w.gateway.gateway.records().into_iter().find(|r| r.record.session == id).unwrap(); rec.verify().unwrap(); let sv = &rec.record.detail["streams"][0]; let at = &sv["arrival_timing"]; println!("burst {burst}: sent {sent:?}; arrivals {at}; receive {}", sv["receive"]); assert_eq!((sv["read"]["frames"].as_u64(), sv["read"]["frames_incomplete"].as_u64(), sv["lost"].as_u64()), (Some(12), Some(0), Some(0)), "every frame read whole: {sv}"); assert_eq!((at["sender_type"].as_str(), at["cmax"].as_u64(), at["frames_read"].as_u64()), (Some("2110TPN"), Some(4), Some(11)), "{at}"); assert_eq!(sv["receive"]["datagrams"].as_u64(), Some(sent.packets), "the receive thread read every datagram"); if burst == 0 { assert_eq!(at["frames_outside"].as_u64(), Some(11), "every frame of a whole-frame burst is outside a narrow sender's model: {at}"); } cinst.push(at["cinst_max"].as_u64().unwrap()); } // A frame is about fifty packets: the burst's bucket held most of one. assert!(cinst[0] >= 30 && cinst[1] * 3 <= cinst[0], "C_INST at most: {cinst:?}"); w.stop().await; }