207 lines
11 KiB
Rust
207 lines
11 KiB
Rust
//! Enhanced RTMP from a local encoder, used purely as a test client. Each case runs the
|
|
//! encoder on a generated test pattern and tone, publishes to the gateway with a stream key
|
|
//! from `/rtmp`, and a subscriber at the relay receives every group, verifies every
|
|
//! receipt under the gateway's key, writes each track's frames out as an elementary stream,
|
|
//! and decodes it with the encoder's tool, a conformant decoder: the count of decoded frames
|
|
//! is the count of objects the relay carried, which is the count the encoder made.
|
|
//!
|
|
//! The encoder sends no media before the gateway's `NetStream.Publish.Start`, so a proxy
|
|
//! between the two holds that answer until the subscriber is in place: every group is seen.
|
|
//!
|
|
//! These tests need the encoder binary (`WAI_TEST_FFMPEG`, or `ffmpeg` and `ffprobe` on the
|
|
//! PATH) and, for the hardware cases, the platform's media engine. They are ignored by
|
|
//! default and run where both are present: `cargo test --test ertmp_ffmpeg -- --ignored`.
|
|
//! A case whose encoder is missing fails; it is never passed over.
|
|
|
|
mod common;
|
|
|
|
use common::rtmp::stream_key;
|
|
use common::*;
|
|
use joule_moq_ingest::frame::Codec;
|
|
use joule_moq_ingest::room::media;
|
|
use std::process::Command;
|
|
use std::time::Duration;
|
|
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
|
use tokio::net::{TcpListener, TcpStream};
|
|
use tokio::time::timeout;
|
|
|
|
fn tool(name: &str) -> String {
|
|
match std::env::var("WAI_TEST_FFMPEG") {
|
|
Ok(p) if name == "ffmpeg" => p,
|
|
Ok(p) => std::path::Path::new(&p).with_file_name(name).to_string_lossy().into_owned(),
|
|
Err(_) => name.to_string(),
|
|
}
|
|
}
|
|
|
|
/// A proxy from a listener of its own to the gateway's RTMP port; it holds the gateway's
|
|
/// bytes once the publish is admitted until `ready` is set.
|
|
async fn proxy(w: &World, ready: std::sync::Arc<std::sync::atomic::AtomicBool>) -> u16 {
|
|
let l = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
|
let port = l.local_addr().unwrap().port();
|
|
let up = w.gateway.rtmp_addr.unwrap();
|
|
let gw = w.gateway.gateway.clone();
|
|
tokio::spawn(async move {
|
|
let (c, _) = l.accept().await.unwrap();
|
|
let s = TcpStream::connect(up).await.unwrap();
|
|
let (mut cr, mut cw) = c.into_split();
|
|
let (mut sr, mut sw) = s.into_split();
|
|
tokio::spawn(async move {
|
|
let mut b = vec![0u8; 65536];
|
|
while let Ok(n) = cr.read(&mut b).await {
|
|
if n == 0 || sw.write_all(&b[..n]).await.is_err() {
|
|
break;
|
|
}
|
|
}
|
|
let _ = sw.shutdown().await;
|
|
});
|
|
let mut b = vec![0u8; 65536];
|
|
while let Ok(n) = sr.read(&mut b).await {
|
|
if n == 0 {
|
|
break;
|
|
}
|
|
if gw.rtmp_publishing() {
|
|
while !ready.load(std::sync::atomic::Ordering::Relaxed) {
|
|
tokio::time::sleep(Duration::from_millis(5)).await;
|
|
}
|
|
}
|
|
if cw.write_all(&b[..n]).await.is_err() {
|
|
break;
|
|
}
|
|
}
|
|
});
|
|
port
|
|
}
|
|
|
|
struct Case {
|
|
name: &'static str,
|
|
/// The encoder's arguments between its inputs and its output.
|
|
args: &'static [&'static str],
|
|
video: Codec,
|
|
audio: Option<Codec>,
|
|
/// The elementary stream's extension, as the decoder reads it.
|
|
vformat: &'static str,
|
|
aformat: Option<&'static str>,
|
|
}
|
|
|
|
const FRAMES: usize = 90;
|
|
|
|
async fn run(c: &Case) {
|
|
let w = World::start(|g| g.rtmp.listen = Some("127.0.0.1:0".parse().unwrap())).await;
|
|
let auth = w.auth(&w.publisher, 0);
|
|
let key = stream_key(&w, &auth).await;
|
|
let ready = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
|
|
let port = proxy(&w, ready.clone()).await;
|
|
let url = format!("rtmp://127.0.0.1:{port}/live/{key}");
|
|
let mut args: Vec<String> = ["-hide_banner", "-loglevel", "error", "-re", "-f", "lavfi", "-i", "testsrc2=size=640x360:rate=30", "-f", "lavfi", "-i", "sine=frequency=440:sample_rate=48000"]
|
|
.iter()
|
|
.map(|s| s.to_string())
|
|
.collect();
|
|
args.extend(["-frames:v".into(), FRAMES.to_string(), "-t".into(), "3".into()]);
|
|
args.extend(c.args.iter().map(|s| s.to_string()));
|
|
if c.audio.is_none() {
|
|
args.push("-an".into());
|
|
}
|
|
args.extend(["-f".into(), "flv".into(), url]);
|
|
let mut child = tokio::process::Command::new(tool("ffmpeg")).args(&args).spawn().unwrap_or_else(|e| panic!("{}: the encoder ({}): {e}", c.name, tool("ffmpeg")));
|
|
// Subscribe once the publish is admitted, then let the encoder start.
|
|
assert!(until(STEP, || w.gateway.gateway.rtmp_publishing()).await, "{}: the publish is admitted", c.name);
|
|
let kinds: Vec<u8> = if c.audio.is_some() { vec![media::AUDIO, media::VIDEO] } else { vec![media::VIDEO] };
|
|
let watcher = watch_kinds(&w, &kinds).await;
|
|
ready.store(true, std::sync::atomic::Ordering::Relaxed);
|
|
let st = timeout(Duration::from_secs(60), child.wait()).await.expect("the encoder finishes").unwrap();
|
|
assert!(st.success(), "{}: the encoder's status {st}", c.name);
|
|
assert!(until(STEP, || watcher.ended.lock().unwrap().len() == kinds.len()).await, "{}: every track ends", c.name);
|
|
settle(&watcher, Duration::from_millis(400)).await;
|
|
let dir = std::env::temp_dir().join(format!("ertmp-ffmpeg-{}-{}", c.name, std::process::id()));
|
|
std::fs::create_dir_all(&dir).unwrap();
|
|
let video = watcher.tracks.iter().find(|t| t.kind == media::VIDEO).unwrap();
|
|
let (frames, groups) = verified(&watcher, video, &w.gateway_key, c.video);
|
|
assert_eq!(frames.len(), FRAMES, "{}: every frame the encoder made reached the subscriber", c.name);
|
|
assert!(groups >= 3, "{}: {groups} groups", c.name);
|
|
let env = envelopes(&watcher, video);
|
|
let es = elementary(c.video, &env.iter().map(|e| e.payload.clone()).collect::<Vec<_>>());
|
|
let vpath = dir.join(format!("video.{}", c.vformat));
|
|
std::fs::write(&vpath, es).unwrap();
|
|
assert_eq!(decoded(&vpath, c.vformat, "v"), FRAMES, "{}: the decoder decodes every frame", c.name);
|
|
if let Some(ac) = c.audio {
|
|
let audio = watcher.tracks.iter().find(|t| t.kind == media::AUDIO).unwrap();
|
|
let (af_, _) = verified(&watcher, audio, &w.gateway_key, ac);
|
|
assert!(!af_.is_empty(), "{}: audio arrived", c.name);
|
|
// An Opus packet is not a stream a decoder's command line reads without a
|
|
// container, so its packets are verified and counted, not decoded here.
|
|
if let Some(af) = c.aformat {
|
|
let env = envelopes(&watcher, audio);
|
|
let es = elementary(ac, &env.iter().map(|e| e.payload.clone()).collect::<Vec<_>>());
|
|
let apath = dir.join(format!("audio.{af}"));
|
|
std::fs::write(&apath, es).unwrap();
|
|
assert_eq!(decoded(&apath, af, "a"), env.len(), "{}: the decoder decodes every audio frame", c.name);
|
|
}
|
|
}
|
|
assert_eq!(w.relay.relay().stats().refused_at_door, 0, "{}: the relay's door verified every object", c.name);
|
|
assert!(until(STEP, || !w.gateway.gateway.records().is_empty()).await);
|
|
let rec = w.gateway.gateway.records().remove(0);
|
|
rec.verify().unwrap();
|
|
let counts = rec.record.detail["counts"].as_object().unwrap();
|
|
eprintln!("{}: what the gateway read and did not carry: {counts:?}", c.name);
|
|
assert!(counts.keys().all(|k| k == "metadata" || k == "packet_not_carried"), "{}: no frame refused: {counts:?}", c.name);
|
|
eprintln!("{}: {} video objects in {groups} groups, catalog codec from the configuration; decoded", c.name, frames.len());
|
|
let _ = std::fs::remove_dir_all(&dir);
|
|
w.stop().await;
|
|
}
|
|
|
|
/// The objects' payloads as one elementary stream the decoder reads.
|
|
fn elementary(codec: Codec, payloads: &[Vec<u8>]) -> Vec<u8> {
|
|
match codec {
|
|
// Annex B access units and ADTS frames concatenate.
|
|
Codec::Avc | Codec::Hevc | Codec::Aac => payloads.concat(),
|
|
// SPEC §5's AV1 framing: 24 bytes, then the temporal unit (a low-overhead stream).
|
|
Codec::Av1 => payloads.iter().flat_map(|p| p[24..].to_vec()).collect(),
|
|
// One-frame IVF files: the first's file header, then each frame header and frame.
|
|
Codec::Vp8 | Codec::Vp9 => {
|
|
let mut o = payloads[0][..32].to_vec();
|
|
o[24..28].copy_from_slice(&(payloads.len() as u32).to_le_bytes());
|
|
for p in payloads {
|
|
o.extend_from_slice(&p[32..]);
|
|
}
|
|
o
|
|
}
|
|
// A FLAC stream: the first object's, then the frames of the others.
|
|
Codec::Flac => {
|
|
let mut o = payloads[0].clone();
|
|
for p in &payloads[1..] {
|
|
o.extend_from_slice(&p[42..]);
|
|
}
|
|
o
|
|
}
|
|
_ => payloads.concat(),
|
|
}
|
|
}
|
|
|
|
/// The frames the decoder decodes from `path` read as `format`, by the stream type `s`.
|
|
fn decoded(path: &std::path::Path, format: &str, s: &str) -> usize {
|
|
let fmt = format;
|
|
let out = Command::new(tool("ffprobe"))
|
|
.args(["-v", "error", "-f", fmt, "-count_frames", "-select_streams", &format!("{s}:0"), "-show_entries", "stream=nb_read_frames", "-of", "csv=p=0"])
|
|
.arg(path)
|
|
.output()
|
|
.expect("the decoder runs");
|
|
assert!(out.status.success(), "decoding {}: {}", path.display(), String::from_utf8_lossy(&out.stderr));
|
|
String::from_utf8_lossy(&out.stdout).trim().trim_end_matches(',').parse().unwrap_or_else(|_| panic!("a frame count: {:?}", String::from_utf8_lossy(&out.stdout)))
|
|
}
|
|
|
|
macro_rules! case {
|
|
($fn:ident, $c:expr) => {
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
#[ignore = "needs a local encoder (WAI_TEST_FFMPEG or ffmpeg on the PATH); run with --ignored"]
|
|
async fn $fn() {
|
|
timeout(Duration::from_secs(120), run(&$c)).await.expect("in time");
|
|
}
|
|
};
|
|
}
|
|
|
|
case!(legacy_avc_with_b_frames_and_aac, Case { name: "x264-aac", args: &["-c:v", "libx264", "-preset", "veryfast", "-g", "30", "-bf", "2", "-c:a", "aac", "-b:a", "96k"], video: Codec::Avc, audio: Some(Codec::Aac), vformat: "h264", aformat: Some("aac") });
|
|
case!(hevc_on_the_media_engine_and_aac, Case { name: "hw-hevc-aac", args: &["-c:v", "hevc_videotoolbox", "-g", "30", "-b:v", "2M", "-c:a", "aac", "-b:a", "96k"], video: Codec::Hevc, audio: Some(Codec::Aac), vformat: "hevc", aformat: Some("aac") });
|
|
case!(h264_on_the_media_engine, Case { name: "hw-h264", args: &["-c:v", "h264_videotoolbox", "-g", "30", "-b:v", "2M"], video: Codec::Avc, audio: None, vformat: "h264", aformat: None });
|
|
case!(hevc_in_software_and_opus, Case { name: "x265-opus", args: &["-c:v", "libx265", "-preset", "ultrafast", "-x265-params", "keyint=30:min-keyint=30:log-level=error", "-c:a", "libopus", "-b:a", "96k"], video: Codec::Hevc, audio: Some(Codec::Opus), vformat: "hevc", aformat: None });
|
|
case!(av1_and_flac, Case { name: "svtav1-flac", args: &["-c:v", "libsvtav1", "-preset", "12", "-g", "30", "-svtav1-params", "keyint=30:scd=0", "-c:a", "flac"], video: Codec::Av1, audio: Some(Codec::Flac), vformat: "obu", aformat: Some("flac") });
|
|
case!(vp9_and_opus, Case { name: "vp9-opus", args: &["-c:v", "libvpx-vp9", "-deadline", "realtime", "-cpu-used", "8", "-g", "30", "-b:v", "1M", "-c:a", "libopus"], video: Codec::Vp9, audio: Some(Codec::Opus), vformat: "ivf", aformat: None });
|