use alloc::string::{String, ToString};
use alloc::vec::Vec;
use core::marker::PhantomData;
use broadcast_common::{Demand, Parse, Stage, Timestamp, Unpackage};
use crate::error::{Error, Result};
use crate::init_segment::{
ChunkLargeOffsetBox, ChunkOffsetBox, MovieBox, SampleSizeBox, SampleToChunkBox, StblChild,
SyncSampleBox, TrackBox,
};
use crate::media::{
Media, SkippedTrack, Track, find_top_box, refine_legacy_config, skipped_track,
track_spec_from_trak,
};
use crate::pipeline::Sample;
use crate::timing::{CompositionOffsetBox, TimeToSampleBox};
#[derive(Debug, Clone)]
pub struct ProgressiveDemux<'a> {
_marker: PhantomData<&'a [u8]>,
buf: Vec<u8>,
media: Option<Media>,
finished: bool,
poisoned: bool,
max_bytes: usize,
}
impl ProgressiveDemux<'_> {
pub fn new(max_bytes: usize) -> Result<Self> {
if max_bytes == 0 {
return Err(Error::InvalidInput(
"ProgressiveDemux::new: max_bytes must be non-zero — a zero cap can never accept \
any bytes and would permanently wedge the Stage",
));
}
Ok(Self {
_marker: PhantomData,
buf: Vec::new(),
media: None,
finished: false,
poisoned: false,
max_bytes,
})
}
fn poison_error(&self) -> Error {
Error::BufferCapExceeded {
what: "ProgressiveDemux Stage buffer",
cap: self.max_bytes,
}
}
}
impl<'a> Unpackage for ProgressiveDemux<'a> {
type Input = &'a [u8];
type Media = Media;
type Error = Error;
fn unpackage(&mut self, input: &'a [u8]) -> Result<Media> {
demux_progressive(input)
}
}
fn demux_progressive(input: &[u8]) -> Result<Media> {
let moov_bytes =
find_top_box(input, b"moov").ok_or(Error::UnexpectedBox { expected: "moov" })?;
let moov = MovieBox::parse(moov_bytes)?;
let movie_timescale = moov.mvhd.timescale;
let mut tracks = Vec::with_capacity(moov.tracks.len());
let mut skipped: Vec<SkippedTrack> = Vec::new();
for trak in &moov.tracks {
let mut spec = match track_spec_from_trak(trak) {
Ok(spec) => spec,
Err(err) => {
skipped.push(skipped_track(err));
continue;
}
};
let samples = match samples_from_stbl(input, trak) {
Ok(samples) => samples,
Err(err) => {
skipped.push(SkippedTrack {
fourcc: String::from("unknown"),
reason: err.to_string(),
});
continue;
}
};
refine_legacy_config(&mut spec.config, &samples);
tracks.push(Track::new(spec, samples));
}
let mut media = Media::new(tracks, movie_timescale);
media.skipped = skipped;
Ok(media)
}
impl Stage for ProgressiveDemux<'_> {
type In<'a> = &'a [u8];
type Out = Media;
type Error = Error;
fn feed(&mut self, input: &[u8], _now: Timestamp) -> Result<()> {
if self.poisoned {
return Err(self.poison_error());
}
if self.finished {
return Err(Error::InvalidInput(
"ProgressiveDemux::feed after finish: the whole-file parse has already run, so \
these bytes would never be parsed",
));
}
let new_len = self.buf.len().saturating_add(input.len());
if new_len > self.max_bytes {
self.poisoned = true;
return Err(self.poison_error());
}
self.buf.extend_from_slice(input);
Ok(())
}
fn poll(&mut self) -> Option<Media> {
self.media.take()
}
fn finish(&mut self) -> Result<()> {
if self.poisoned {
return Err(self.poison_error());
}
if self.finished {
return Ok(());
}
self.finished = true;
let media = demux_progressive(&self.buf);
self.buf = Vec::new();
self.media = Some(media?);
Ok(())
}
fn next_deadline(&self) -> Option<Timestamp> {
None
}
fn on_deadline(&mut self, _now: Timestamp) {}
fn demand(&self) -> Demand {
if self.poisoned || self.finished {
return Demand::saturated();
}
let remaining = self.max_bytes.saturating_sub(self.buf.len());
if remaining == 0 {
Demand::saturated()
} else {
Demand::new(remaining)
}
}
}
fn samples_from_stbl(file: &[u8], trak: &TrackBox) -> Result<Vec<Sample>> {
let stbl = trak
.mdia
.as_ref()
.and_then(|m| m.minf.as_ref())
.and_then(|m| m.stbl.as_ref())
.ok_or(Error::UnexpectedBox { expected: "stbl" })?;
let stts = stbl
.children
.iter()
.find_map(|c| match c {
StblChild::Stts(b) => Some(b),
_ => None,
})
.ok_or(Error::UnexpectedBox { expected: "stts" })?;
let ctts = stbl.children.iter().find_map(|c| match c {
StblChild::Ctts(b) => Some(b),
_ => None,
});
let stss = stbl.children.iter().find_map(|c| match c {
StblChild::Stss(b) => Some(b),
_ => None,
});
let stsz = stbl
.children
.iter()
.find_map(|c| match c {
StblChild::Stsz(b) => Some(b),
_ => None,
})
.ok_or(Error::UnexpectedBox { expected: "stsz" })?;
let stsc = stbl
.children
.iter()
.find_map(|c| match c {
StblChild::Stsc(b) => Some(b),
_ => None,
})
.ok_or(Error::UnexpectedBox { expected: "stsc" })?;
let co64 = stbl.children.iter().find_map(|c| match c {
StblChild::Co64(b) => Some(b),
_ => None,
});
let stco = stbl.children.iter().find_map(|c| match c {
StblChild::Stco(b) => Some(b),
_ => None,
});
let chunk_offsets = chunk_offsets(co64, stco)?;
let samples_per_chunk = expand_stsc(stsc, chunk_offsets.len());
let total_samples: usize = samples_per_chunk.iter().map(|&n| n as usize).sum();
let layout = chunk_layout(&chunk_offsets, &samples_per_chunk, stsz, total_samples)?;
let durations = expand_stts(stts, total_samples)?;
let composition_offsets = expand_ctts(ctts, total_samples)?;
let sync_flags = expand_stss(stss, total_samples);
let mut samples = Vec::with_capacity(total_samples);
let mut next_dts: i64 = 0;
for i in 0..total_samples {
let (start, size) = layout[i];
let end = start
.checked_add(size)
.ok_or(Error::InvalidInput("sample byte range overflow"))?;
if end > file.len() {
return Err(Error::BufferTooShort {
need: end,
have: file.len(),
what: "progressive sample data",
});
}
let dts = next_dts;
samples.push(Sample::new(
file[start..end].to_vec(),
Some(dts),
Some(dts + composition_offsets[i] as i64),
Some(durations[i]),
sync_flags[i],
));
next_dts += durations[i] as i64;
}
Ok(samples)
}
fn chunk_offsets(
co64: Option<&ChunkLargeOffsetBox>,
stco: Option<&ChunkOffsetBox>,
) -> Result<Vec<u64>> {
if let Some(co64) = co64 {
Ok(co64.entries.clone())
} else if let Some(stco) = stco {
Ok(stco.entries.iter().map(|&o| o as u64).collect())
} else {
Err(Error::UnexpectedBox {
expected: "stco or co64",
})
}
}
fn expand_stsc(stsc: &SampleToChunkBox, num_chunks: usize) -> Vec<u32> {
let mut table = alloc::vec![0u32; num_chunks];
for (i, entry) in stsc.entries.iter().enumerate() {
let start = entry.first_chunk as usize;
let end = stsc
.entries
.get(i + 1)
.map(|next| next.first_chunk as usize)
.unwrap_or(num_chunks + 1);
for chunk in start..end {
if chunk >= 1 && chunk <= num_chunks {
table[chunk - 1] = entry.samples_per_chunk;
}
}
}
table
}
fn chunk_layout(
chunk_offsets: &[u64],
samples_per_chunk: &[u32],
stsz: &SampleSizeBox,
total_samples: usize,
) -> Result<Vec<(usize, usize)>> {
let mut layout = Vec::with_capacity(total_samples);
let mut sample_index = 0usize;
for (chunk, &count) in samples_per_chunk.iter().enumerate() {
let mut cursor = chunk_offsets[chunk];
for _ in 0..count {
let size = sample_size(stsz, sample_index)?;
let start = usize::try_from(cursor)
.map_err(|_| Error::InvalidInput("chunk offset exceeds addressable range"))?;
layout.push((start, size));
cursor += size as u64;
sample_index += 1;
}
}
if layout.len() != total_samples {
return Err(Error::InvalidInput(
"stsc-derived sample count does not match chunk layout",
));
}
Ok(layout)
}
fn sample_size(stsz: &SampleSizeBox, index: usize) -> Result<usize> {
if stsz.sample_size != 0 {
Ok(stsz.sample_size as usize)
} else {
stsz.entries
.get(index)
.map(|&s| s as usize)
.ok_or(Error::InvalidInput("stsz has fewer entries than samples"))
}
}
fn expand_stts(stts: &TimeToSampleBox, total_samples: usize) -> Result<Vec<u32>> {
let mut out = Vec::with_capacity(total_samples);
for entry in &stts.entries {
for _ in 0..entry.sample_count {
out.push(entry.sample_delta);
}
}
if out.len() != total_samples {
return Err(Error::InvalidInput(
"stts sample count does not match chunk layout",
));
}
Ok(out)
}
fn expand_ctts(ctts: Option<&CompositionOffsetBox>, total_samples: usize) -> Result<Vec<i32>> {
let Some(ctts) = ctts else {
return Ok(alloc::vec![0i32; total_samples]);
};
let mut out = Vec::with_capacity(total_samples);
for entry in &ctts.entries {
for _ in 0..entry.sample_count {
out.push(entry.sample_offset);
}
}
if out.len() != total_samples {
return Err(Error::InvalidInput(
"ctts sample count does not match chunk layout",
));
}
Ok(out)
}
fn expand_stss(stss: Option<&SyncSampleBox>, total_samples: usize) -> Vec<bool> {
let Some(stss) = stss else {
return alloc::vec![true; total_samples];
};
let mut flags = alloc::vec![false; total_samples];
for &one_based in &stss.entries {
let idx = one_based as usize;
if idx >= 1 && idx <= total_samples {
flags[idx - 1] = true;
}
}
flags
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn new_zero_cap_is_rejected() {
let err =
ProgressiveDemux::new(0).expect_err("a zero cap must be rejected at construction");
assert!(
matches!(err, Error::InvalidInput(_)),
"expected Error::InvalidInput for a zero cap, got {err:?}"
);
}
#[test]
fn new_nonzero_cap_still_works() {
use broadcast_common::{Stage, Timestamp};
let mut demux = ProgressiveDemux::new(16).expect("non-zero cap must construct");
Stage::feed(&mut demux, &[0u8; 4], Timestamp::ZERO).expect("feed under the cap fits");
assert!(
!Stage::demand(&demux).saturated,
"headroom remains under the cap"
);
}
}