From 6df44cfda141bae7a43b30cc8b8be2a44d109ede Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Thu, 18 Jun 2026 17:36:17 +0100 Subject: [PATCH] Add incremental Muninn ADTS media sender --- crates/muninn-daemon/src/media_packetizer.rs | 320 +++++++++++++++++++ 1 file changed, 320 insertions(+) diff --git a/crates/muninn-daemon/src/media_packetizer.rs b/crates/muninn-daemon/src/media_packetizer.rs index 90dae8f..94dd9a4 100644 --- a/crates/muninn-daemon/src/media_packetizer.rs +++ b/crates/muninn-daemon/src/media_packetizer.rs @@ -208,6 +208,21 @@ pub struct AudioPacketWireOptions<'a> { pub source_role: &'a str, } +pub struct AudioAdtsStreamSendConfig { + pub stream_id: String, + pub session_id: String, + pub codec: String, + pub first_packet_id: u64, + pub first_pts_ticks: i64, + pub packet_duration_ticks: u32, + pub timebase_num: u32, + pub timebase_den: u32, + pub deadline_delay_ticks: i64, + pub max_pending_bytes: usize, + pub source_runtime_id: String, + pub source_role: String, +} + pub struct ReceiverFeedbackOptions<'a> { pub stream_id: &'a str, pub session_id: &'a str, @@ -403,6 +418,141 @@ impl VideoAnnexBStreamSendState { } } +pub struct AudioAdtsStreamSendState { + config: AudioAdtsStreamSendConfig, + pending: Vec, + next_packet_id: u64, + next_pts_ticks: i64, +} + +impl AudioAdtsStreamSendState { + pub fn new(config: AudioAdtsStreamSendConfig) -> Result { + if !matches!( + config.codec.trim().to_ascii_lowercase().as_str(), + "aac" | "aac-adts" | "adts" | "audio/aac" + ) { + return Err(anyhow!("ADTS stream sender requires AAC/ADTS codec")); + } + if config.stream_id.is_empty() { + return Err(anyhow!("stream_id must be non-empty")); + } + if config.session_id.is_empty() { + return Err(anyhow!("session_id must be non-empty")); + } + if config.packet_duration_ticks == 0 { + return Err(anyhow!("packet_duration_ticks must be greater than zero")); + } + if config.timebase_num == 0 || config.timebase_den == 0 { + return Err(anyhow!("audio timebase must be non-zero")); + } + if config.deadline_delay_ticks < 0 { + return Err(anyhow!("deadline_delay_ticks must be non-negative")); + } + if config.max_pending_bytes == 0 { + return Err(anyhow!("max_pending_bytes must be greater than zero")); + } + if config.source_runtime_id.is_empty() { + return Err(anyhow!("source_runtime_id must be non-empty")); + } + if config.source_role.is_empty() { + return Err(anyhow!("source_role must be non-empty")); + } + + Ok(Self { + next_packet_id: config.first_packet_id, + next_pts_ticks: config.first_pts_ticks, + config, + pending: Vec::new(), + }) + } + + pub fn push(&mut self, stored_at: &str, bytes: &[u8]) -> Result> { + if !bytes.is_empty() { + self.pending.extend_from_slice(bytes); + } + if self.pending.len() > self.config.max_pending_bytes { + return Err(anyhow!( + "ADTS stream sender pending buffer exceeded {} bytes without a complete packet", + self.config.max_pending_bytes + )); + } + self.emit_available(stored_at) + } + + pub fn finish(&mut self, stored_at: &str) -> Result> { + let payloads = self.emit_available(stored_at)?; + if !self.pending.is_empty() { + return Err(anyhow!( + "ADTS stream sender finished with {} trailing bytes", + self.pending.len() + )); + } + Ok(payloads) + } + + pub fn pending_bytes(&self) -> usize { + self.pending.len() + } + + pub fn next_packet_id(&self) -> u64 { + self.next_packet_id + } + + fn emit_available(&mut self, stored_at: &str) -> Result> { + if stored_at.is_empty() { + return Err(anyhow!("stored_at must be non-empty")); + } + + let frames = complete_adts_frames(&self.pending)?; + if frames.consumed_bytes == 0 { + return Ok(Vec::new()); + } + + let mut payloads = Vec::new(); + for frame in frames.frames { + let deadline_ticks = self + .next_pts_ticks + .checked_add(self.config.deadline_delay_ticks) + .ok_or_else(|| anyhow!("audio deadline_ticks overflow"))?; + payloads.push(audio_packet_send_payload( + AudioPacketWireOptions { + packetize: AudioPacketizeOptions { + stream_id: &self.config.stream_id, + session_id: &self.config.session_id, + codec: &self.config.codec, + packet_id: self.next_packet_id, + pts_ticks: self.next_pts_ticks, + duration_ticks: self.config.packet_duration_ticks, + timebase_num: self.config.timebase_num, + timebase_den: self.config.timebase_den, + deadline_ticks, + }, + stored_at, + source_runtime_id: &self.config.source_runtime_id, + source_role: &self.config.source_role, + }, + &frame, + )?); + self.advance_packet_clock()?; + } + + self.pending.drain(..frames.consumed_bytes); + Ok(payloads) + } + + fn advance_packet_clock(&mut self) -> Result<()> { + self.next_packet_id = self + .next_packet_id + .checked_add(1) + .ok_or_else(|| anyhow!("audio packet_id overflow"))?; + self.next_pts_ticks = self + .next_pts_ticks + .checked_add(i64::from(self.config.packet_duration_ticks)) + .ok_or_else(|| anyhow!("audio pts_ticks overflow"))?; + Ok(()) + } +} + #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] pub struct VideoChunkKey { pub frame_id: u64, @@ -1466,6 +1616,53 @@ fn normalized_video_codec(codec: &str) -> Option<&'static str> { } } +struct CompleteAdtsFrames { + frames: Vec>, + consumed_bytes: usize, +} + +fn complete_adts_frames(input: &[u8]) -> Result { + let mut frames = Vec::new(); + let mut offset = 0_usize; + + while offset < input.len() { + let remaining = &input[offset..]; + if remaining.len() < 2 { + break; + } + if remaining[0] != 0xff || (remaining[1] & 0xf0) != 0xf0 { + return Err(anyhow!("ADTS frame must start with sync word")); + } + if remaining.len() < 7 { + break; + } + + let protection_absent = (remaining[1] & 0x01) != 0; + let header_len = if protection_absent { 7 } else { 9 }; + let frame_len = (((remaining[3] & 0x03) as usize) << 11) + | ((remaining[4] as usize) << 3) + | ((remaining[5] as usize) >> 5); + if frame_len < header_len { + return Err(anyhow!( + "ADTS frame length {frame_len} is shorter than header length {header_len}" + )); + } + if remaining.len() < frame_len { + break; + } + + frames.push(remaining[..frame_len].to_vec()); + offset = offset + .checked_add(frame_len) + .ok_or_else(|| anyhow!("ADTS consumed byte count overflow"))?; + } + + Ok(CompleteAdtsFrames { + frames, + consumed_bytes: offset, + }) +} + fn annex_b_nal_units<'a>( input: &'a [u8], codec_name: &str, @@ -2762,6 +2959,38 @@ mod tests { } } + fn adts_stream_send_config() -> AudioAdtsStreamSendConfig { + AudioAdtsStreamSendConfig { + stream_id: "muninn.raven.av.rudp".to_string(), + session_id: "session-1".to_string(), + codec: "aac-adts".to_string(), + first_packet_id: 12, + first_pts_ticks: 48_000, + packet_duration_ticks: 1_024, + timebase_num: 1, + timebase_den: 48_000, + deadline_delay_ticks: 1_024, + max_pending_bytes: 128, + source_runtime_id: "muninn-test".to_string(), + source_role: "media-test".to_string(), + } + } + + fn adts_frame(payload: &[u8]) -> Vec { + let frame_length = 7 + payload.len(); + let mut frame = vec![ + 0xff, + 0xf1, + 0x4c, + 0x80 | (((frame_length >> 11) & 0x03) as u8), + ((frame_length >> 3) & 0xff) as u8, + (((frame_length & 0x07) << 5) as u8) | 0x1f, + 0xfc, + ]; + frame.extend_from_slice(payload); + frame + } + #[test] fn annex_b_stream_send_state_emits_only_completed_frames() -> Result<()> { let mut first_frame = Vec::new(); @@ -2843,6 +3072,97 @@ mod tests { Ok(()) } + #[test] + fn adts_stream_send_state_emits_complete_packets() -> Result<()> { + let first_frame = adts_frame(&[0x11, 0x22, 0x33]); + let second_frame = adts_frame(&[0x44, 0x55]); + let mut sender = AudioAdtsStreamSendState::new(adts_stream_send_config())?; + + let first_push = sender.push("2026-06-18T00:00:00Z", &first_frame[..4])?; + + assert!(first_push.is_empty()); + assert_eq!(sender.pending_bytes(), 4); + assert_eq!(sender.next_packet_id(), 12); + + let mut tail = first_frame[4..].to_vec(); + tail.extend_from_slice(&second_frame); + let second_push = sender.push("2026-06-18T00:00:00Z", &tail)?; + + assert_eq!(second_push.len(), 2); + assert_eq!(second_push[0].channel_id, MUNINN_MEDIA_RUDP_CHANNEL); + assert_eq!(second_push[1].channel_id, MUNINN_MEDIA_RUDP_CHANNEL); + + let MuninnMediaWireRecord::Audio(first) = + decode_media_wire_record(&second_push[0].payload)? + else { + panic!("expected audio media record"); + }; + assert_eq!(first.packet_id, 12); + assert_eq!(first.pts_ticks, 48_000); + assert_eq!(first.deadline_ticks, 49_024); + assert_eq!(first.codec, "aac-adts"); + assert_eq!(first.payload, first_frame); + + let MuninnMediaWireRecord::Audio(second) = + decode_media_wire_record(&second_push[1].payload)? + else { + panic!("expected audio media record"); + }; + assert_eq!(second.packet_id, 13); + assert_eq!(second.pts_ticks, 49_024); + assert_eq!(second.deadline_ticks, 50_048); + assert_eq!(second.payload, second_frame); + assert_eq!(sender.pending_bytes(), 0); + assert_eq!(sender.next_packet_id(), 14); + Ok(()) + } + + #[test] + fn adts_stream_send_state_rejects_wrong_sync() -> Result<()> { + let mut sender = AudioAdtsStreamSendState::new(adts_stream_send_config())?; + + let error = sender + .push("2026-06-18T00:00:00Z", &[0x47, 0x40, 0x00, 0x10]) + .unwrap_err(); + + assert!( + error + .to_string() + .contains("ADTS frame must start with sync word") + ); + Ok(()) + } + + #[test] + fn adts_stream_send_state_rejects_trailing_bytes_on_finish() -> Result<()> { + let mut sender = AudioAdtsStreamSendState::new(adts_stream_send_config())?; + + let push = sender.push("2026-06-18T00:00:00Z", &[0xff])?; + assert!(push.is_empty()); + + let error = sender.finish("2026-06-18T00:00:00Z").unwrap_err(); + + assert!(error.to_string().contains("trailing bytes")); + Ok(()) + } + + #[test] + fn adts_stream_send_state_rejects_non_adts_codecs() { + let mut config = adts_stream_send_config(); + config.codec = "opus".to_string(); + + let error = match AudioAdtsStreamSendState::new(config) { + Ok(_) => panic!("Opus must not create an ADTS stream sender"), + Err(error) => error, + }; + + assert!( + error + .to_string() + .contains("ADTS stream sender requires AAC/ADTS codec") + ); + } + #[test] fn media_wire_encodes_audio_packet_for_sender() -> Result<()> { let wire = encode_audio_packet_wire_record(