use std::time::Duration;
use bytes::Bytes;
use hang::catalog::{AudioCodec, VideoCodec};
use super::{Export, Import};
fn avcc() -> Vec<u8> {
let sps = [0x67u8, 0x42, 0xc0, 0x1f];
let mut out = vec![0x01, 0x42, 0xc0, 0x1f, 0xff, 0xe1, 0x00, sps.len() as u8];
out.extend_from_slice(&sps);
out.extend_from_slice(&[0x01, 0x00, 0x04, 0x68, 0xce, 0x3c, 0x80]);
out
}
const ASC: [u8; 2] = [0x12, 0x10];
fn write_tag(out: &mut Vec<u8>, tag_type: u8, timestamp: u32, body: &[u8]) {
out.push(tag_type);
out.extend_from_slice(&(body.len() as u32).to_be_bytes()[1..]);
out.extend_from_slice(×tamp.to_be_bytes()[1..]);
out.push((timestamp >> 24) as u8);
out.extend_from_slice(&[0, 0, 0]);
out.extend_from_slice(body);
out.extend_from_slice(&(11 + body.len() as u32).to_be_bytes());
}
fn synth_flv() -> Vec<u8> {
let mut out = Vec::new();
out.extend_from_slice(b"FLV");
out.push(1);
out.push(0x05);
out.extend_from_slice(&9u32.to_be_bytes());
out.extend_from_slice(&0u32.to_be_bytes());
let mut vseq = vec![
(super::FRAME_TYPE_KEY << 4) | super::VIDEO_CODEC_AVC,
super::AVC_SEQUENCE_HEADER,
0,
0,
0,
];
vseq.extend_from_slice(&avcc());
write_tag(&mut out, super::TAG_VIDEO, 0, &vseq);
let mut aseq = vec![super::AAC_AUDIO_TAG_HEADER, super::AAC_SEQUENCE_HEADER];
aseq.extend_from_slice(&ASC);
write_tag(&mut out, super::TAG_AUDIO, 0, &aseq);
let idr = [0, 0, 0, 5, 0x65, 0x88, 0x84, 0x21, 0x00];
let mut vkey = vec![
(super::FRAME_TYPE_KEY << 4) | super::VIDEO_CODEC_AVC,
super::AVC_NALU,
0,
0,
0,
];
vkey.extend_from_slice(&idr);
write_tag(&mut out, super::TAG_VIDEO, 0, &vkey);
let p = [0, 0, 0, 4, 0x41, 0xe0, 0x12, 0x34];
let mut vinter = vec![
(super::FRAME_TYPE_INTER << 4) | super::VIDEO_CODEC_AVC,
super::AVC_NALU,
0,
0,
0,
];
vinter.extend_from_slice(&p);
write_tag(&mut out, super::TAG_VIDEO, 33, &vinter);
let mut a0 = vec![super::AAC_AUDIO_TAG_HEADER, super::AAC_RAW];
a0.extend_from_slice(&[0xde, 0xad]);
write_tag(&mut out, super::TAG_AUDIO, 0, &a0);
let mut a1 = vec![super::AAC_AUDIO_TAG_HEADER, super::AAC_RAW];
a1.extend_from_slice(&[0xbe, 0xef]);
write_tag(&mut out, super::TAG_AUDIO, 23, &a1);
out
}
async fn drain_export(mut exporter: Export, mut importer: Import) -> Vec<u8> {
importer.finish().unwrap();
let mut exported = Vec::new();
let mut importer = Some(importer);
for _ in 0..64 {
match tokio::time::timeout(Duration::from_millis(100), exporter.next()).await {
Ok(Ok(Some(chunk))) => exported.extend_from_slice(&chunk),
Ok(Ok(None)) => break,
Ok(Err(e)) => panic!("exporter error: {e}"),
Err(_) => importer = None, }
}
drop(importer);
exported
}
#[tokio::test(start_paused = true)]
async fn export_roundtrips_through_import() {
let mut producer = moq_net::Broadcast::new().produce();
let consumer = producer.consume();
let mut catalog = crate::catalog::Producer::new(&mut producer).unwrap();
let mut importer = Import::new(producer, catalog.clone());
importer.decode(&bytes::BytesMut::from(synth_flv().as_slice())).unwrap();
catalog.finish().unwrap();
let exporter = Export::new(consumer).unwrap();
let exported = drain_export(exporter, importer).await;
assert_eq!(&exported[0..3], b"FLV");
let mut bcast2 = moq_net::Broadcast::new().produce();
let cat2 = crate::catalog::Producer::new(&mut bcast2).unwrap();
let mut imp2 = Import::new(bcast2, cat2.clone());
imp2.decode(&bytes::BytesMut::from(exported.as_slice())).unwrap();
imp2.finish().unwrap();
let snap = cat2.snapshot();
assert_eq!(snap.video.renditions.len(), 1);
assert_eq!(snap.audio.renditions.len(), 1);
let v = snap.video.renditions.values().next().unwrap();
assert!(matches!(v.codec, VideoCodec::H264(_)));
assert_eq!(v.description.as_ref().map(|b| b.as_ref()), Some(avcc().as_slice()));
let a = snap.audio.renditions.values().next().unwrap();
assert!(matches!(a.codec, AudioCodec::AAC(_)));
assert_eq!(a.sample_rate, 44100);
assert_eq!(a.description.as_ref().map(|b| b.as_ref()), Some(&ASC[..]));
}
#[tokio::test(start_paused = true)]
async fn export_emits_sequence_headers_and_frames() {
let mut producer = moq_net::Broadcast::new().produce();
let consumer = producer.consume();
let mut catalog = crate::catalog::Producer::new(&mut producer).unwrap();
let mut importer = Import::new(producer, catalog.clone());
importer.decode(&bytes::BytesMut::from(synth_flv().as_slice())).unwrap();
catalog.finish().unwrap();
let exporter = Export::new(consumer).unwrap();
let exported = drain_export(exporter, importer).await;
let tags = parse_tags(&exported);
let avc_seq = tags
.iter()
.filter(|t| t.tag_type == super::TAG_VIDEO && t.body[1] == super::AVC_SEQUENCE_HEADER)
.count();
let aac_seq = tags
.iter()
.filter(|t| t.tag_type == super::TAG_AUDIO && t.body[1] == super::AAC_SEQUENCE_HEADER)
.count();
assert_eq!(avc_seq, 1, "expected one AVC sequence header");
assert_eq!(aac_seq, 1, "expected one AAC sequence header");
let video_frames = tags
.iter()
.filter(|t| t.tag_type == super::TAG_VIDEO && t.body[1] == super::AVC_NALU)
.count();
let audio_frames = tags
.iter()
.filter(|t| t.tag_type == super::TAG_AUDIO && t.body[1] == super::AAC_RAW)
.count();
assert_eq!(video_frames, 2, "expected two video frames");
assert_eq!(audio_frames, 2, "expected two audio frames");
}
const VP9_KEYFRAME: &[u8] = &[0x82, 0x49, 0x83, 0x42, 0x20, 0x13, 0xf0, 0x0e, 0xf0, 0x00];
fn synth_enhanced_flv() -> Vec<u8> {
let head = crate::codec::opus::Config {
sample_rate: 48000,
channel_count: 2,
}
.encode()
.unwrap();
let mut out = Vec::new();
out.extend_from_slice(b"FLV");
out.push(1);
out.push(0x05);
out.extend_from_slice(&9u32.to_be_bytes());
out.extend_from_slice(&0u32.to_be_bytes());
let mut aseq = vec![(super::AUDIO_FORMAT_EX << 4) | super::AUDIO_PACKET_SEQUENCE_START];
aseq.extend_from_slice(b"Opus");
aseq.extend_from_slice(&head);
write_tag(&mut out, super::TAG_AUDIO, 0, &aseq);
let mut vkey = vec![super::VIDEO_EX_HEADER | (super::FRAME_TYPE_KEY << 4) | super::VIDEO_PACKET_CODED_FRAMES];
vkey.extend_from_slice(b"vp09");
vkey.extend_from_slice(VP9_KEYFRAME);
write_tag(&mut out, super::TAG_VIDEO, 0, &vkey);
let mut a0 = vec![(super::AUDIO_FORMAT_EX << 4) | super::AUDIO_PACKET_CODED_FRAMES];
a0.extend_from_slice(b"Opus");
a0.extend_from_slice(&[0xfc, 0xff, 0xfe]);
write_tag(&mut out, super::TAG_AUDIO, 20, &a0);
out
}
#[tokio::test(start_paused = true)]
async fn export_roundtrips_enhanced() {
let mut producer = moq_net::Broadcast::new().produce();
let consumer = producer.consume();
let mut catalog = crate::catalog::Producer::new(&mut producer).unwrap();
let mut importer = Import::new(producer, catalog.clone());
importer
.decode(&bytes::BytesMut::from(synth_enhanced_flv().as_slice()))
.unwrap();
catalog.finish().unwrap();
let exporter = Export::new(consumer).unwrap();
let exported = drain_export(exporter, importer).await;
let tags = parse_tags(&exported);
assert!(
tags.iter().any(|t| t.tag_type == super::TAG_VIDEO
&& t.body[0] & super::VIDEO_EX_HEADER != 0
&& &t.body[1..5] == b"vp09"),
"expected an enhanced vp09 video tag"
);
assert!(
tags.iter().any(|t| t.tag_type == super::TAG_AUDIO
&& (t.body[0] >> 4) == super::AUDIO_FORMAT_EX
&& &t.body[1..5] == b"Opus"),
"expected an enhanced Opus audio tag"
);
let mut bcast2 = moq_net::Broadcast::new().produce();
let cat2 = crate::catalog::Producer::new(&mut bcast2).unwrap();
let mut imp2 = Import::new(bcast2, cat2.clone());
imp2.decode(&bytes::BytesMut::from(exported.as_slice())).unwrap();
imp2.finish().unwrap();
let snap = cat2.snapshot();
assert!(matches!(
snap.video.renditions.values().next().unwrap().codec,
VideoCodec::VP9(_)
));
assert!(matches!(
snap.audio.renditions.values().next().unwrap().codec,
AudioCodec::Opus
));
}
#[tokio::test(start_paused = true)]
async fn export_roundtrips_mp3() {
let mut mp3 = vec![0xFF, 0xFB, 0x90, 0x44];
mp3.resize(417, 0xAA);
let mut flv = Vec::new();
flv.extend_from_slice(b"FLV");
flv.push(1);
flv.push(0x04); flv.extend_from_slice(&9u32.to_be_bytes());
flv.extend_from_slice(&0u32.to_be_bytes());
let mut tag = vec![super::MP3_AUDIO_TAG_HEADER];
tag.extend_from_slice(&mp3);
write_tag(&mut flv, super::TAG_AUDIO, 0, &tag);
let mut producer = moq_net::Broadcast::new().produce();
let consumer = producer.consume();
let mut catalog = crate::catalog::Producer::new(&mut producer).unwrap();
let mut importer = Import::new(producer, catalog.clone());
importer.decode(&bytes::BytesMut::from(flv.as_slice())).unwrap();
catalog.finish().unwrap();
let exporter = Export::new(consumer).unwrap();
let exported = drain_export(exporter, importer).await;
let tags = parse_tags(&exported);
assert!(
tags.iter()
.any(|t| t.tag_type == super::TAG_AUDIO && (t.body[0] >> 4) == super::AUDIO_FORMAT_MP3),
"expected a legacy MP3 audio tag"
);
let mut bcast2 = moq_net::Broadcast::new().produce();
let cat2 = crate::catalog::Producer::new(&mut bcast2).unwrap();
let mut imp2 = Import::new(bcast2, cat2.clone());
imp2.decode(&bytes::BytesMut::from(exported.as_slice())).unwrap();
imp2.finish().unwrap();
let snap = cat2.snapshot();
let a = snap.audio.renditions.values().next().unwrap();
assert!(matches!(a.codec, AudioCodec::Mp3));
assert_eq!(a.sample_rate, 44100);
assert_eq!(a.channel_count, 2);
}
fn avcc_level(level: u8) -> Vec<u8> {
let sps = [0x67u8, 0x42, 0xc0, level];
let mut out = vec![0x01, 0x42, 0xc0, level, 0xff, 0xe1, 0x00, sps.len() as u8];
out.extend_from_slice(&sps);
out.extend_from_slice(&[0x01, 0x00, 0x04, 0x68, 0xce, 0x3c, 0x80]);
out
}
type Keepalive = (
moq_net::BroadcastProducer,
crate::catalog::Producer,
Vec<crate::container::Producer<crate::catalog::hang::Container>>,
);
fn build_multitrack_broadcast() -> (moq_net::BroadcastConsumer, Vec<Vec<u8>>, Keepalive) {
use hang::catalog::{AAC, AudioConfig, Container, H264, VideoConfig};
use crate::container::{Producer, Timestamp};
let mut producer = moq_net::Broadcast::new().produce();
let consumer = producer.consume();
let mut catalog = crate::catalog::Producer::new(&mut producer).unwrap();
let mut tracks = Vec::new();
let descriptions: Vec<Vec<u8>> = vec![avcc_level(0x1f), avcc_level(0x1e)];
for description in &descriptions {
let track = producer
.create_track(moq_net::Track::new(producer.unique_name(".avc1")))
.unwrap();
let mut config = VideoConfig::new(H264 {
profile: 0x42,
constraints: 0xc0,
level: description[3],
inline: false,
});
config.container = Container::Legacy;
config.description = Some(Bytes::from(description.clone()));
catalog.lock().video.renditions.insert(track.name().to_string(), config);
let mut video = Producer::new(track, crate::catalog::hang::Container::Legacy);
video
.write(crate::container::Frame {
timestamp: Timestamp::from_millis(0).unwrap(),
duration: None,
payload: Bytes::from_static(&[0, 0, 0, 1, 0x65]),
keyframe: true,
})
.unwrap();
video.finish().unwrap();
tracks.push(video);
}
let audio_track = producer
.create_track(moq_net::Track::new(producer.unique_name(".aac")))
.unwrap();
let mut audio_config = AudioConfig::new(AAC { profile: 2 }, 44100, 2);
audio_config.container = Container::Legacy;
audio_config.description = Some(Bytes::from_static(&ASC));
catalog
.lock()
.audio
.renditions
.insert(audio_track.name().to_string(), audio_config);
let mut audio = crate::container::Producer::new(audio_track, crate::catalog::hang::Container::Legacy);
audio
.write(crate::container::Frame {
timestamp: Timestamp::from_millis(0).unwrap(),
duration: None,
payload: Bytes::from_static(&[0xde, 0xad]),
keyframe: true,
})
.unwrap();
audio.finish().unwrap();
tracks.push(audio);
catalog.finish().unwrap();
(consumer, descriptions, (producer, catalog, tracks))
}
async fn drain_to_end(mut exporter: Export, keepalive: Keepalive) -> Vec<u8> {
let mut exported = Vec::new();
let mut keepalive = Some(keepalive);
for _ in 0..64 {
match tokio::time::timeout(Duration::from_millis(100), exporter.next()).await {
Ok(Ok(Some(chunk))) => exported.extend_from_slice(&chunk),
Ok(Ok(None)) => break,
Ok(Err(e)) => panic!("exporter error: {e}"),
Err(_) => keepalive = None, }
}
drop(keepalive);
exported
}
#[tokio::test(start_paused = true)]
async fn export_multitrack_roundtrips_all_renditions() {
let (consumer, descriptions, keepalive) = build_multitrack_broadcast();
let exporter = Export::new(consumer).unwrap().with_multitrack(true);
let exported = drain_to_end(exporter, keepalive).await;
assert_eq!(&exported[0..3], b"FLV");
let tags = parse_tags(&exported);
let mut video_track_ids: Vec<u8> = tags
.iter()
.filter(|t| {
t.tag_type == super::TAG_VIDEO
&& t.body[0] & super::VIDEO_EX_HEADER != 0
&& t.body[0] & 0x0f == super::VIDEO_PACKET_MULTITRACK
&& t.body[1] & 0x0f == super::VIDEO_PACKET_CODED_FRAMES
&& &t.body[2..6] == b"avc1"
})
.map(|t| t.body[6])
.collect();
video_track_ids.sort_unstable();
video_track_ids.dedup();
assert_eq!(video_track_ids, vec![0, 1], "expected two multitrack video track ids");
let mut bcast2 = moq_net::Broadcast::new().produce();
let cat2 = crate::catalog::Producer::new(&mut bcast2).unwrap();
let mut imp2 = Import::new(bcast2, cat2.clone());
imp2.decode(&bytes::BytesMut::from(exported.as_slice())).unwrap();
imp2.finish().unwrap();
let snap = cat2.snapshot();
assert_eq!(
snap.video.renditions.len(),
2,
"both video renditions should round-trip"
);
assert_eq!(snap.audio.renditions.len(), 1);
let mut got: Vec<Vec<u8>> = snap
.video
.renditions
.values()
.filter_map(|c| c.description.as_ref().map(|b| b.to_vec()))
.collect();
got.sort();
let mut want = descriptions.clone();
want.sort();
assert_eq!(got, want, "each rendition should keep its own avcC");
}
#[tokio::test(start_paused = true)]
async fn export_without_multitrack_keeps_first_rendition() {
let (consumer, _, keepalive) = build_multitrack_broadcast();
let exporter = Export::new(consumer).unwrap();
let exported = drain_to_end(exporter, keepalive).await;
let tags = parse_tags(&exported);
assert!(
tags.iter()
.all(|t| !(t.tag_type == super::TAG_VIDEO && t.body[0] & 0x0f == super::VIDEO_PACKET_MULTITRACK)),
"single-track export must not use multitrack framing"
);
let mut bcast2 = moq_net::Broadcast::new().produce();
let cat2 = crate::catalog::Producer::new(&mut bcast2).unwrap();
let mut imp2 = Import::new(bcast2, cat2.clone());
imp2.decode(&bytes::BytesMut::from(exported.as_slice())).unwrap();
imp2.finish().unwrap();
assert_eq!(
cat2.snapshot().video.renditions.len(),
1,
"only one rendition without multitrack"
);
}
struct ParsedTag {
tag_type: u8,
timestamp: u32,
body: Vec<u8>,
}
fn parse_tags(flv: &[u8]) -> Vec<ParsedTag> {
let mut tags = Vec::new();
let mut off = 9 + 4; while off + 11 <= flv.len() {
let tag_type = flv[off];
let size = super::read_u24(&flv[off + 1..off + 4]) as usize;
let timestamp = super::read_u24(&flv[off + 4..off + 7]) | ((flv[off + 7] as u32) << 24);
let body_start = off + 11;
if body_start + size + 4 > flv.len() {
break;
}
tags.push(ParsedTag {
tag_type,
timestamp,
body: flv[body_start..body_start + size].to_vec(),
});
off = body_start + size + 4;
}
tags
}
#[tokio::test(start_paused = true)]
async fn export_preserves_timestamps() {
let mut producer = moq_net::Broadcast::new().produce();
let consumer = producer.consume();
let mut catalog = crate::catalog::Producer::new(&mut producer).unwrap();
let mut importer = Import::new(producer, catalog.clone());
importer.decode(&bytes::BytesMut::from(synth_flv().as_slice())).unwrap();
catalog.finish().unwrap();
let exporter = Export::new(consumer).unwrap();
let exported = drain_export(exporter, importer).await;
let tags = parse_tags(&exported);
let video_pts: Vec<i64> = tags
.iter()
.filter(|t| t.tag_type == super::TAG_VIDEO && t.body[1] == super::AVC_NALU)
.map(|t| i64::from(t.timestamp) + i64::from(super::read_i24(&t.body[2..5])))
.collect();
assert_eq!(video_pts, vec![0, 33]);
}
#[tokio::test(start_paused = true)]
async fn export_authors_dts_and_composition_time_for_reordered_avc() {
use crate::container::Timestamp;
use hang::catalog::{AAC, AudioConfig, Container, H264, VideoConfig};
let broadcast = moq_net::Broadcast::new();
let mut producer = broadcast.produce();
let consumer = producer.consume();
let mut catalog = crate::catalog::Producer::new(&mut producer).unwrap();
let video_track = producer
.create_track(moq_net::Track::new(producer.unique_name(".avc1")))
.unwrap();
let audio_track = producer
.create_track(moq_net::Track::new(producer.unique_name(".aac")))
.unwrap();
let mut video_config = VideoConfig::new(H264 {
profile: 0x42,
constraints: 0xc0,
level: 0x1f,
inline: false,
});
video_config.container = Container::Legacy;
video_config.description = Some(Bytes::from(avcc()));
video_config.jitter = Some(moq_net::Time::try_from(Duration::from_millis(80)).unwrap());
catalog
.lock()
.video
.renditions
.insert(video_track.name().to_string(), video_config);
let mut audio_config = AudioConfig::new(AAC { profile: 2 }, 44100, 2);
audio_config.container = Container::Legacy;
audio_config.description = Some(Bytes::from_static(&ASC));
catalog
.lock()
.audio
.renditions
.insert(audio_track.name().to_string(), audio_config);
let mut video = crate::container::Producer::new(video_track, crate::catalog::hang::Container::Legacy);
let video_frame = |timestamp_ms: u64, payload: &'static [u8], keyframe| crate::container::Frame {
timestamp: Timestamp::from_millis(timestamp_ms).unwrap(),
duration: None,
payload: Bytes::from_static(payload),
keyframe,
};
video.write(video_frame(0, &[0, 0, 0, 1, 0x65], true)).unwrap();
video.write(video_frame(80, &[0, 0, 0, 1, 0x41], false)).unwrap();
video.write(video_frame(40, &[0, 0, 0, 1, 0x01], false)).unwrap();
video.write(video_frame(120, &[0, 0, 0, 1, 0x41], false)).unwrap();
video.finish().unwrap();
let mut audio = crate::container::Producer::new(audio_track, crate::catalog::hang::Container::Legacy);
audio
.write(crate::container::Frame {
timestamp: Timestamp::from_millis(20).unwrap(),
duration: None,
payload: Bytes::from_static(&[0xde, 0xad]),
keyframe: true,
})
.unwrap();
audio.finish().unwrap();
catalog.finish().unwrap();
let exporter = Export::new(consumer).unwrap();
let exported = drain_exporter_chunks(exporter, 6).await;
let tags = parse_tags(&exported);
let tag_timestamps: Vec<u32> = tags.iter().map(|tag| tag.timestamp).collect();
assert!(
tag_timestamps.windows(2).all(|pair| pair[1] >= pair[0]),
"FLV tag timestamps must not go backwards: {tag_timestamps:?}"
);
let video_frames: Vec<(u32, i32)> = tags
.iter()
.filter(|tag| tag.tag_type == super::TAG_VIDEO && tag.body[1] == super::AVC_NALU)
.map(|tag| (tag.timestamp, super::read_i24(&tag.body[2..5])))
.collect();
assert_eq!(
video_frames.iter().map(|(dts, _)| *dts).collect::<Vec<_>>(),
vec![0, 1, 2, 40],
"video tag timestamps should be authored DTS"
);
assert_eq!(
video_frames
.iter()
.map(|(dts, cts)| i64::from(*dts) + i64::from(*cts))
.collect::<Vec<_>>(),
vec![0, 80, 40, 120],
"composition time should recover the source PTS"
);
}
async fn drain_exporter_chunks(mut exporter: Export, chunks: usize) -> Vec<u8> {
let mut exported = Vec::new();
for _ in 0..chunks {
match tokio::time::timeout(Duration::from_millis(100), exporter.next()).await {
Ok(Ok(Some(chunk))) => exported.extend_from_slice(&chunk),
Ok(Ok(None)) => panic!("exporter ended early"),
Ok(Err(e)) => panic!("exporter error: {e}"),
Err(_) => panic!("exporter timed out"),
}
}
exported
}