c526021c67
contrib_srt's an_srt_publisher_reaches_a_relay_subscriber_with_every_receipt failed once in CI
(97de4c1dc, the aarch64 leg) with "the path dropped datagrams". That assertion requires the
lossy proxy to have dropped more than ten datagrams toward the gateway, and it ran as soon as
the publisher had queued its last payload. The proxy's seed (0xabc) drops five datagrams at
random among its first hundred, then the burst at indices 100 to 105, so the count passes ten
exactly when the proxy has handled its 106th datagram. Instrumented runs on a loaded aarch64
Linux host show the proxy at 177 to 193 datagrams at that moment; the failing runner had not
reached 106: the check raced the proxy's task, and nothing had been lost. The check now runs
after the gateway has acknowledged every packet, when every one of the publisher's 108 data
packets has crossed the proxy.
The same runs show the host dropping loopback datagrams at full receive queues under load (the
namespace's Udp RcvbufErrors rose by 9 to 45 a round with eight copies of the test beside each
other), which the link recovers but the exact accounting of recovered first sends would count,
and which made the twenty-megabit test drop packets as too late. joule_contrib::sock asks for
an 8 MiB receive buffer (halving the ask where the host refuses it, the host's limit bounding
it), used by the SRT listener and caller and by both proxy sockets.
Held: the contribution tests pass repeated with copies of themselves beside them (6 and 9 runs
on a loaded aarch64 host, 40 before the change on an aarch64 Linux host showed only the
twenty-megabit failures at eight copies); a proxy losing every copy of one data packet fails
the test ("every access unit, byte for byte"), and a path dropping nothing fails it ("the path
dropped datagrams"). clippy -D warnings and the tests of joule-contrib and joule-moq-ingest
pass on aarch64.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
361 lines
21 KiB
Rust
361 lines
21 KiB
Rust
//! SRT into a room's tracks, end to end on loopback: a publisher POSTs `/srt` with its grant
|
|
//! and gets a one-time session; it calls the gateway's listener with the stream ID and the
|
|
//! passphrase, through a path that drops datagrams each way, and sends a transport stream
|
|
//! of H.264 and AAC. A subscriber at the relay receives every group of both tracks, verifies
|
|
//! every receipt under the gateway's key, and gets every access unit and every ADTS frame
|
|
//! as sent. The link record's parts verify as a series, account for every sequence number,
|
|
//! show the losses recovered, and name the head of each track's receipt chain, which is the
|
|
//! last group receipt the subscriber verified; they are on the owner's `link` track too.
|
|
//! Then the refusals, the gateway calling a publisher's listener, and HEVC.
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use joule_contrib::lossy::{Counts, Impair, Proxy};
|
|
use joule_contrib::srt::{self, conn::CallError, packet::{rej, Packet}, Admitted, CallConfig, ConnRequest, ListenerConfig, Role};
|
|
use joule_contrib::ts;
|
|
use joule_moq::carriage::ReceiptBundle;
|
|
use joule_moq::receipt::ReceiptVerifier;
|
|
use joule_moq::{GroupId, ObjectId};
|
|
use joule_moq_ingest::catalog;
|
|
use joule_moq_ingest::frame::Codec;
|
|
use joule_moq_ingest::linkrec::{self, ended, LinkRecord};
|
|
use joule_moq_ingest::room::media;
|
|
use std::collections::BTreeSet;
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
use tokio::time::timeout;
|
|
|
|
/// Every object of `t` verified against its receipts under `key`, as its raw payload, in
|
|
/// group order: one object a group.
|
|
fn verified_raw(wt: &Watcher, t: &Watched, key: &ed25519_dalek::SigningKey) -> Vec<Vec<u8>> {
|
|
let v = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
|
|
let objs = wt.objects.lock().unwrap();
|
|
let mut out = Vec::new();
|
|
let mut prev = None;
|
|
for ((alias, g, o), p) in objs.iter() {
|
|
if *alias != t.media_alias {
|
|
continue;
|
|
}
|
|
let b = objs.iter().find(|(k, _)| k.0 == t.receipts_alias && k.1 == *g).map(|(_, b)| ReceiptBundle::from_payload(b).unwrap()).expect("each group's receipts arrived");
|
|
let vg = v.verify_group(&b.group).expect("the group receipt verifies under the gateway's key");
|
|
if let Some(link) = prev {
|
|
assert_eq!(b.group.parent_receipt_hash, Some(link), "each group chains to the one before");
|
|
}
|
|
prev = Some(b.group.link_hash());
|
|
let r = &b.objects[*o as usize];
|
|
r.check_moqt(b.group.track_alias, GroupId(*g), ObjectId(*o)).unwrap();
|
|
v.verify_payload(p, r, &vg).expect("the object verifies against its receipt");
|
|
out.push(p.clone());
|
|
}
|
|
out
|
|
}
|
|
|
|
async fn open_srt(w: &World, auth: &Auth) -> serde_json::Value {
|
|
let r = auth.send("POST", &format!("{}/srt", w.base()), &[("content-type", "application/json")], b"").await;
|
|
assert_eq!(r.status, 201, "POST /srt: {}", r.text());
|
|
assert!(r.header("location").unwrap().starts_with("/srt/s/"));
|
|
serde_json::from_slice(&r.body).unwrap()
|
|
}
|
|
|
|
fn call_cfg(s: &serde_json::Value) -> CallConfig {
|
|
CallConfig {
|
|
stream_id: s["stream_id"].as_str().map(str::to_string),
|
|
passphrase: s["passphrase"].as_str().map(str::to_string),
|
|
keylen: s["pbkeylen"].as_u64().unwrap() as usize,
|
|
latency_ms: 300,
|
|
..CallConfig::default()
|
|
}
|
|
}
|
|
|
|
/// The link records of the gateway, read as one series.
|
|
fn parts(w: &World) -> Vec<LinkRecord> {
|
|
let recs = w.gateway.gateway.link_records();
|
|
linkrec::verify_series(&recs).unwrap_or_else(|e| panic!("the link record's parts verify as a series: {e:?}"))
|
|
}
|
|
|
|
|
|
/// The sequence numbers whose first send the proxy dropped on its way to the receiver:
|
|
/// each must be found missing and recovered. Control packets and retransmissions it
|
|
/// dropped are not among them.
|
|
fn first_sends_lost(c: &Counts) -> BTreeSet<u32> {
|
|
c.dropped_log.lock().unwrap().iter().filter_map(|b| match Packet::read(b) {
|
|
Some(Packet::Data(d)) if !d.rexmit => Some(d.seq),
|
|
_ => None,
|
|
}).collect()
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn an_srt_publisher_reaches_a_relay_subscriber_with_every_receipt() {
|
|
timeout(Duration::from_secs(120), run()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
async fn run() {
|
|
let w = World::start(|c| {
|
|
c.contrib.srt.listen = Some("127.0.0.1:0".parse().unwrap());
|
|
c.contrib.record_interval_ms = 1_000;
|
|
c.contrib.srt.latency_ms = 250;
|
|
})
|
|
.await;
|
|
let auth = w.auth(&w.publisher, 0);
|
|
let s = open_srt(&w, &auth).await;
|
|
let sid = s["stream_id"].as_str().unwrap();
|
|
assert!(sid.starts_with(&format!("#!::r={}/{},m=publish,s=", hex::encode(w.room), w.owner())), "{sid}");
|
|
// 3% of the datagrams dropped each way, and a burst of six toward the gateway, mid-stream.
|
|
// The publisher closes after about 211 datagrams and sends its three SHUTDOWNs back to
|
|
// back; a burst at 200 swallowed all three when a host sent a few fewer before the close,
|
|
// and the gateway ended the session by its idle timeout.
|
|
let up = Impair { loss_ppm: 30_000, drop_index: &[100, 101, 102, 103, 104, 105], ..Impair::default() };
|
|
let px = Proxy::start(w.gateway.srt_addr.unwrap(), up, Impair { loss_ppm: 30_000, ..Impair::default() }, 0xabc).await.unwrap();
|
|
let link = timeout(STEP, srt::call(px.addr, call_cfg(&s))).await.unwrap().expect("the call is admitted");
|
|
assert_eq!(link.info.cipher_bits, 128);
|
|
// The tracks are published once the call is admitted; a subscriber joins before the
|
|
// first frame.
|
|
assert!(contrib_live(&w, s["session"].as_str().unwrap()).await, "the session publishes its tracks");
|
|
let watcher = watch(&w).await;
|
|
// A watcher of the owner's link track and its receipts.
|
|
let links = watch_tracks(&w, vec![(0, catalog::catalog_namespace(&w.room, &w.owner()), "link".into())]).await;
|
|
let sent = send_ts(&link, 3, 30, ts::ST_H264).await;
|
|
tokio::time::sleep(Duration::from_millis(800)).await;
|
|
// Close once the gateway has acknowledged every packet: a fixed wait alone left a loss
|
|
// unrecovered on a loaded host, and the accounting below counted it outstanding.
|
|
assert!(until(STEP, || link.sent.lock().unwrap().unacknowledged == 0).await, "the gateway acknowledged every packet before the publisher closed");
|
|
// The path's losses, read once every packet has crossed it: each was acknowledged, so the
|
|
// proxy has handled each of the publisher's 108 data packets, and its seed's drops (five at
|
|
// random in its first hundred datagrams, then the burst of six) are behind it. Read as soon
|
|
// as the publisher had queued its last payload, the count raced the proxy's task: a loaded
|
|
// runner read it before the proxy had handled its 106th datagram.
|
|
assert!(px.up.dropped.load(std::sync::atomic::Ordering::Relaxed) > 10, "the path dropped datagrams");
|
|
link.close();
|
|
assert!(until(STEP, || watcher.ended.lock().unwrap().len() == 2).await, "PUBLISH_DONE on both media tracks");
|
|
settle(&watcher, Duration::from_millis(500)).await;
|
|
|
|
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 (vf, vg) = verified(&watcher, video, &w.gateway_key, Codec::Avc);
|
|
let (af, _) = verified(&watcher, audio, &w.gateway_key, Codec::Aac);
|
|
same(&vf, &sent.video, "every access unit, byte for byte");
|
|
same(&af, &sent.audio, "every ADTS frame, byte for byte");
|
|
assert_eq!(vg, 3, "one video group per keyframe");
|
|
assert_eq!(w.relay.relay().stats().refused_at_door, 0, "the relay's door verified every object");
|
|
let env = envelopes(&watcher, video);
|
|
assert_eq!(env[0].manifest["model_requirement"]["capability"], "wai.video.avc");
|
|
assert_eq!(env[0].manifest["live"]["rate"], 90_000);
|
|
assert_eq!(envelopes(&watcher, audio)[0].manifest["model_requirement"]["capability"], "wai.audio.aac");
|
|
|
|
// The link record.
|
|
assert!(until(STEP, || w.gateway.gateway.link_records().iter().any(|b| LinkRecord::read(b).is_ok_and(|r| r.last))).await, "the last part is signed");
|
|
let ps = parts(&w);
|
|
assert!(ps.len() >= 3, "a part a second, and the last: {}", ps.len());
|
|
let last = ps.last().unwrap();
|
|
assert!(last.last && last.ended == ended::PEER_SHUTDOWN, "the last part ends the session as the publisher ended it: last={} ended={}", last.last, last.ended);
|
|
let (mut delivered, mut first, mut dropped, mut count) = (0, 0, 0, 0);
|
|
for p in &ps {
|
|
assert_eq!((p.protocol, p.mode, p.cipher), (1, 1, 128));
|
|
assert_eq!(p.window, 300_000, "the larger of the two ends' latencies");
|
|
delivered += p.account.delivered;
|
|
first += p.account.first_try;
|
|
dropped += p.account.dropped;
|
|
count += p.account.seq_count;
|
|
}
|
|
assert_eq!(dropped, 0, "every loss recovered within the window");
|
|
// The burst of six dropped datagrams holds as many first sends of data as the timing of the
|
|
// control packets between them leaves: count them, rather than assume all six.
|
|
let lost = first_sends_lost(&px.up);
|
|
assert!(lost.len() >= 2, "the path lost first sends of data: {lost:?}");
|
|
assert_eq!(delivered - first, lost.len() as u64, "each first send lost was found and recovered");
|
|
assert_eq!(count, delivered + last.account.outstanding);
|
|
let counts = link.sent.lock().unwrap().clone();
|
|
assert_eq!((delivered, counts.expired), (counts.sent, 0), "every payload the publisher sent, accounted for, none given up: {counts:?}");
|
|
// Each track's head is the link hash of the last group receipt the subscriber verified.
|
|
for t in [video, audio] {
|
|
let objs = watcher.objects.lock().unwrap();
|
|
let bundles: Vec<ReceiptBundle> = objs.iter().filter(|(k, _)| k.0 == t.receipts_alias).map(|(_, p)| ReceiptBundle::from_payload(p).unwrap()).collect();
|
|
let ns = catalog::media_namespace(&w.room, &w.owner(), t.kind);
|
|
let head = last.tracks.iter().find(|h| h.track == catalog::track_digest(&ns, "0")).expect("the record names the track");
|
|
assert_eq!(head.groups, bundles.len() as u64);
|
|
assert_eq!(head.head, bundles.last().unwrap().group.link_hash(), "the record names the head of the track's receipt chain");
|
|
}
|
|
// Every part is on the owner's link track, each object receipted by the gateway.
|
|
settle(&links, Duration::from_millis(300)).await;
|
|
let lt = &links.tracks[0];
|
|
let got = verified_raw(&links, lt, &w.gateway_key);
|
|
assert_eq!(got, w.gateway.gateway.link_records(), "the link track carries every part, as signed");
|
|
|
|
// The record's session record names the protocol.
|
|
assert!(until(STEP, || w.gateway.gateway.records().iter().any(|r| r.record.protocol == "srt")).await);
|
|
let rec = w.gateway.gateway.records().into_iter().find(|r| r.record.protocol == "srt").unwrap();
|
|
rec.verify().unwrap();
|
|
assert_eq!(rec.record.ended, "peer_shutdown");
|
|
|
|
// The session's resource is gone.
|
|
let loc = format!("{}/srt/s/{}", w.base(), s["session"].as_str().unwrap());
|
|
assert_eq!(auth.send("DELETE", &loc, &[], b"").await.status, 404);
|
|
px.stop();
|
|
w.stop().await;
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn the_listener_refuses_with_the_mappings_codes() {
|
|
timeout(Duration::from_secs(120), refusals()).await.expect("in time");
|
|
}
|
|
|
|
async fn refusals() {
|
|
let w = World::start(|c| c.contrib.srt.listen = Some("127.0.0.1:0".parse().unwrap())).await;
|
|
let addr = w.gateway.srt_addr.unwrap();
|
|
let auth = w.auth(&w.publisher, 0);
|
|
let s = open_srt(&w, &auth).await;
|
|
let call = |cfg: CallConfig| srt::call(addr, CallConfig { connect_timeout_ms: 3_000, ..cfg });
|
|
// No stream ID; one without a session; a session the gateway never gave.
|
|
assert_eq!(call(CallConfig { passphrase: s["passphrase"].as_str().map(str::to_string), ..CallConfig::default() }).await.err(), Some(CallError::Rejected(rej::UNAUTHORIZED)));
|
|
let fake = format!("#!::r={}/{},m=publish,s={}", hex::encode(w.room), w.owner(), "00".repeat(16));
|
|
assert_eq!(call(CallConfig { stream_id: Some(fake), ..call_cfg(&s) }).await.err(), Some(CallError::Rejected(rej::UNAUTHORIZED)));
|
|
// Play mode, which this gateway does not serve.
|
|
let req = s["stream_id"].as_str().unwrap().replace("m=publish", "m=request");
|
|
assert_eq!(call(CallConfig { stream_id: Some(req), ..call_cfg(&s) }).await.err(), Some(CallError::Rejected(rej::BAD_MODE)));
|
|
// Another room; another owner.
|
|
let other_room = s["stream_id"].as_str().unwrap().replace(&hex::encode(w.room), &"ab".repeat(32));
|
|
assert_eq!(call(CallConfig { stream_id: Some(other_room), ..call_cfg(&s) }).await.err(), Some(CallError::Rejected(rej::NOTFOUND)));
|
|
let other_owner = s["stream_id"].as_str().unwrap().replace(&w.owner(), &"cd".repeat(32));
|
|
assert_eq!(call(CallConfig { stream_id: Some(other_owner), ..call_cfg(&s) }).await.err(), Some(CallError::Rejected(rej::FORBIDDEN)));
|
|
// The right session and a wrong passphrase: refused, and the session is spent.
|
|
assert_eq!(call(CallConfig { passphrase: Some("not the session's passphrase".into()), ..call_cfg(&s) }).await.err(), Some(CallError::Rejected(rej::BADSECRET)));
|
|
assert_eq!(call(call_cfg(&s)).await.err(), Some(CallError::Rejected(rej::UNAUTHORIZED)), "a session ID is good for one use");
|
|
// A new session, used once.
|
|
let s2 = open_srt(&w, &auth).await;
|
|
let l = call(call_cfg(&s2)).await.expect("admitted");
|
|
assert_eq!(call(call_cfg(&s2)).await.err(), Some(CallError::Rejected(rej::UNAUTHORIZED)));
|
|
// While its owner publishes, another session of the same owner is refused: 1409.
|
|
let s3 = open_srt(&w, &auth).await;
|
|
assert!(until(STEP, || w.gateway.gateway.contrib.active.load(std::sync::atomic::Ordering::SeqCst) == 1).await);
|
|
assert_eq!(call(call_cfg(&s3)).await.err(), Some(CallError::Rejected(rej::CONFLICT)));
|
|
l.close();
|
|
// A viewer's grant cannot open a contribution session.
|
|
let viewer = w.auth(&w.publisher, 1);
|
|
let r = viewer.send("POST", &format!("{}/srt", w.base()), &[], b"").await;
|
|
assert_eq!(r.status, 403, "{}", r.text());
|
|
// DELETE of a session waiting to connect cancels it.
|
|
let s4 = open_srt(&w, &auth).await;
|
|
let loc = format!("{}/srt/s/{}", w.base(), s4["session"].as_str().unwrap());
|
|
assert_eq!(auth.send("DELETE", &loc, &[], b"").await.status, 200);
|
|
assert_eq!(call(call_cfg(&s4)).await.err(), Some(CallError::Rejected(rej::UNAUTHORIZED)));
|
|
w.stop().await;
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn the_gateway_calls_a_publishers_listener() {
|
|
timeout(Duration::from_secs(120), calls()).await.expect("in time");
|
|
}
|
|
|
|
async fn calls() {
|
|
let w = World::start(|c| c.contrib.srt.listen = Some("127.0.0.1:0".parse().unwrap())).await;
|
|
// The publisher's listener: it sends to whoever calls with its passphrase.
|
|
let adm = |_: &ConnRequest| Ok(Admitted { passphrase: Some("the encoder's passphrase".into()), role: Role::Send, latency_ms: Some(200), tag: 0 });
|
|
let mut enc = srt::Listener::bind(ListenerConfig::default(), Arc::new(adm)).await.unwrap();
|
|
let auth = w.auth(&w.publisher, 0);
|
|
let body = serde_json::json!({"call": enc.addr.to_string(), "passphrase": "the encoder's passphrase"}).to_string();
|
|
// Not in call_allow: refused.
|
|
let r = auth.send("POST", &format!("{}/srt", w.base()), &[("content-type", "application/json")], body.as_bytes()).await;
|
|
assert_eq!(r.status, 403, "{}", r.text());
|
|
w.stop().await;
|
|
|
|
let w = World::start(|c| {
|
|
c.contrib.srt.listen = Some("127.0.0.1:0".parse().unwrap());
|
|
c.contrib.srt.call_allow = vec!["127.0.0.1".into()];
|
|
})
|
|
.await;
|
|
let auth = w.auth(&w.publisher, 0);
|
|
let r = auth.send("POST", &format!("{}/srt", w.base()), &[("content-type", "application/json")], body.as_bytes()).await;
|
|
assert_eq!(r.status, 201, "{}", r.text());
|
|
let id: serde_json::Value = serde_json::from_slice(&r.body).unwrap();
|
|
let link = timeout(STEP, enc.accept()).await.unwrap().unwrap();
|
|
assert!(contrib_live(&w, id["session"].as_str().unwrap()).await);
|
|
let watcher = watch(&w).await;
|
|
let sent = send_ts(&link, 2, 30, ts::ST_HEVC).await;
|
|
tokio::time::sleep(Duration::from_millis(700)).await;
|
|
link.close();
|
|
assert!(until(STEP, || watcher.ended.lock().unwrap().len() == 2).await);
|
|
settle(&watcher, Duration::from_millis(500)).await;
|
|
let video = watcher.tracks.iter().find(|t| t.kind == media::VIDEO).unwrap();
|
|
let (vf, vg) = verified(&watcher, video, &w.gateway_key, Codec::Hevc);
|
|
same(&vf, &sent.video, "every HEVC access unit, byte for byte");
|
|
assert_eq!(vg, 2, "one group per IRAP");
|
|
let ps = parts(&w);
|
|
assert!(ps.iter().all(|p| p.mode == 2 && p.cipher == 128), "the gateway called");
|
|
w.stop().await;
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn contribution_at_twenty_megabits_passes_the_door() {
|
|
timeout(Duration::from_secs(120), big_groups()).await.expect("in time");
|
|
}
|
|
|
|
/// Contribution at 20 Mbit/s for four seconds, a keyframe each second: video groups of about
|
|
/// 2.6 MB beside the audio. The relay's door holds each object until its group's receipts
|
|
/// follow the group's last object, so it reads each group whole off its stream before it
|
|
/// releases any of it, which the relay's stream window (`limits.quic_stream_window_bytes`)
|
|
/// must hold: under QUIC's default of about 1.25 MB the stream stops and the door refuses.
|
|
/// The link's window is two seconds: the door is the subject, and on a loaded runner a stall
|
|
/// past 300 ms had the gateway drop access units as too late (§4.6) before they reached it.
|
|
async fn big_groups() {
|
|
let w = World::start(|c| {
|
|
c.contrib.srt.listen = Some("127.0.0.1:0".parse().unwrap());
|
|
c.contrib.max_bitrate_kbps = 1_000_000;
|
|
c.limits.max_queue_bytes = 256 << 20;
|
|
})
|
|
.await;
|
|
let auth = w.auth(&w.publisher, 0);
|
|
let s = open_srt(&w, &auth).await;
|
|
let cfg = CallConfig { latency_ms: 2_000, ..call_cfg(&s) };
|
|
let link = timeout(STEP, srt::call(w.gateway.srt_addr.unwrap(), cfg)).await.unwrap().unwrap();
|
|
assert!(contrib_live(&w, s["session"].as_str().unwrap()).await);
|
|
let watcher = watch(&w).await;
|
|
let chunks = joule_contrib::cli::synthetic(4, 20_000);
|
|
joule_contrib::cli::pace(&link, &chunks, 800).await.unwrap();
|
|
assert!(until(STEP, || link.sent.lock().unwrap().unacknowledged == 0).await, "the gateway acknowledged every packet before the publisher closed");
|
|
link.close();
|
|
assert!(until(STEP, || watcher.ended.lock().unwrap().len() == 2).await, "PUBLISH_DONE on both media tracks");
|
|
settle(&watcher, Duration::from_millis(500)).await;
|
|
assert_eq!(w.relay.relay().stats().refused_at_door, 0, "the door released every group");
|
|
assert!(until(STEP, || w.gateway.gateway.link_records().iter().any(|b| LinkRecord::read(b).is_ok_and(|r| r.last))).await, "the last part is signed");
|
|
let ps = parts(&w);
|
|
assert!(ps.iter().all(|p| p.window == 2_000_000), "the caller's window, the larger of the two ends'");
|
|
assert_eq!(ps.iter().map(|p| p.account.dropped).sum::<u64>(), 0, "no packet dropped as too late");
|
|
let all: Vec<u8> = chunks.iter().flat_map(|c| c.1.clone()).collect();
|
|
let mut d = ts::Demux::new();
|
|
let mut pes = d.push(&all);
|
|
pes.extend(d.flush());
|
|
let want: Vec<Vec<u8>> = pes.iter().filter(|p| p.stream_type == ts::ST_H264).map(|p| p.data.clone()).collect();
|
|
let video = watcher.tracks.iter().find(|t| t.kind == media::VIDEO).unwrap();
|
|
let (vf, vg) = verified(&watcher, video, &w.gateway_key, Codec::Avc);
|
|
assert!(want.iter().map(|f| f.len()).sum::<usize>() > 9_000_000, "groups of megabytes");
|
|
same(&vf, &want, "every access unit");
|
|
assert_eq!(vg, 4);
|
|
w.stop().await;
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn contribution_at_twenty_megabits_passes_the_door_with_no_subscriber() {
|
|
timeout(Duration::from_secs(120), async {
|
|
let w = World::start(|c| {
|
|
c.contrib.srt.listen = Some("127.0.0.1:0".parse().unwrap());
|
|
c.contrib.max_bitrate_kbps = 1_000_000;
|
|
c.limits.max_queue_bytes = 256 << 20;
|
|
})
|
|
.await;
|
|
let auth = w.auth(&w.publisher, 0);
|
|
let s = open_srt(&w, &auth).await;
|
|
let link = timeout(STEP, srt::call(w.gateway.srt_addr.unwrap(), call_cfg(&s))).await.unwrap().unwrap();
|
|
let chunks = joule_contrib::cli::synthetic(4, 20_000);
|
|
joule_contrib::cli::pace(&link, &chunks, 800).await.unwrap();
|
|
link.close();
|
|
assert!(until(STEP, || w.gateway.gateway.records().iter().any(|r| r.record.protocol == "srt")).await, "the session ended");
|
|
assert_eq!(w.relay.relay().stats().refused_at_door, 0, "the door released every group");
|
|
let rec = w.gateway.gateway.records().into_iter().find(|r| r.record.protocol == "srt").unwrap();
|
|
assert_eq!(rec.record.ended, "peer_shutdown", "the session ended as the publisher ended it, not by the relay");
|
|
w.stop().await;
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|