use alloc::format;
use alloc::string::String;
use alloc::vec::Vec;
use core::marker::PhantomData;
use broadcast_common::{Package, Parse, Serialize, Unpackage};
use bytes::Bytes;
use rtp_packet::RtpPacket as RtpPacketWire;
use crate::annexb::NAL_LENGTH_SIZE;
use crate::error::{Error, Result};
use crate::media::Media;
use crate::pipeline::CodecConfig;
const RTP_HEADER_LEN: usize = rtp_packet::FIXED_HEADER_LEN;
const RTP_PT_MASK: u8 = 0x7F;
pub const DEFAULT_VIDEO_PT: u8 = 96;
pub const DEFAULT_AUDIO_PT: u8 = 97;
pub const DEFAULT_MTU: usize = 1400;
pub const VIDEO_CLOCK_RATE: u32 = 90_000;
pub const DEFAULT_KLV_PT: u8 = 98;
pub const KLV_ENCODING_NAME: &str = "smpte336m";
const NAL_TYPE_MASK: u8 = 0x1F;
const NAL_FNRI_MASK: u8 = 0xE0;
const NAL_TYPE_STAP_A: u8 = 24;
const NAL_TYPE_FU_A: u8 = 28;
const FU_START_MASK: u8 = 0x80;
const FU_END_MASK: u8 = 0x40;
const STAP_A_SIZE_LEN: usize = 2;
pub const NAL_TYPE_IDR: u8 = 5;
const AAC_SIZE_LENGTH: u32 = 13;
const AAC_INDEX_LENGTH: u32 = 3;
const AAC_INDEX_DELTA_LENGTH: u32 = 3;
const AAC_AU_HEADER_LEN: usize = 2;
const AAC_AU_HEADERS_LENGTH_LEN: usize = 2;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize))]
#[non_exhaustive]
pub enum RtpMediaKind {
H264,
Aac,
}
impl RtpMediaKind {
pub fn name(&self) -> &'static str {
match self {
RtpMediaKind::H264 => "video",
RtpMediaKind::Aac => "audio",
}
}
}
broadcast_common::impl_spec_display!(RtpMediaKind);
#[derive(Debug, Clone)]
pub struct RtpPacket {
pub header: Bytes,
pub payload: Bytes,
}
impl RtpPacket {
pub fn as_contiguous(&self) -> Bytes {
use bytes::BytesMut;
let mut buf = BytesMut::with_capacity(self.header.len() + self.payload.len());
buf.extend_from_slice(&self.header);
buf.extend_from_slice(&self.payload);
buf.freeze()
}
}
#[derive(Debug, Clone)]
pub struct RtpStream {
pub pt: u8,
pub kind: RtpMediaKind,
pub packets: Vec<RtpPacket>,
}
#[derive(Debug, Clone)]
pub struct RtpOutput {
pub streams: Vec<RtpStream>,
pub sdp: String,
}
#[derive(Debug, Clone)]
pub struct RtpPacketiser {
pub mtu: usize,
pub video_pt: u8,
pub audio_pt: u8,
pub ssrc: u32,
pub stap_a_parameter_sets: bool,
}
impl Default for RtpPacketiser {
fn default() -> Self {
Self {
mtu: DEFAULT_MTU,
video_pt: DEFAULT_VIDEO_PT,
audio_pt: DEFAULT_AUDIO_PT,
ssrc: 0x1234_5678,
stap_a_parameter_sets: true,
}
}
}
impl RtpPacketiser {
pub fn new() -> Self {
Self::default()
}
}
struct SeqCounter(u16);
impl SeqCounter {
fn new(start: u16) -> Self {
Self(start)
}
fn next(&mut self) -> u16 {
let v = self.0;
self.0 = self.0.wrapping_add(1);
v
}
}
fn rtp_header(pt: u8, marker: bool, seq: u16, timestamp: u32, ssrc: u32) -> Bytes {
let pkt = RtpPacketWire {
marker,
payload_type: pt & RTP_PT_MASK,
sequence_number: seq,
timestamp,
ssrc,
csrc: Vec::new(),
extension: None,
padding: None,
payload: &[],
};
let len = pkt.serialized_len();
let mut buf = bytes::BytesMut::with_capacity(len);
buf.resize(len, 0);
pkt.serialize_into(&mut buf)
.expect("simple V=2 P=0 X=0 CC=0 header always serializes");
buf.freeze()
}
impl Package for RtpPacketiser {
type Media = Media;
type Output = RtpOutput;
type Error = Error;
fn package(&mut self, media: &Media) -> Result<RtpOutput> {
if media.tracks.is_empty() {
return Err(Error::InvalidInput(
"cannot packetise a Media with no tracks",
));
}
let mut streams = Vec::new();
let mut sdp_media = String::new();
let mut used_video_pt = false;
let mut used_audio_pt = false;
for track in &media.tracks {
match &track.spec.config {
CodecConfig::Avc { config, .. } => {
let pt = if used_video_pt {
self.video_pt.wrapping_add(2)
} else {
used_video_pt = true;
self.video_pt
};
let packets = self.packetise_video(track, pt)?;
streams.push(RtpStream {
pt,
kind: RtpMediaKind::H264,
packets,
});
sdp_media.push_str(&sdp_video(pt, &config.config)?);
}
CodecConfig::Aac {
esds,
channel_count,
sample_rate,
..
} => {
let pt = if used_audio_pt {
self.audio_pt.wrapping_add(2)
} else {
used_audio_pt = true;
self.audio_pt
};
let clock = if track.spec.timescale != 0 {
track.spec.timescale
} else {
*sample_rate
};
let packets = self.packetise_audio(track, pt, clock)?;
streams.push(RtpStream {
pt,
kind: RtpMediaKind::Aac,
packets,
});
let asc = asc_bytes(esds)?;
sdp_media.push_str(&sdp_audio(pt, clock, *channel_count, asc)?);
}
_ => {
return Err(Error::InvalidInput(
"RTP packetiser supports only AVC video and AAC audio tracks",
));
}
}
}
if streams.is_empty() {
return Err(Error::InvalidInput(
"no AVC/AAC tracks to packetise into RTP",
));
}
let sdp = build_sdp(&sdp_media);
Ok(RtpOutput { streams, sdp })
}
}
impl RtpPacketiser {
pub fn packetise_video(&self, track: &crate::media::Track, pt: u8) -> Result<Vec<RtpPacket>> {
let timescale = if track.spec.timescale != 0 {
track.spec.timescale
} else {
VIDEO_CLOCK_RATE
};
let mut packets = Vec::new();
let mut seq = SeqCounter::new(0);
let mut timestamp: u32 = 0;
if self.stap_a_parameter_sets {
if let CodecConfig::Avc { config, .. } = &track.spec.config {
let mut param_nals: Vec<Vec<u8>> = Vec::new();
for sps in &config.config.sps {
param_nals.push(sps.0.clone());
}
for pps in &config.config.pps {
param_nals.push(pps.0.clone());
}
if !param_nals.is_empty() {
let pkt = build_stap_a(pt, ¶m_nals, &mut seq, timestamp, self.ssrc)?;
packets.push(pkt);
}
}
}
for (i, sample) in track.samples.iter().enumerate() {
timestamp = rescale_ts(sample_dts(track, i), timescale, VIDEO_CLOCK_RATE);
let nals = split_length_prefixed(&sample.data)?;
if nals.is_empty() {
continue;
}
let last_nal = nals.len() - 1;
for (n, nal) in nals.iter().enumerate() {
let is_last_nal = n == last_nal;
if nal.len() + RTP_HEADER_LEN <= self.mtu {
let marker = is_last_nal;
let header = rtp_header(pt, marker, seq.next(), timestamp, self.ssrc);
let nal_offset = nal.as_ptr() as usize - sample.data.as_ptr() as usize;
let payload = sample.data.slice(nal_offset..nal_offset + nal.len());
packets.push(RtpPacket { header, payload });
} else {
fragment_fu_a(
nal,
&sample.data,
pt,
is_last_nal,
self.mtu,
&mut seq,
timestamp,
self.ssrc,
&mut packets,
)?;
}
}
}
Ok(packets)
}
fn packetise_audio(
&self,
track: &crate::media::Track,
pt: u8,
clock: u32,
) -> Result<Vec<RtpPacket>> {
let mut packets = Vec::with_capacity(track.samples.len());
let mut seq = SeqCounter::new(0);
let timescale = if track.spec.timescale != 0 {
track.spec.timescale
} else {
clock
};
for (i, sample) in track.samples.iter().enumerate() {
let au = &sample.data;
if au.len() >= (1usize << AAC_SIZE_LENGTH) {
return Err(Error::InvalidValue {
field: "aac_au_size",
value: au.len() as u64,
reason: "exceeds 13-bit AAC-hbr AU-size field",
});
}
let timestamp = rescale_ts(sample_dts(track, i), timescale, clock);
let au_headers_len_bits = (AAC_AU_HEADER_LEN * 8) as u16;
let hdr = (au.len() as u16) << AAC_INDEX_LENGTH;
let rtp_hdr = rtp_header(pt, true, seq.next(), timestamp, self.ssrc);
let mut buf = bytes::BytesMut::with_capacity(
rtp_hdr.len() + AAC_AU_HEADERS_LENGTH_LEN + AAC_AU_HEADER_LEN + au.len(),
);
buf.extend_from_slice(&rtp_hdr);
buf.extend_from_slice(&au_headers_len_bits.to_be_bytes());
buf.extend_from_slice(&hdr.to_be_bytes());
buf.extend_from_slice(au);
let full = buf.freeze();
let header_len = rtp_hdr.len() + AAC_AU_HEADERS_LENGTH_LEN + AAC_AU_HEADER_LEN;
let header = full.slice(0..header_len);
let payload = full.slice(header_len..);
packets.push(RtpPacket { header, payload });
}
Ok(packets)
}
}
fn sample_dts(track: &crate::media::Track, i: usize) -> u64 {
if let (Some(first), Some(cur)) = (
track.samples.first().and_then(|s| s.dts),
track.samples.get(i).and_then(|s| s.dts),
) {
return (cur - first).max(0) as u64;
}
track.samples[..i]
.iter()
.map(|s| s.duration.unwrap_or(0) as u64)
.sum()
}
fn rescale_ts(ticks: u64, from: u32, to: u32) -> u32 {
if from == 0 || from == to {
return ticks as u32;
}
((ticks * to as u64 + from as u64 / 2) / from as u64) as u32
}
fn split_length_prefixed(data: &[u8]) -> Result<Vec<&[u8]>> {
crate::annexb::iter_length_prefixed_nals(data)
}
fn build_stap_a(
pt: u8,
nals: &[Vec<u8>],
seq: &mut SeqCounter,
timestamp: u32,
ssrc: u32,
) -> Result<RtpPacket> {
let mut max_nri = 0u8;
let mut forbidden = 0u8;
for nal in nals {
if let Some(&octet) = nal.first() {
max_nri = max_nri.max(octet & 0x60);
forbidden |= octet & 0x80;
}
}
let stap_hdr = forbidden | max_nri | NAL_TYPE_STAP_A;
let total_nal_bytes: usize = nals.iter().map(|n| n.len() + STAP_A_SIZE_LEN).sum();
let rtp_hdr = rtp_header(pt, false, seq.next(), timestamp, ssrc);
let total = rtp_hdr.len() + 1 + total_nal_bytes;
let mut buf = bytes::BytesMut::with_capacity(total);
buf.extend_from_slice(&rtp_hdr);
buf.extend_from_slice(&[stap_hdr]);
for nal in nals {
if nal.len() > u16::MAX as usize {
return Err(Error::InvalidValue {
field: "stap_a_nal_size",
value: nal.len() as u64,
reason: "exceeds 16-bit STAP-A size prefix",
});
}
buf.extend_from_slice(&(nal.len() as u16).to_be_bytes());
buf.extend_from_slice(nal);
}
let full = buf.freeze();
let header = full.slice(0..rtp_hdr.len());
let payload = full.slice(rtp_hdr.len()..);
Ok(RtpPacket { header, payload })
}
#[allow(clippy::too_many_arguments)]
fn fragment_fu_a(
nal: &[u8],
sample_data: &Bytes,
pt: u8,
au_is_last_nal: bool,
mtu: usize,
seq: &mut SeqCounter,
timestamp: u32,
ssrc: u32,
out: &mut Vec<RtpPacket>,
) -> Result<()> {
if nal.is_empty() {
return Err(Error::InvalidInput("cannot FU-A fragment an empty NAL"));
}
let nal_octet = nal[0];
let fnri = nal_octet & NAL_FNRI_MASK;
let nal_type = nal_octet & NAL_TYPE_MASK;
let fu_indicator = fnri | NAL_TYPE_FU_A;
let payload = &nal[1..];
let per_packet = mtu
.checked_sub(RTP_HEADER_LEN + 2)
.filter(|&b| b > 0)
.ok_or(Error::InvalidInput("MTU too small for FU-A fragmentation"))?;
let base_offset = nal.as_ptr() as usize - sample_data.as_ptr() as usize + 1;
let total = payload.len();
let num_frags = total.div_ceil(per_packet).max(1);
for f in 0..num_frags {
let start = f * per_packet;
let end = (start + per_packet).min(total);
let is_start = f == 0;
let is_end = f == num_frags - 1;
let mut fu_header = nal_type;
if is_start {
fu_header |= FU_START_MASK;
}
if is_end {
fu_header |= FU_END_MASK;
}
let marker = is_end && au_is_last_nal;
let rtp_hdr = rtp_header(pt, marker, seq.next(), timestamp, ssrc);
let mut header_buf = bytes::BytesMut::with_capacity(rtp_hdr.len() + 2);
header_buf.extend_from_slice(&rtp_hdr);
header_buf.extend_from_slice(&[fu_indicator, fu_header]);
let header = header_buf.freeze();
let slice_start = base_offset + start;
let slice_end = base_offset + end;
let payload_slice = sample_data.slice(slice_start..slice_end);
out.push(RtpPacket {
header,
payload: payload_slice,
});
}
Ok(())
}
fn asc_bytes(esds: &crate::mp4esds::EsdsBox) -> Result<&[u8]> {
esds.es_descriptor
.decoder_config
.as_ref()
.and_then(|dc| dc.decoder_specific_info.as_ref())
.map(|dsi| dsi.data.as_slice())
.ok_or(Error::InvalidInput(
"AAC esds has no DecoderSpecificInfo (AudioSpecificConfig)",
))
}
fn build_sdp(media_blocks: &str) -> String {
let mut s = String::new();
s.push_str("v=0\r\n");
s.push_str("o=- 0 0 IN IP4 127.0.0.1\r\n");
s.push_str("s=transmux RTP\r\n");
s.push_str("t=0 0\r\n");
s.push_str(media_blocks);
s
}
fn sdp_video(pt: u8, config: &crate::avc_config::AVCDecoderConfigurationRecord) -> Result<String> {
let profile_level_id = format!(
"{:02X}{:02X}{:02X}",
config.profile_indication, config.profile_compatibility, config.level_indication
);
let mut sprop = String::new();
let mut first = true;
for sps in &config.sps {
if !first {
sprop.push(',');
}
sprop.push_str(&base64_encode(&sps.0));
first = false;
}
for pps in &config.pps {
if !first {
sprop.push(',');
}
sprop.push_str(&base64_encode(&pps.0));
first = false;
}
let mut s = String::new();
s.push_str(&format!("m=video 0 RTP/AVP {pt}\r\n"));
s.push_str(&format!("a=rtpmap:{pt} H264/{VIDEO_CLOCK_RATE}\r\n"));
s.push_str(&format!(
"a=fmtp:{pt} packetization-mode=1; profile-level-id={profile_level_id}; sprop-parameter-sets={sprop}\r\n"
));
Ok(s)
}
fn sdp_audio(pt: u8, clock: u32, channels: u16, asc: &[u8]) -> Result<String> {
let config = hex_encode(asc);
let mut s = String::new();
s.push_str(&format!("m=audio 0 RTP/AVP {pt}\r\n"));
s.push_str(&format!(
"a=rtpmap:{pt} mpeg4-generic/{clock}/{channels}\r\n"
));
s.push_str(&format!(
"a=fmtp:{pt} streamtype=5; profile-level-id=1; mode=AAC-hbr; config={config}; \
sizeLength={AAC_SIZE_LENGTH}; indexLength={AAC_INDEX_LENGTH}; \
indexDeltaLength={AAC_INDEX_DELTA_LENGTH}\r\n"
));
Ok(s)
}
#[derive(Debug, Clone)]
pub struct RtpInputStream {
pub kind: RtpMediaKind,
pub packets: Vec<Vec<u8>>,
}
#[derive(Debug, Clone)]
pub struct RtpInput {
pub streams: Vec<RtpInputStream>,
}
#[derive(Debug, Default, Clone)]
pub struct RtpDepacketiser {
_marker: PhantomData<()>,
}
impl RtpDepacketiser {
pub fn new() -> Self {
Self::default()
}
}
impl Unpackage for RtpDepacketiser {
type Input = RtpInput;
type Media = Media;
type Error = Error;
fn unpackage(&mut self, input: RtpInput) -> Result<Media> {
let mut tracks = Vec::new();
for (idx, stream) in input.streams.iter().enumerate() {
let samples = match stream.kind {
RtpMediaKind::H264 => depacketise_video(&stream.packets)?,
RtpMediaKind::Aac => depacketise_audio(&stream.packets)?,
};
tracks.push(RtpTrack {
id: idx as u32 + 1,
samples,
});
}
Ok(rtp_tracks_to_media(tracks))
}
}
struct RtpTrack {
id: u32,
samples: Vec<ReassembledAu>,
}
const RTP_TS_WRAP: i64 = 1 << 32;
const RTP_TS_WRAP_HALF: i64 = RTP_TS_WRAP / 2;
#[derive(Default)]
struct RtpWrapState {
initialized: bool,
prev_raw: u32,
prev_uw: i64,
}
impl RtpWrapState {
fn push(&mut self, raw: u32) -> i64 {
if !self.initialized {
self.initialized = true;
self.prev_raw = raw;
self.prev_uw = raw as i64;
return self.prev_uw;
}
let mut delta = raw as i64 - self.prev_raw as i64;
if delta > RTP_TS_WRAP_HALF {
delta -= RTP_TS_WRAP; } else if delta < -RTP_TS_WRAP_HALF {
delta += RTP_TS_WRAP; }
let uw = self.prev_uw + delta;
self.prev_raw = raw;
self.prev_uw = uw;
uw
}
}
fn rtp_tracks_to_media(tracks: Vec<RtpTrack>) -> Media {
use crate::pipeline::Sample;
let ir_tracks = tracks
.into_iter()
.map(|t| {
let mut wrap = RtpWrapState::default();
let stamped: Vec<(i64, bool, Vec<u8>)> = t
.samples
.into_iter()
.map(|au| (wrap.push(au.timestamp), au.is_sync, au.data))
.collect();
let n = stamped.len();
let samples: Vec<Sample> = stamped
.iter()
.enumerate()
.map(|(i, &(dts, is_sync, ref data))| {
let duration = if i + 1 < n {
Some((stamped[i + 1].0 - dts).max(0) as u32)
} else if n >= 2 {
Some((dts - stamped[i - 1].0).max(0) as u32)
} else {
None
};
Sample {
data: data.clone().into(),
dts: Some(dts),
pts: Some(dts),
duration,
flags: crate::ir::SampleFlags::new(is_sync),
provenance: None,
}
})
.collect();
let anchor = samples
.first()
.and_then(|s| s.dts)
.map(|d| d.max(0) as u64)
.unwrap_or(0);
crate::media::Track::new_at(placeholder_spec(t.id), samples, anchor)
})
.collect();
Media::new(ir_tracks, 0)
}
fn placeholder_spec(track_id: u32) -> crate::pipeline::TrackSpec {
use crate::avc_config::{AVCConfigurationBox, AVCDecoderConfigurationRecord};
use crate::pipeline::{CodecConfig, TrackSpec};
let record = AVCDecoderConfigurationRecord {
configuration_version: 1,
profile_indication: 0,
profile_compatibility: 0,
level_indication: 0,
length_size_minus_one: (NAL_LENGTH_SIZE - 1) as u8,
sps: Vec::new(),
pps: Vec::new(),
chroma_format: None,
bit_depth_luma_minus8: None,
bit_depth_chroma_minus8: None,
sps_ext: Vec::new(),
};
TrackSpec::new(
track_id,
VIDEO_CLOCK_RATE,
CodecConfig::Avc {
config: AVCConfigurationBox::new(record),
width: 0,
height: 0,
},
)
}
pub(crate) struct ReassembledAu {
pub timestamp: u32,
pub is_sync: bool,
pub data: Vec<u8>,
}
pub(crate) fn reassemble_video(packets: &[Vec<u8>]) -> Result<Vec<ReassembledAu>> {
let mut aus: Vec<ReassembledAu> = Vec::new();
let mut cur_nals: Vec<Vec<u8>> = Vec::new();
let mut cur_ts: Option<u32> = None;
let mut fu_buf: Vec<u8> = Vec::new();
let mut fu_active = false;
fn flush_au(aus: &mut Vec<ReassembledAu>, nals: &mut Vec<Vec<u8>>, ts: u32) {
if nals.is_empty() {
return;
}
let is_sync = nals
.iter()
.any(|n| !n.is_empty() && (n[0] & NAL_TYPE_MASK) == NAL_TYPE_IDR);
aus.push(ReassembledAu {
timestamp: ts,
is_sync,
data: length_prefix_nals(nals),
});
nals.clear();
}
for pkt in packets {
let hdr = parse_rtp_header(pkt)?;
let payload = hdr.payload;
if payload.is_empty() {
continue;
}
if let Some(ts) = cur_ts {
if ts != hdr.timestamp && !cur_nals.is_empty() {
flush_au(&mut aus, &mut cur_nals, ts);
}
}
cur_ts = Some(hdr.timestamp);
let nal_type = payload[0] & NAL_TYPE_MASK;
match nal_type {
NAL_TYPE_STAP_A => {
let mut off = 1usize;
while off < payload.len() {
if off + STAP_A_SIZE_LEN > payload.len() {
return Err(Error::BufferTooShort {
need: off + STAP_A_SIZE_LEN,
have: payload.len(),
what: "STAP-A size prefix",
});
}
let size = u16::from_be_bytes([payload[off], payload[off + 1]]) as usize;
off += STAP_A_SIZE_LEN;
let end = off + size;
if end > payload.len() {
return Err(Error::BufferTooShort {
need: end,
have: payload.len(),
what: "STAP-A NAL",
});
}
cur_nals.push(payload[off..end].to_vec());
off = end;
}
}
NAL_TYPE_FU_A => {
if payload.len() < 2 {
return Err(Error::BufferTooShort {
need: 2,
have: payload.len(),
what: "FU-A header",
});
}
let fu_indicator = payload[0];
let fu_header = payload[1];
let is_start = fu_header & FU_START_MASK != 0;
let is_end = fu_header & FU_END_MASK != 0;
let orig_type = fu_header & NAL_TYPE_MASK;
let fnri = fu_indicator & NAL_FNRI_MASK;
if is_start {
fu_buf.clear();
fu_buf.push(fnri | orig_type);
fu_active = true;
}
if !fu_active {
return Err(Error::InvalidInput("FU-A fragment before start"));
}
fu_buf.extend_from_slice(&payload[2..]);
if is_end {
cur_nals.push(core::mem::take(&mut fu_buf));
fu_active = false;
}
}
_ => cur_nals.push(payload.to_vec()),
}
if hdr.marker && !cur_nals.is_empty() && !fu_active {
let ts = hdr.timestamp;
flush_au(&mut aus, &mut cur_nals, ts);
cur_ts = None;
}
}
if let Some(ts) = cur_ts {
flush_au(&mut aus, &mut cur_nals, ts);
}
Ok(aus)
}
pub(crate) fn reassemble_audio(packets: &[Vec<u8>]) -> Result<Vec<ReassembledAu>> {
let mut aus = Vec::new();
for pkt in packets {
let hdr = parse_rtp_header(pkt)?;
let payload = hdr.payload;
if payload.len() < AAC_AU_HEADERS_LENGTH_LEN {
return Err(Error::BufferTooShort {
need: AAC_AU_HEADERS_LENGTH_LEN,
have: payload.len(),
what: "AAC AU-headers-length",
});
}
let au_headers_len_bits = u16::from_be_bytes([payload[0], payload[1]]) as usize;
let header_bytes = au_headers_len_bits.div_ceil(8);
let num_headers = au_headers_len_bits / (AAC_AU_HEADER_LEN * 8);
let mut off = AAC_AU_HEADERS_LENGTH_LEN;
if off + header_bytes > payload.len() {
return Err(Error::BufferTooShort {
need: off + header_bytes,
have: payload.len(),
what: "AAC AU headers",
});
}
let mut sizes = Vec::with_capacity(num_headers);
for h in 0..num_headers {
let hoff = off + h * AAC_AU_HEADER_LEN;
let ah = u16::from_be_bytes([payload[hoff], payload[hoff + 1]]);
sizes.push((ah >> AAC_INDEX_LENGTH) as usize);
}
off += header_bytes;
for size in sizes {
let end = off + size;
if end > payload.len() {
return Err(Error::BufferTooShort {
need: end,
have: payload.len(),
what: "AAC AU payload",
});
}
aus.push(ReassembledAu {
timestamp: hdr.timestamp,
is_sync: true,
data: payload[off..end].to_vec(),
});
off = end;
}
}
Ok(aus)
}
fn depacketise_video(packets: &[Vec<u8>]) -> Result<Vec<ReassembledAu>> {
reassemble_video(packets)
}
fn length_prefix_nals(nals: &[Vec<u8>]) -> Vec<u8> {
let total: usize = nals.iter().map(|n| NAL_LENGTH_SIZE + n.len()).sum();
let mut out = Vec::with_capacity(total);
for nal in nals {
out.extend_from_slice(&(nal.len() as u32).to_be_bytes());
out.extend_from_slice(nal);
}
out
}
fn depacketise_audio(packets: &[Vec<u8>]) -> Result<Vec<ReassembledAu>> {
reassemble_audio(packets)
}
#[derive(Debug, Clone, Copy)]
pub(crate) struct RtpHeader<'a> {
pub(crate) marker: bool,
#[allow(dead_code)]
payload_type: u8,
pub(crate) sequence: u16,
pub(crate) timestamp: u32,
pub(crate) ssrc: u32,
payload: &'a [u8],
}
pub(crate) fn parse_rtp_header(pkt: &[u8]) -> Result<RtpHeader<'_>> {
let parsed = RtpPacketWire::parse(pkt).map_err(map_rtp_error)?;
Ok(RtpHeader {
marker: parsed.marker,
payload_type: parsed.payload_type,
sequence: parsed.sequence_number,
timestamp: parsed.timestamp,
ssrc: parsed.ssrc,
payload: parsed.payload,
})
}
fn map_rtp_error(e: rtp_packet::Error) -> Error {
match e {
rtp_packet::Error::BufferTooShort { need, have, what } => {
Error::BufferTooShort { need, have, what }
}
rtp_packet::Error::InvalidVersion(v) => Error::InvalidValue {
field: "rtp_version",
value: u64::from(v),
reason: "must be 2",
},
rtp_packet::Error::InvalidValue {
field,
value,
reason,
} => Error::InvalidValue {
field,
value,
reason,
},
rtp_packet::Error::InvalidPadding { count, reason } => Error::InvalidValue {
field: "rtp_padding",
value: u64::from(count),
reason,
},
rtp_packet::Error::ExtensionNotWordAligned { data_len } => Error::InvalidValue {
field: "rtp_extension_length",
value: data_len as u64,
reason: "extension data length is not a multiple of 4 bytes",
},
_ => Error::InvalidInput("invalid RTP header"),
}
}
pub fn packetise_klv(
klv_unit: &Bytes,
pt: u8,
seq_start: u16,
timestamp: u32,
ssrc: u32,
mtu: usize,
) -> Result<Vec<RtpPacket>> {
if klv_unit.is_empty() {
return Err(Error::InvalidInput("cannot packetise an empty KLV unit"));
}
let per_packet = mtu
.checked_sub(RTP_HEADER_LEN)
.filter(|&b| b > 0)
.ok_or(Error::InvalidInput("MTU too small for KLV-over-RTP"))?;
let total = klv_unit.len();
let num_frags = total.div_ceil(per_packet).max(1);
let mut seq = SeqCounter::new(seq_start);
let mut packets = Vec::with_capacity(num_frags);
for f in 0..num_frags {
let start = f * per_packet;
let end = (start + per_packet).min(total);
let is_last = f == num_frags - 1;
let header = rtp_header(pt, is_last, seq.next(), timestamp, ssrc);
let payload = klv_unit.slice(start..end);
packets.push(RtpPacket { header, payload });
}
Ok(packets)
}
pub fn depacketise_klv(packets: &[Vec<u8>]) -> Result<Vec<Vec<u8>>> {
let mut units: Vec<Vec<u8>> = Vec::new();
let mut cur: Vec<u8> = Vec::new();
let mut cur_ts: Option<u32> = None;
for pkt in packets {
let hdr = parse_rtp_header(pkt)?;
if let Some(ts) = cur_ts {
if ts != hdr.timestamp && !cur.is_empty() {
units.push(core::mem::take(&mut cur));
}
}
cur_ts = Some(hdr.timestamp);
cur.extend_from_slice(hdr.payload);
if hdr.marker {
units.push(core::mem::take(&mut cur));
cur_ts = None;
}
}
if !cur.is_empty() {
units.push(cur);
}
Ok(units)
}
const B64_ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
pub fn base64_encode(data: &[u8]) -> String {
let mut out = String::with_capacity(data.len().div_ceil(3) * 4);
for chunk in data.chunks(3) {
let b0 = chunk[0] as u32;
let b1 = *chunk.get(1).unwrap_or(&0) as u32;
let b2 = *chunk.get(2).unwrap_or(&0) as u32;
let n = (b0 << 16) | (b1 << 8) | b2;
out.push(B64_ALPHABET[((n >> 18) & 0x3F) as usize] as char);
out.push(B64_ALPHABET[((n >> 12) & 0x3F) as usize] as char);
if chunk.len() > 1 {
out.push(B64_ALPHABET[((n >> 6) & 0x3F) as usize] as char);
} else {
out.push('=');
}
if chunk.len() > 2 {
out.push(B64_ALPHABET[(n & 0x3F) as usize] as char);
} else {
out.push('=');
}
}
out
}
pub fn base64_decode(s: &str) -> Result<Vec<u8>> {
fn val(c: u8) -> Option<u32> {
match c {
b'A'..=b'Z' => Some((c - b'A') as u32),
b'a'..=b'z' => Some((c - b'a' + 26) as u32),
b'0'..=b'9' => Some((c - b'0' + 52) as u32),
b'+' => Some(62),
b'/' => Some(63),
_ => None,
}
}
let bytes: Vec<u8> = s.bytes().filter(|&b| b != b'=').collect();
let mut out = Vec::with_capacity(bytes.len() * 3 / 4);
let mut acc = 0u32;
let mut nbits = 0u32;
for &b in &bytes {
let v = val(b).ok_or(Error::InvalidValue {
field: "base64",
value: b as u64,
reason: "not a base64 character",
})?;
acc = (acc << 6) | v;
nbits += 6;
if nbits >= 8 {
nbits -= 8;
out.push((acc >> nbits) as u8);
}
}
Ok(out)
}
pub fn hex_encode(data: &[u8]) -> String {
const HEX: &[u8; 16] = b"0123456789abcdef";
let mut out = String::with_capacity(data.len() * 2);
for &b in data {
out.push(HEX[(b >> 4) as usize] as char);
out.push(HEX[(b & 0x0F) as usize] as char);
}
out
}
pub fn hex_decode(s: &str) -> Result<Vec<u8>> {
fn nibble(c: u8) -> Option<u8> {
match c {
b'0'..=b'9' => Some(c - b'0'),
b'a'..=b'f' => Some(c - b'a' + 10),
b'A'..=b'F' => Some(c - b'A' + 10),
_ => None,
}
}
let bytes = s.as_bytes();
if bytes.len() % 2 != 0 {
return Err(Error::InvalidValue {
field: "hex",
value: bytes.len() as u64,
reason: "odd-length hex string",
});
}
let mut out = Vec::with_capacity(bytes.len() / 2);
for pair in bytes.chunks(2) {
let hi = nibble(pair[0]).ok_or(Error::InvalidValue {
field: "hex",
value: pair[0] as u64,
reason: "not a hex digit",
})?;
let lo = nibble(pair[1]).ok_or(Error::InvalidValue {
field: "hex",
value: pair[1] as u64,
reason: "not a hex digit",
})?;
out.push((hi << 4) | lo);
}
Ok(out)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn reassemble_video_reports_timestamp_and_sync() {
fn pkt(seq: u16, ts: u32, marker: bool, nal: &[u8]) -> Vec<u8> {
let mut p = alloc::vec![0x80u8, if marker { 0x80 | 96 } else { 96 }];
p.extend_from_slice(&seq.to_be_bytes());
p.extend_from_slice(&ts.to_be_bytes());
p.extend_from_slice(&[0, 0, 0, 0]); p.extend_from_slice(nal);
p
}
let idr = [0x65u8, 0xAA]; let non = [0x41u8, 0xBB]; let packets = alloc::vec![pkt(1, 1000, true, &idr), pkt(2, 4000, true, &non)];
let aus = reassemble_video(&packets).unwrap();
assert_eq!(aus.len(), 2);
assert_eq!(aus[0].timestamp, 1000);
assert!(aus[0].is_sync, "IDR AU must be sync");
assert_eq!(aus[1].timestamp, 4000);
assert!(!aus[1].is_sync, "non-IDR AU must not be sync");
assert_eq!(&aus[0].data[..4], &[0, 0, 0, 2]);
assert_eq!(&aus[0].data[4..], &idr);
}
#[test]
fn base64_round_trip() {
let data = b"\x67\x42\xc0\x1e\xd9";
let enc = base64_encode(data);
assert_eq!(base64_decode(&enc).unwrap(), data);
}
#[test]
fn base64_known_vector() {
assert_eq!(base64_encode(b"foobar"), "Zm9vYmFy");
assert_eq!(base64_encode(b"fo"), "Zm8=");
assert_eq!(base64_decode("Zm9vYmFy").unwrap(), b"foobar");
}
#[test]
fn hex_round_trip() {
let data = b"\x12\x08\x56\xe5\x00";
let enc = hex_encode(data);
assert_eq!(enc, "12085 6e500".replace(' ', ""));
assert_eq!(hex_decode(&enc).unwrap(), data);
}
#[test]
fn rtp_header_layout() {
let h = rtp_header(96, true, 7, 0x0001_0000, 0xDEAD_BEEF);
assert_eq!(h.len(), RTP_HEADER_LEN);
assert_eq!(h[0], 0x80); assert_eq!(h[1], 0x80 | 96); assert_eq!(u16::from_be_bytes([h[2], h[3]]), 7);
assert_eq!(u32::from_be_bytes([h[4], h[5], h[6], h[7]]), 0x0001_0000);
let parsed = parse_rtp_header(&h).unwrap();
assert!(parsed.marker);
assert_eq!(parsed.payload_type, 96);
}
}