use std::{
collections::VecDeque,
ffi::{CStr, c_int},
io::{Read, Seek},
mem,
num::NonZeroI32,
path::Path,
ptr::{addr_of, read_unaligned},
sync::Arc,
};
use derive_more::{IsVariant, TryUnwrap, Unwrap};
use ffmpeg_next::{
Packet, Rational,
ffi::{
AV_DISPOSITION_ATTACHED_PIC, AV_DISPOSITION_TIMED_THUMBNAILS, AV_NOPTS_VALUE, AVDictionary,
AVStream, av_dict_get,
},
format::{self, context::Input},
};
use mediadecode::{
Timebase, Timestamp,
demuxer::{
AttachmentPacket, AttachmentTrackPacket, AttachmentTrackParams, AudioTrackPacket,
AudioTrackParams, DataTrackPacket, DataTrackParams, DemuxedPacket, Demuxer,
SubtitleTrackPacket, SubtitleTrackParams, TrackIndex, TrackInfo, TrackKind, TrackParams,
UnknownTrackParams, VideoTrackPacket, VideoTrackParams,
},
};
use smol_str::SmolStr;
use crate::{
Ffmpeg, boundary,
buffer::PacketBufferError,
codec_id::CodecId,
extras::{AttachmentPacketExtra, TrackExtra},
limits::DemuxLimits,
reader_guard::{GuardedReader, PanicLatch},
sample_format::SampleFormat,
};
fn av_time_base_q() -> Timebase {
Timebase::new(1, NonZeroI32::new(1_000_000).expect("1e6 is non-zero"))
}
pub struct CarrierDemuxer<C: crate::FfmpegCarrier> {
input: Input,
tracks: Vec<TrackInfo<Ffmpeg>>,
pending: VecDeque<(
TrackIndex,
AttachmentPacket<AttachmentPacketExtra, C::Buffer>,
)>,
eof: bool,
reader_panic: Option<Arc<PanicLatch>>,
limits: DemuxLimits,
unconverted: Option<(Packet, crate::buffer::PayloadProvenance)>,
}
impl<C: crate::FfmpegCarrier + crate::CarrierOps> CarrierDemuxer<C> {
pub(crate) fn open_impl<P: AsRef<Path> + ?Sized>(path: &P) -> Result<Self, DemuxError> {
Self::open_with_impl(path, DemuxLimits::default())
}
pub(crate) fn open_with_impl<P: AsRef<Path> + ?Sized>(
path: &P,
limits: DemuxLimits,
) -> Result<Self, DemuxError> {
Self::from_input(
format::input_with_dictionary(path, probe_options(limits))?,
limits,
)
}
pub(crate) fn open_reader_impl<R: Read + Seek + Send + 'static>(
reader: R,
filename: Option<&str>,
) -> Result<Self, DemuxError> {
Self::open_reader_with_impl(reader, filename, DemuxLimits::default())
}
pub(crate) fn open_reader_with_impl<R: Read + Seek + Send + 'static>(
reader: R,
filename: Option<&str>,
limits: DemuxLimits,
) -> Result<Self, DemuxError> {
let (guarded, latch, meter) = GuardedReader::new(reader, limits.max_probe_bytes());
let io = format::context::StreamIo::from_read_seek(guarded)?;
let input =
format::input_from_stream(io, filename, Some(probe_options(limits))).map_err(|e| {
reader_panic(&latch)
.or_else(|| {
meter.tripped().then(|| {
DemuxError::ProbeBudgetExhausted(ProbeBudgetExhausted::new(
meter.read(),
meter.budget(),
))
})
})
.unwrap_or(DemuxError::Ffmpeg(e))
})?;
if meter.tripped() {
return Err(DemuxError::ProbeBudgetExhausted(ProbeBudgetExhausted::new(
meter.read(),
meter.budget(),
)));
}
meter.release();
if let Some(panicked) = reader_panic(&latch) {
return Err(panicked);
}
let mut demuxer = Self::from_input(input, limits)?;
demuxer.reader_panic = Some(latch);
Ok(demuxer)
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub(crate) const fn input_impl(&self) -> &Input {
&self.input
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub(crate) const fn limits_impl(&self) -> DemuxLimits {
self.limits
}
fn from_input(input: Input, limits: DemuxLimits) -> Result<Self, DemuxError> {
let (tracks, pending) = build_tracks::<C>(&input, limits)?;
Ok(Self {
input,
tracks,
pending,
unconverted: None,
eof: false,
reader_panic: None,
limits,
})
}
fn panicked(&self) -> Option<DemuxError> {
self.reader_panic.as_deref().and_then(reader_panic)
}
}
fn probe_options(limits: DemuxLimits) -> ffmpeg_next::Dictionary<'static> {
let mut options = ffmpeg_next::Dictionary::new();
let probe = limits.max_probe_bytes().to_string();
options.set("probesize", &probe);
options.set("formatprobesize", &probe);
options.set("max_streams", &limits.max_streams().to_string());
options
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
#[error("libavformat read {read} bytes probing the container, over a budget of {budget}")]
pub struct ProbeBudgetExhausted {
read: u64,
budget: u64,
}
impl ProbeBudgetExhausted {
#[inline]
pub const fn new(read: u64, budget: u64) -> Self {
Self { read, budget }
}
#[inline]
pub const fn read(&self) -> u64 {
self.read
}
#[inline]
pub const fn budget(&self) -> u64 {
self.budget
}
}
fn reader_panic(latch: &PanicLatch) -> Option<DemuxError> {
latch
.message()
.map(|message| DemuxError::ReaderPanic(ReaderPanic::new(message)))
}
impl<C: crate::FfmpegCarrier + crate::CarrierOps> CarrierDemuxer<C> {
pub(crate) fn tracks_impl(&self) -> &[TrackInfo<Ffmpeg>] {
&self.tracks
}
pub(crate) fn take_tracks_impl(&mut self) -> Vec<TrackInfo<Ffmpeg>> {
mem::take(&mut self.tracks)
}
pub(crate) fn next_packet_impl(
&mut self,
) -> Result<Option<DemuxedPacket<Ffmpeg, C::Buffer>>, DemuxError> {
if let Some(panicked) = self.panicked() {
return Err(panicked);
}
if let Some((track, packet)) = self.pending.pop_front() {
return Ok(Some(DemuxedPacket::Attachment(AttachmentTrackPacket::new(
track, packet,
))));
}
loop {
let (packet, parked_provenance) = match self.unconverted.take() {
Some((packet, provenance)) => (packet, Some(provenance)),
None => {
let mut packet = Packet::empty();
let read = packet.read(&mut self.input);
if let Some(panicked) = self.panicked() {
return Err(panicked);
}
match read {
Ok(()) => {}
Err(ffmpeg_next::Error::Eof) => {
self.eof = true;
return Ok(None);
}
Err(ffmpeg_next::Error::InvalidData) => continue,
Err(e) => return Err(DemuxError::Ffmpeg(e)),
}
(packet, None)
}
};
let index = packet.stream();
let Some(info) = self.tracks.get(index) else {
continue;
};
let track = TrackIndex::new(index);
let time_base = info.timebase();
let provenance = match parked_provenance {
Some(provenance) => provenance,
None if unsafe { is_streams_attached_pic(&self.input, index, &packet) } => {
crate::buffer::PayloadProvenance::AttachedPicture
}
None => crate::buffer::PayloadProvenance::DemuxDelivered,
};
let converted = match info.kind() {
TrackKind::Video => boundary::video_packet_from_borrowed::<C>(
&packet,
time_base,
self.limits.packet(),
provenance,
)
.map(|built| built.map(|p| DemuxedPacket::Video(VideoTrackPacket::new(track, p)))),
TrackKind::Audio => boundary::audio_packet_from_borrowed::<C>(
&packet,
time_base,
self.limits.packet(),
provenance,
)
.map(|built| built.map(|p| DemuxedPacket::Audio(AudioTrackPacket::new(track, p)))),
TrackKind::Subtitle => boundary::subtitle_packet_from_borrowed::<C>(
&packet,
time_base,
self.limits.packet(),
provenance,
)
.map(|built| built.map(|p| DemuxedPacket::Subtitle(SubtitleTrackPacket::new(track, p)))),
TrackKind::Data => boundary::data_packet_from_borrowed::<C>(
&packet,
time_base,
self.limits.packet(),
provenance,
)
.map(|built| built.map(|p| DemuxedPacket::Data(DataTrackPacket::new(track, p)))),
TrackKind::Attachment => continue,
TrackKind::Unknown => continue,
};
let built = match converted {
Ok(built) => built,
Err(source) => {
if source.parks_in_demux() {
self.unconverted = Some((packet, provenance));
}
return Err(DemuxError::PacketBuffer(PacketBuffer::new(index, source)));
}
};
if let Some(out) = built {
return Ok(Some(out));
}
}
}
pub(crate) fn seek_impl(&mut self, target: Timestamp) -> Result<(), DemuxError> {
let ts = target.rescale_to(av_time_base_q()).pts();
if self.eof {
self.input.clear_eof();
self.eof = false;
}
let sought = self.input.seek(ts, ..ts);
if let Some(panicked) = self.panicked() {
return Err(panicked);
}
sought?;
self.unconverted = None;
Ok(())
}
}
macro_rules! demuxer_lane_face {
($($lane:ty),+ $(,)?) => { $(
impl CarrierDemuxer<$lane> {
pub fn open<P: AsRef<Path> + ?Sized>(path: &P) -> Result<Self, DemuxError> {
Self::open_impl(path)
}
pub fn open_with<P: AsRef<Path> + ?Sized>(
path: &P,
limits: DemuxLimits,
) -> Result<Self, DemuxError> {
Self::open_with_impl(path, limits)
}
pub fn open_reader<R: Read + Seek + Send + 'static>(
reader: R,
url: Option<&str>,
) -> Result<Self, DemuxError> {
Self::open_reader_impl(reader, url)
}
pub fn open_reader_with<R: Read + Seek + Send + 'static>(
reader: R,
url: Option<&str>,
limits: DemuxLimits,
) -> Result<Self, DemuxError> {
Self::open_reader_with_impl(reader, url, limits)
}
pub const fn input(&self) -> &Input {
self.input_impl()
}
pub const fn limits(&self) -> DemuxLimits {
self.limits_impl()
}
}
impl Demuxer for CarrierDemuxer<$lane> {
type Adapter = Ffmpeg;
type Buffer = <$lane as crate::FfmpegCarrier>::Buffer;
type Error = DemuxError;
fn tracks(&self) -> &[TrackInfo<Ffmpeg>] {
self.tracks_impl()
}
fn take_tracks(&mut self) -> Vec<TrackInfo<Ffmpeg>> {
self.take_tracks_impl()
}
fn next_packet(
&mut self,
) -> Result<Option<DemuxedPacket<Ffmpeg, Self::Buffer>>, DemuxError> {
self.next_packet_impl()
}
fn seek(&mut self, target: Timestamp) -> Result<(), DemuxError> {
self.seek_impl(target)
}
}
)+ };
}
demuxer_lane_face!(crate::View, crate::Owned);
#[derive(thiserror::Error, Debug, Clone, Copy, PartialEq, Eq)]
#[error(
"the attachment on stream {stream_index} is {bytes} bytes, over the {limit}-byte per-attachment budget"
)]
pub struct AttachmentTooLarge {
stream_index: usize,
bytes: usize,
limit: usize,
}
impl AttachmentTooLarge {
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn new(stream_index: usize, bytes: usize, limit: usize) -> Self {
Self {
stream_index,
bytes,
limit,
}
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn stream_index(&self) -> usize {
self.stream_index
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn bytes(&self) -> usize {
self.bytes
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn limit(&self) -> usize {
self.limit
}
}
#[derive(thiserror::Error, Debug, Clone, Copy, PartialEq, Eq)]
#[error(
"the attachment on stream {stream_index} brings the file's attachments to {total} bytes, over the {limit}-byte budget"
)]
pub struct AttachmentBudgetExhausted {
stream_index: usize,
total: usize,
limit: usize,
}
impl AttachmentBudgetExhausted {
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn new(stream_index: usize, total: usize, limit: usize) -> Self {
Self {
stream_index,
total,
limit,
}
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn stream_index(&self) -> usize {
self.stream_index
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn total(&self) -> usize {
self.total
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn limit(&self) -> usize {
self.limit
}
}
#[derive(thiserror::Error, Debug, Clone, Copy, PartialEq, Eq)]
#[error(
"the codec parameters on stream {stream_index} hold {bytes} heap bytes, over the {limit}-byte budget"
)]
pub struct ParametersTooLarge {
stream_index: usize,
bytes: usize,
limit: usize,
}
impl ParametersTooLarge {
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn new(stream_index: usize, bytes: usize, limit: usize) -> Self {
Self {
stream_index,
bytes,
limit,
}
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn stream_index(&self) -> usize {
self.stream_index
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn bytes(&self) -> usize {
self.bytes
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn limit(&self) -> usize {
self.limit
}
}
#[derive(thiserror::Error, Debug, Clone, Copy, PartialEq, Eq)]
#[error(
"the codec parameters on stream {stream_index} bring the file's to {total} heap bytes, over the {limit}-byte budget"
)]
pub struct ParametersBudgetExhausted {
stream_index: usize,
total: usize,
limit: usize,
}
impl ParametersBudgetExhausted {
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn new(stream_index: usize, total: usize, limit: usize) -> Self {
Self {
stream_index,
total,
limit,
}
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn stream_index(&self) -> usize {
self.stream_index
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn total(&self) -> usize {
self.total
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn limit(&self) -> usize {
self.limit
}
}
#[derive(thiserror::Error, Debug, Clone)]
#[error("the codec parameters for stream {stream_index} were never allocated")]
pub struct ParametersMissing {
stream_index: usize,
}
impl ParametersMissing {
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn new(stream_index: usize) -> Self {
Self { stream_index }
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn stream_index(&self) -> usize {
self.stream_index
}
}
#[derive(thiserror::Error, Debug, Clone)]
#[error("out of memory allocating the codec parameters for stream {stream_index}")]
pub struct ParametersAlloc {
stream_index: usize,
}
impl ParametersAlloc {
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn new(stream_index: usize) -> Self {
Self { stream_index }
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn stream_index(&self) -> usize {
self.stream_index
}
}
#[derive(thiserror::Error, Debug, Clone)]
#[error("the codec parameters for stream {stream_index} could not be copied: {source}")]
pub struct ParametersCopy {
stream_index: usize,
#[source]
source: ffmpeg_next::Error,
}
impl ParametersCopy {
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn new(stream_index: usize, source: ffmpeg_next::Error) -> Self {
Self {
stream_index,
source,
}
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn stream_index(&self) -> usize {
self.stream_index
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn source(&self) -> &ffmpeg_next::Error {
&self.source
}
}
#[derive(thiserror::Error, Debug, Clone)]
#[error("stream {stream_index}: {source}")]
pub struct PacketBuffer {
stream_index: usize,
#[source]
source: PacketBufferError,
}
impl PacketBuffer {
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn new(stream_index: usize, source: PacketBufferError) -> Self {
Self {
stream_index,
source,
}
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn stream_index(&self) -> usize {
self.stream_index
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn source(&self) -> &PacketBufferError {
&self.source
}
}
#[derive(thiserror::Error, Debug, Clone)]
#[error("the reader panicked: {message}")]
pub struct ReaderPanic {
message: SmolStr,
}
impl ReaderPanic {
#[cfg_attr(not(tarpaulin), inline(always))]
pub const fn new(message: SmolStr) -> Self {
Self { message }
}
#[cfg_attr(not(tarpaulin), inline(always))]
pub fn message(&self) -> &str {
self.message.as_str()
}
}
#[derive(thiserror::Error, Debug, Clone, IsVariant, Unwrap, TryUnwrap)]
#[unwrap(ref, ref_mut)]
#[try_unwrap(ref, ref_mut)]
pub enum DemuxError {
#[error(transparent)]
Ffmpeg(#[from] ffmpeg_next::Error),
#[error(transparent)]
ProbeBudgetExhausted(#[from] ProbeBudgetExhausted),
#[error(transparent)]
AttachmentTooLarge(#[from] AttachmentTooLarge),
#[error(transparent)]
AttachmentBudgetExhausted(#[from] AttachmentBudgetExhausted),
#[error(transparent)]
ParametersTooLarge(#[from] ParametersTooLarge),
#[error(transparent)]
ParametersBudgetExhausted(#[from] ParametersBudgetExhausted),
#[error(transparent)]
ParametersMissing(#[from] ParametersMissing),
#[error(transparent)]
ParametersAlloc(#[from] ParametersAlloc),
#[error(transparent)]
ParametersCopy(#[from] ParametersCopy),
#[error(transparent)]
PacketBuffer(#[from] PacketBuffer),
#[error(transparent)]
ReaderPanic(#[from] ReaderPanic),
}
type BuiltTracks<C> = (
Vec<TrackInfo<Ffmpeg>>,
VecDeque<(
TrackIndex,
AttachmentPacket<AttachmentPacketExtra, <C as crate::FfmpegCarrier>::Buffer>,
)>,
);
fn build_tracks<C: crate::FfmpegCarrier + crate::CarrierOps>(
input: &Input,
limits: DemuxLimits,
) -> Result<BuiltTracks<C>, DemuxError> {
admit_streams(input, limits)?;
let count = input.streams().len();
let mut tracks = Vec::with_capacity(count);
let mut pending = VecDeque::new();
for stream in input.streams() {
let index = stream.index();
debug_assert_eq!(
index,
tracks.len(),
"AVStream indices are dense and ordered"
);
let parameters = stream.parameters();
let par = unsafe { parameters.as_ptr() };
let medium = boundary::media_kind_of(¶meters);
let codec =
CodecId::from_raw(unsafe { read_unaligned(addr_of!((*par).codec_id).cast::<i32>()) });
let disposition = unsafe { (*stream.as_ptr()).disposition };
let attached_pic = is_attachment_disposition(disposition);
let time_base = rational_to_timebase(stream.time_base());
let raw_duration = stream.duration();
let duration = (raw_duration != AV_NOPTS_VALUE && raw_duration > 0)
.then(|| Timestamp::new(raw_duration, time_base));
let raw_start = stream.start_time();
let frames = stream.frames();
let params = if attached_pic {
TrackParams::Attachment(AttachmentTrackParams::new(codec))
} else {
match medium {
boundary::MediaKind::Video => TrackParams::Video(VideoTrackParams::new(
codec,
unsafe { (*par).width }.max(0) as u32,
unsafe { (*par).height }.max(0) as u32,
boundary::from_av_pixel_format(unsafe { (*par).format }),
rate_to_timebase(stream.avg_frame_rate()),
)),
boundary::MediaKind::Audio => {
let ch_layout = unsafe { std::ptr::addr_of!((*par).ch_layout) };
let channel_layout =
unsafe { crate::channel_layout::channel_layout_description_from_raw_ptr(ch_layout) };
TrackParams::Audio(AudioTrackParams::new(
codec,
unsafe { (*par).sample_rate }.max(0) as u32,
channel_layout.channels().min(255) as u8,
SampleFormat::from_raw(unsafe { (*par).format }),
channel_layout,
))
}
boundary::MediaKind::Subtitle => TrackParams::Subtitle(SubtitleTrackParams::new(codec)),
boundary::MediaKind::Data => TrackParams::Data(DataTrackParams::new(codec)),
boundary::MediaKind::Attachment => {
TrackParams::Attachment(AttachmentTrackParams::new(codec))
}
boundary::MediaKind::Unknown => TrackParams::Unknown(UnknownTrackParams::new(codec)),
}
};
let extradata_policy = if medium.is_attachment() {
crate::extras::ExtradataPolicy::Omit
} else {
crate::extras::ExtradataPolicy::Copy
};
let parameters_copy = crate::extras::bounded_clone_parameters_with(
¶meters,
index,
limits.max_codec_parameter_bytes(),
extradata_policy,
)?;
let extra = TrackExtra::new(index as i32, parameters_copy)?
.with_disposition(disposition)
.with_start_time((raw_start != AV_NOPTS_VALUE).then_some(raw_start))
.with_frame_count((frames > 0).then_some(frames));
let metadata = unsafe { (*stream.as_ptr()).metadata };
let info = TrackInfo::new(time_base, params, extra)
.with_duration(duration)
.with_filename(unsafe { metadata_text(metadata, c"filename") })
.with_mime_type(unsafe { metadata_text(metadata, c"mimetype") });
if info.kind() == TrackKind::Attachment {
let packet = if attached_pic {
let pkt = unsafe { std::ptr::addr_of!((*stream.as_ptr()).attached_pic) };
unsafe { attached_pic_payload::<C>(pkt, index, limits) }?
} else {
extradata_payload::<C>(&stream, limits)?
};
pending.push_back((TrackIndex::new(index), packet));
}
tracks.push(info);
}
Ok((tracks, pending))
}
unsafe fn is_streams_attached_pic(input: &Input, index: usize, packet: &Packet) -> bool {
let stream = unsafe {
let context = input.as_ptr();
if index >= (*context).nb_streams as usize {
return false;
}
*(*context).streams.add(index)
};
if stream.is_null() {
return false;
}
unsafe { packet_is_parked_picture(stream, packet) }
}
unsafe fn packet_is_parked_picture(stream: *const AVStream, packet: &Packet) -> bool {
use ffmpeg_next::packet::Ref;
unsafe {
let parked = (*stream).attached_pic.buf;
let carried = (*packet.as_ptr()).buf;
if parked.is_null() || carried.is_null() {
return false;
}
(*parked).buffer == (*carried).buffer
}
}
const fn is_attachment_disposition(disposition: c_int) -> bool {
disposition & AV_DISPOSITION_ATTACHED_PIC != 0
&& disposition & AV_DISPOSITION_TIMED_THUMBNAILS == 0
}
const METADATA_VALUE_MAX_BYTES: usize = 64 * 1024;
unsafe fn metadata_text(dict: *const AVDictionary, key: &CStr) -> Option<SmolStr> {
if dict.is_null() {
return None;
}
let entry = unsafe { av_dict_get(dict, key.as_ptr(), std::ptr::null(), 0) };
if entry.is_null() {
return None;
}
let value = unsafe { (*entry).value };
if value.is_null() {
return None;
}
for len in 0..METADATA_VALUE_MAX_BYTES {
if unsafe { *value.add(len).cast::<u8>() } == 0 {
let bytes = unsafe { std::slice::from_raw_parts(value.cast::<u8>(), len) };
return Some(SmolStr::new(std::string::String::from_utf8_lossy(bytes)));
}
}
None
}
unsafe fn attached_pic_payload<C: crate::FfmpegCarrier + crate::CarrierOps>(
pkt: *const ffmpeg_next::ffi::AVPacket,
index: usize,
limits: DemuxLimits,
) -> Result<AttachmentPacket<AttachmentPacketExtra, C::Buffer>, DemuxError> {
let captured = unsafe {
crate::buffer::payload_of::<C>(
pkt,
limits.max_attachment_bytes(),
crate::buffer::PayloadProvenance::AttachedPicture,
)
}
.map_err(|source| DemuxError::PacketBuffer(PacketBuffer::new(index, source)))?;
let extra = AttachmentPacketExtra::new(index as i32);
Ok(match captured {
Some(payload) => {
let flags = unsafe { boundary::md_flags_from_av_packet(pkt) }
.map_err(|source| DemuxError::PacketBuffer(PacketBuffer::new(index, source)))?;
AttachmentPacket::new(payload, extra).with_flags(flags)
}
None => AttachmentPacket::new(C::empty(), extra.with_synthesized(true)),
})
}
fn extradata_payload<C: crate::FfmpegCarrier + crate::CarrierOps>(
stream: &ffmpeg_next::format::stream::Stream<'_>,
limits: DemuxLimits,
) -> Result<AttachmentPacket<AttachmentPacketExtra, C::Buffer>, DemuxError> {
let index = stream.index();
let parameters = stream.parameters();
let par = unsafe { parameters.as_ptr() };
let ptr = unsafe { (*par).extradata };
let len = unsafe { (*par).extradata_size }.max(0) as usize;
if len > limits.max_attachment_bytes() {
return Err(DemuxError::AttachmentTooLarge(AttachmentTooLarge::new(
index,
len,
limits.max_attachment_bytes(),
)));
}
let bytes: &[u8] = if ptr.is_null() || len == 0 {
&[]
} else {
unsafe { std::slice::from_raw_parts(ptr, len) }
};
Ok(AttachmentPacket::new(
C::from_bytes(bytes).ok_or_else(|| {
DemuxError::PacketBuffer(PacketBuffer::new(
index,
crate::buffer::PacketBufferError::CaptureFailed(crate::buffer::CaptureFailed::new(len)),
))
})?,
AttachmentPacketExtra::new(index as i32).with_synthesized(true),
))
}
fn admit_streams(input: &Input, limits: DemuxLimits) -> Result<(), DemuxError> {
let mut attachment_spent: usize = 0;
let mut parameter_spent: usize = 0;
for stream in input.streams() {
let index = stream.index();
let parameters = stream.parameters();
let par = unsafe { parameters.as_ptr() };
if par.is_null() {
return Err(DemuxError::ParametersMissing(ParametersMissing::new(index)));
}
let footprint =
unsafe { crate::extras::measure_parameters(par) }.ok_or(DemuxError::ParametersTooLarge(
ParametersTooLarge::new(index, usize::MAX, limits.max_codec_parameter_bytes()),
))?;
let disposition = unsafe { (*stream.as_ptr()).disposition };
let cover_art = is_attachment_disposition(disposition);
let synthesized = !cover_art && boundary::media_kind_of(¶meters).is_attachment();
let retained_parameters = if synthesized {
footprint.total_without_extradata()
} else {
footprint.total()
}
.ok_or(DemuxError::ParametersTooLarge(ParametersTooLarge::new(
index,
usize::MAX,
limits.max_codec_parameter_bytes(),
)))?;
if retained_parameters > limits.max_codec_parameter_bytes() {
return Err(DemuxError::ParametersTooLarge(ParametersTooLarge::new(
index,
retained_parameters,
limits.max_codec_parameter_bytes(),
)));
}
parameter_spent = parameter_spent.saturating_add(retained_parameters);
if parameter_spent > limits.max_total_codec_parameter_bytes() {
return Err(DemuxError::ParametersBudgetExhausted(
ParametersBudgetExhausted::new(
index,
parameter_spent,
limits.max_total_codec_parameter_bytes(),
),
));
}
let carrier = if cover_art {
unsafe {
let pkt = std::ptr::addr_of!((*stream.as_ptr()).attached_pic);
(*pkt).size
}
.max(0) as usize
} else if synthesized {
footprint.extradata_payload()
} else {
continue;
};
charge_attachment(index, carrier, limits, &mut attachment_spent)?;
}
Ok(())
}
fn charge_attachment(
index: usize,
declared: usize,
limits: DemuxLimits,
spent: &mut usize,
) -> Result<(), DemuxError> {
if declared > limits.max_attachment_bytes() {
return Err(DemuxError::AttachmentTooLarge(AttachmentTooLarge::new(
index,
declared,
limits.max_attachment_bytes(),
)));
}
let total = spent.saturating_add(declared);
if total > limits.max_total_attachment_bytes() {
return Err(DemuxError::AttachmentBudgetExhausted(
AttachmentBudgetExhausted::new(index, total, limits.max_total_attachment_bytes()),
));
}
*spent = total;
Ok(())
}
fn rational_to_timebase(value: Rational) -> Timebase {
Timebase::new(
value.numerator(),
NonZeroI32::new(value.denominator().max(1)).expect("clamped to at least 1"),
)
}
fn rate_to_timebase(value: Rational) -> Option<Timebase> {
let (num, den) = (value.numerator(), value.denominator());
(num > 0 && den > 0).then(|| Timebase::new(num, NonZeroI32::new(den).expect("checked above")))
}
#[cfg(test)]
mod tests {
use ffmpeg_next::ffi::{av_dict_free, av_dict_set};
use ffmpeg_next::codec::Parameters;
use super::*;
use crate::extras::TrackExtra;
fn dict_with(key: &CStr, value: &[u8]) -> *mut AVDictionary {
let mut dict: *mut AVDictionary = std::ptr::null_mut();
let mut terminated = value.to_vec();
terminated.push(0);
let rc = unsafe {
av_dict_set(
&mut dict,
key.as_ptr(),
terminated.as_ptr().cast::<std::ffi::c_char>(),
0,
)
};
assert!(rc >= 0, "av_dict_set failed: {rc}");
dict
}
#[test]
fn metadata_that_is_not_utf8_is_read_lossily_not_unsoundly() {
let raw = b"caf\xE9.ttf".to_vec();
assert!(
std::str::from_utf8(&raw).is_err(),
"the source bytes really are not UTF-8",
);
let dict = dict_with(c"filename", &raw);
let text = unsafe { metadata_text(dict, c"filename") }.expect("the entry exists");
assert_eq!(text.as_str(), "caf\u{FFFD}.ttf");
assert_eq!(unsafe { metadata_text(dict, c"mimetype") }, None);
assert_eq!(
unsafe { metadata_text(std::ptr::null(), c"filename") },
None
);
unsafe { av_dict_free(&mut { dict }) };
}
#[test]
fn valid_metadata_survives_unchanged() {
let dict = dict_with(c"mimetype", b"application/x-truetype-font");
assert_eq!(
unsafe { metadata_text(dict, c"mimetype") }.as_deref(),
Some("application/x-truetype-font"),
);
unsafe { av_dict_free(&mut { dict }) };
}
#[test]
fn an_unterminated_length_is_refused_rather_than_truncated() {
let long = vec![b'a'; METADATA_VALUE_MAX_BYTES + 1];
let dict = dict_with(c"filename", &long);
assert_eq!(unsafe { metadata_text(dict, c"filename") }, None);
unsafe { av_dict_free(&mut { dict }) };
}
struct PanicsWithAHostilePayload;
struct PanicOnDrop;
impl Drop for PanicOnDrop {
fn drop(&mut self) {
panic!("and the payload went too");
}
}
impl std::io::Read for PanicsWithAHostilePayload {
fn read(&mut self, _buf: &mut [u8]) -> std::io::Result<usize> {
std::panic::panic_any(PanicOnDrop);
}
}
impl std::io::Seek for PanicsWithAHostilePayload {
fn seek(&mut self, _pos: std::io::SeekFrom) -> std::io::Result<u64> {
std::panic::panic_any(PanicOnDrop);
}
}
#[test]
fn a_reader_panic_with_a_hostile_payload_does_not_abort_the_process() {
crate::fault_subprocess::in_subprocess(
"demuxer::tests::a_reader_panic_with_a_hostile_payload_does_not_abort_the_process",
|| {
let previous = std::panic::take_hook();
std::panic::set_hook(Box::new(|_| {}));
let opened =
CarrierDemuxer::<crate::Owned>::open_reader(PanicsWithAHostilePayload, Some("x.mkv"));
std::panic::set_hook(previous);
match opened {
Err(DemuxError::ReaderPanic(_)) => {}
Err(other) => panic!("expected ReaderPanic, got {other:?}"),
Ok(_) => panic!("a reader that only panics cannot open a container"),
}
},
);
}
#[test]
fn codec_parameters_that_cannot_be_allocated_are_named() {
crate::fault_subprocess::in_subprocess(
"demuxer::tests::codec_parameters_that_cannot_be_allocated_are_named",
|| {
let source = Parameters::new();
assert!(
!unsafe { source.as_ptr() }.is_null(),
"the source allocates before the cap goes on",
);
crate::fault_subprocess::cap_ffmpeg_allocations(1);
let refused = crate::extras::bounded_clone_parameters(&source, 4, usize::MAX);
crate::fault_subprocess::uncap_ffmpeg_allocations();
assert!(
matches!(
refused,
Err(DemuxError::ParametersAlloc(ref p)) if p.stream_index() == 4
),
"expected ParametersAlloc, got {:?}",
refused.map(|_| ()),
);
crate::extras::bounded_clone_parameters(&source, 4, usize::MAX).expect("an uncapped copy");
},
);
}
#[test]
fn the_public_track_extra_copies_are_checked_too() {
crate::fault_subprocess::in_subprocess(
"demuxer::tests::the_public_track_extra_copies_are_checked_too",
|| {
let source = Parameters::new();
assert!(!unsafe { source.as_ptr() }.is_null(), "allocated uncapped");
let extra = TrackExtra::new(
6,
crate::extras::bounded_clone_parameters(&source, 6, usize::MAX).expect("uncapped"),
)
.expect("real parameters");
crate::fault_subprocess::cap_ffmpeg_allocations(1);
let cloned = extra.try_clone().map(|_| ());
let handed = extra.clone_parameters().map(|_| ());
crate::fault_subprocess::uncap_ffmpeg_allocations();
assert!(
matches!(cloned, Err(DemuxError::ParametersAlloc(ref p)) if p.stream_index() == 6),
"TrackExtra::try_clone: {cloned:?}",
);
assert!(
matches!(handed, Err(DemuxError::ParametersAlloc(ref p)) if p.stream_index() == 6),
"TrackExtra::clone_parameters: {handed:?}",
);
extra.try_clone().expect("an uncapped row copy");
extra.clone_parameters().expect("an uncapped handoff");
},
);
}
#[test]
fn parameters_that_never_allocated_are_refused_at_the_door() {
crate::fault_subprocess::in_subprocess(
"demuxer::tests::parameters_that_never_allocated_are_refused_at_the_door",
|| {
crate::fault_subprocess::cap_ffmpeg_allocations(1);
let never_allocated = Parameters::new();
crate::fault_subprocess::uncap_ffmpeg_allocations();
assert!(
unsafe { never_allocated.as_ptr() }.is_null(),
"the safe constructor really does hand back a null-backed value",
);
let refused = TrackExtra::new(9, never_allocated);
let Err(DemuxError::ParametersMissing(p)) = refused.map(|_| ()) else {
panic!("a null-backed source must not become a track row");
};
assert_eq!(p.stream_index(), 9);
let never_allocated = {
crate::fault_subprocess::cap_ffmpeg_allocations(1);
let p = Parameters::new();
crate::fault_subprocess::uncap_ffmpeg_allocations();
p
};
assert!(matches!(
crate::extras::bounded_clone_parameters(&never_allocated, 9, usize::MAX).map(|_| ()),
Err(DemuxError::ParametersMissing(p)) if p.stream_index() == 9,
));
let real = Parameters::new();
let extra = TrackExtra::new(9, real).expect("real parameters");
extra.try_clone().expect("row copy");
extra.clone_parameters().expect("handoff");
},
);
}
#[cfg(feature = "resample")]
#[test]
fn a_spec_read_from_parameters_that_never_allocated_is_absent() {
crate::fault_subprocess::in_subprocess(
"demuxer::tests::a_spec_read_from_parameters_that_never_allocated_is_absent",
|| {
crate::fault_subprocess::cap_ffmpeg_allocations(1);
let never_allocated = Parameters::new();
crate::fault_subprocess::uncap_ffmpeg_allocations();
assert!(unsafe { never_allocated.as_ptr() }.is_null());
assert_eq!(
crate::ResampleSpec::from_parameters(&never_allocated),
None,
"parameters that do not exist describe no audio",
);
},
);
}
#[test]
fn codec_parameters_whose_copy_fails_are_named() {
crate::fault_subprocess::in_subprocess(
"demuxer::tests::codec_parameters_whose_copy_fails_are_named",
|| {
const EXTRADATA: usize = 8 * 1024 * 1024;
let mut source = Parameters::new();
unsafe {
let par = source.as_mut_ptr();
let extradata = ffmpeg_next::ffi::av_mallocz(EXTRADATA) as *mut u8;
assert!(!extradata.is_null(), "av_mallocz");
(*par).extradata = extradata;
(*par).extradata_size = EXTRADATA as i32;
}
crate::fault_subprocess::cap_ffmpeg_allocations(64 * 1024);
let refused = crate::extras::bounded_clone_parameters(&source, 2, usize::MAX);
crate::fault_subprocess::uncap_ffmpeg_allocations();
match refused {
Err(DemuxError::ParametersCopy(p)) => assert_eq!(p.stream_index(), 2),
Err(other) => panic!("expected ParametersCopy, got {other:?}"),
Ok(_) => panic!("a copy that could not copy the extradata must not succeed"),
}
crate::extras::bounded_clone_parameters(&source, 2, usize::MAX).expect("an uncapped copy");
},
);
}
fn parked_picture_stream(parked: &Packet) -> (Box<AVStream>, Packet) {
use ffmpeg_next::packet::{Mut, Ref};
let mut stream: Box<AVStream> = Box::new(unsafe { std::mem::zeroed() });
let mut queued = Packet::empty();
unsafe {
assert_eq!(
ffmpeg_next::ffi::av_packet_ref(queued.as_mut_ptr(), parked.as_ptr()),
0,
);
stream.attached_pic.buf = (*parked.as_ptr()).buf;
stream.attached_pic.data = (*parked.as_ptr()).data;
stream.attached_pic.size = (*parked.as_ptr()).size;
}
(stream, queued)
}
#[test]
fn a_queued_attached_picture_is_recognised() {
use ffmpeg_next::packet::Ref;
let parked = Packet::copy(&[9u8; 2048]);
let (stream, queued) = parked_picture_stream(&parked);
unsafe {
assert_ne!(
(*queued.as_ptr()).buf,
(*parked.as_ptr()).buf,
"av_packet_ref must mint a new reference struct",
);
}
assert!(unsafe { packet_is_parked_picture(&*stream, &queued) });
let ordinary = Packet::copy(&[1u8; 2048]);
assert!(!unsafe { packet_is_parked_picture(&*stream, &ordinary) });
let bare: Box<AVStream> = Box::new(unsafe { std::mem::zeroed() });
assert!(!unsafe { packet_is_parked_picture(&*bare, &queued) });
}
#[test]
fn the_queued_picture_is_admitted_and_later_packets_take_the_ordinary_road() {
use crate::buffer::{PacketBufferError, PayloadProvenance, payload_of};
use ffmpeg_next::packet::Ref;
let parked = Packet::copy(&[9u8; 2048]);
let (_stream, queued) = parked_picture_stream(&parked);
let parked_buffer = unsafe { (*parked.as_ptr()).buf };
assert!(matches!(
unsafe {
payload_of::<crate::View>(
queued.as_ptr(),
usize::MAX,
PayloadProvenance::CallerSupplied,
)
},
Err(PacketBufferError::SharedPayload(_)),
));
let copied = unsafe {
payload_of::<crate::View>(
queued.as_ptr(),
usize::MAX,
PayloadProvenance::DemuxDelivered,
)
}
.expect("a demux-delivered shared payload is carriable")
.expect("it has a payload");
assert_eq!(copied.as_ref(), &[9u8; 2048][..]);
unsafe {
assert_ne!(
copied.as_ref().as_ptr() as usize,
(*queued.as_ptr()).data as usize,
"a shared demux-delivered payload is copied, not windowed",
);
}
let viewed = unsafe {
payload_of::<crate::View>(
queued.as_ptr(),
usize::MAX,
PayloadProvenance::AttachedPicture,
)
}
.expect("the container's own picture is carriable")
.expect("it has a payload");
assert_eq!(viewed.as_ref(), &[9u8; 2048][..]);
unsafe {
let start = (*parked_buffer).data as usize;
let end = start + (*parked_buffer).size;
let at = viewed.as_ref().as_ptr() as usize;
assert!(
at >= start && at + viewed.len() <= end,
"the queued picture must be viewed, not copied",
);
}
let owned = unsafe {
payload_of::<crate::Owned>(
queued.as_ptr(),
usize::MAX,
PayloadProvenance::AttachedPicture,
)
}
.expect("the owned lane carries it too")
.expect("it has a payload");
assert_eq!(owned.as_ref(), &[9u8; 2048][..]);
let later = Packet::copy(&[4u8; 1024]);
let shared = unsafe {
payload_of::<crate::View>(
later.as_ptr(),
usize::MAX,
PayloadProvenance::DemuxDelivered,
)
}
.expect("an ordinary packet is carriable")
.expect("it has a payload");
unsafe {
assert_eq!(
shared.as_ref().as_ptr() as usize,
(*later.as_ptr()).data as usize,
"a uniquely-referenced packet is still shared, not copied",
);
}
}
#[test]
fn a_timed_thumbnail_stream_is_not_an_attachment() {
assert!(
is_attachment_disposition(AV_DISPOSITION_ATTACHED_PIC),
"a plain attached picture is still an attachment",
);
assert!(
!is_attachment_disposition(AV_DISPOSITION_ATTACHED_PIC | AV_DISPOSITION_TIMED_THUMBNAILS),
"a timed-thumbnail stream is a timed track, whatever else it is flagged",
);
assert!(!is_attachment_disposition(0));
assert!(!is_attachment_disposition(AV_DISPOSITION_TIMED_THUMBNAILS));
assert!(is_attachment_disposition(
AV_DISPOSITION_ATTACHED_PIC | ffmpeg_next::ffi::AV_DISPOSITION_DEFAULT
));
assert!(
ffmpeg_next::format::stream::Disposition::from_bits(AV_DISPOSITION_TIMED_THUMBNAILS)
.is_none(),
"ffmpeg_next mints no TIMED_THUMBNAILS bit — from_bits_truncate would drop it silently",
);
}
#[test]
fn an_uncapturable_cover_still_gets_its_one_packet() {
let empty: ffmpeg_next::ffi::AVPacket = unsafe { std::mem::zeroed() };
let packet = unsafe { attached_pic_payload::<crate::Owned>(&empty, 7, DemuxLimits::default()) }
.expect("an unparked cover is a degenerate track, not an unreadable file");
assert!(packet.data().as_ref().is_empty());
assert!(
packet.extra().synthesized(),
"nothing in the container handed this payload over",
);
assert_eq!(packet.extra().stream_index(), 7);
}
#[test]
fn a_zero_denominator_timebase_is_clamped_not_refused() {
let tb = rational_to_timebase(Rational::new(1, 0));
assert_eq!(tb.den().get(), 1);
assert_eq!(tb.num(), 1);
}
#[test]
fn a_declared_frame_rate_becomes_a_rate_shaped_timebase() {
let ntsc = rate_to_timebase(Rational::new(30_000, 1001)).expect("declared");
assert_eq!((ntsc.num(), ntsc.den().get()), (30_000, 1001));
assert_eq!(
rate_to_timebase(Rational::new(0, 1)),
None,
"0 fps is absent"
);
assert_eq!(
rate_to_timebase(Rational::new(30, 0)),
None,
"no denominator"
);
}
#[test]
fn the_seek_timebase_is_microseconds() {
let tb = av_time_base_q();
assert_eq!((tb.num(), tb.den().get()), (1, 1_000_000));
let target = Timestamp::new(1_500, Timebase::new(1, NonZeroI32::new(1000).expect("ms")));
assert_eq!(target.rescale_to(tb).pts(), 1_500_000);
}
}