diff --git a/crates/muninn-daemon/src/main.rs b/crates/muninn-daemon/src/main.rs index cba71f0..6263f07 100644 --- a/crates/muninn-daemon/src/main.rs +++ b/crates/muninn-daemon/src/main.rs @@ -12344,33 +12344,40 @@ mod tests { assert_eq!(muninn_rudp_video_gop_frames(&sixty_fps), 15); } + fn send_single_access_unit( + config: crate::media_packetizer::VideoAnnexBStreamSendConfig, + access_unit: &[u8], + ) -> Vec { + let mut sender = crate::media_packetizer::VideoAnnexBStreamSendState::new(config).unwrap(); + let mut payloads = sender.push("2026-06-18T00:00:00Z", access_unit).unwrap(); + payloads.extend(sender.finish("2026-06-18T00:00:00Z").unwrap()); + payloads + } + #[test] fn default_rudp_media_packet_size_keeps_late_typed_video_wire_under_udp_mtu() { let mut access_unit = Vec::new(); access_unit.extend_from_slice(&[0, 0, 0, 1, 0x65]); access_unit.resize(MUNINN_RUDP_MEDIA_PACKET_BYTES, 0x80); - let payloads = crate::media_packetizer::video_annex_b_stream_send_payloads( - crate::media_packetizer::VideoAnnexBStreamWireOptions { - packetize: crate::media_packetizer::VideoAnnexBStreamPacketizeOptions { - stream_id: "muninn.raven.av.rudp", - session_id: "session-1", - codec: "h264", - first_frame_id: 1_000_000, - first_pts_ticks: 3_000_000_000, - frame_duration_ticks: 3_000, - timebase_num: 1, - timebase_den: 90_000, - deadline_delay_ticks: 1_800, - max_payload_bytes: MUNINN_RUDP_MEDIA_PACKET_BYTES, - }, - stored_at: "2026-06-18T00:00:00Z", - source_runtime_id: "muninn-test", - source_role: "media-test", + let payloads = send_single_access_unit( + crate::media_packetizer::VideoAnnexBStreamSendConfig { + stream_id: "muninn.raven.av.rudp".to_string(), + session_id: "session-1".to_string(), + codec: "h264".to_string(), + first_frame_id: 1_000_000, + first_pts_ticks: 3_000_000_000, + frame_duration_ticks: 3_000, + timebase_num: 1, + timebase_den: 90_000, + deadline_delay_ticks: 1_800, + max_payload_bytes: MUNINN_RUDP_MEDIA_PACKET_BYTES, + max_pending_bytes: MUNINN_RUDP_MEDIA_PACKET_BYTES * 2, + source_runtime_id: "muninn-test".to_string(), + source_role: "media-test".to_string(), }, &access_unit, - ) - .unwrap(); + ); assert_eq!(payloads.len(), 1); assert!( @@ -12392,27 +12399,24 @@ mod tests { access_unit.extend_from_slice(&[0, 0, 0, 1, 0x65]); access_unit.resize(MUNINN_RUDP_MEDIA_PACKET_BYTES * 64, 0x80); - let payloads = crate::media_packetizer::video_annex_b_stream_send_payloads( - crate::media_packetizer::VideoAnnexBStreamWireOptions { - packetize: crate::media_packetizer::VideoAnnexBStreamPacketizeOptions { - stream_id: "muninn.raven.av.rudp", - session_id: "raven:2026-06-19T14-56-23Z:video", - codec: "h264", - first_frame_id: 1_000_000, - first_pts_ticks: 3_000_000_000, - frame_duration_ticks: 1_500, - timebase_num: 1, - timebase_den: 90_000, - deadline_delay_ticks: 1_500, - max_payload_bytes: MUNINN_RUDP_MEDIA_PACKET_BYTES, - }, - stored_at: "2026-06-19T14:56:23Z", - source_runtime_id: "raven", - source_role: "muninn.rudp.video", + let payloads = send_single_access_unit( + crate::media_packetizer::VideoAnnexBStreamSendConfig { + stream_id: "muninn.raven.av.rudp".to_string(), + session_id: "raven:2026-06-19T14-56-23Z:video".to_string(), + codec: "h264".to_string(), + first_frame_id: 1_000_000, + first_pts_ticks: 3_000_000_000, + frame_duration_ticks: 1_500, + timebase_num: 1, + timebase_den: 90_000, + deadline_delay_ticks: 1_500, + max_payload_bytes: MUNINN_RUDP_MEDIA_PACKET_BYTES, + max_pending_bytes: MUNINN_RUDP_MEDIA_PACKET_BYTES * 65, + source_runtime_id: "raven".to_string(), + source_role: "muninn.rudp.video".to_string(), }, &access_unit, - ) - .unwrap(); + ); assert!(payloads.len() > 1); let largest = payloads diff --git a/crates/muninn-daemon/src/media_packetizer.rs b/crates/muninn-daemon/src/media_packetizer.rs index 25449b9..ee9aab7 100644 --- a/crates/muninn-daemon/src/media_packetizer.rs +++ b/crates/muninn-daemon/src/media_packetizer.rs @@ -159,26 +159,6 @@ pub struct VideoFramePacketizeOptions<'a> { pub max_payload_bytes: usize, } -pub struct VideoAnnexBStreamPacketizeOptions<'a> { - pub stream_id: &'a str, - pub session_id: &'a str, - pub codec: &'a str, - pub first_frame_id: u64, - pub first_pts_ticks: i64, - pub frame_duration_ticks: u32, - pub timebase_num: u32, - pub timebase_den: u32, - pub deadline_delay_ticks: i64, - pub max_payload_bytes: usize, -} - -pub struct VideoAnnexBStreamWireOptions<'a> { - pub packetize: VideoAnnexBStreamPacketizeOptions<'a>, - pub stored_at: &'a str, - pub source_runtime_id: &'a str, - pub source_role: &'a str, -} - pub struct VideoAnnexBStreamSendConfig { pub stream_id: String, pub session_id: String, @@ -214,7 +194,6 @@ pub struct AudioPacketWireOptions<'a> { pub source_role: &'a str, } - pub struct AudioPcmStreamSendConfig { pub stream_id: String, pub session_id: String, @@ -232,12 +211,6 @@ pub struct AudioPcmStreamSendConfig { pub source_role: String, } - - - - - - #[derive(Clone, Debug, PartialEq, Eq)] pub struct MuninnMediaSendPayload { pub channel_id: &'static str, @@ -402,8 +375,6 @@ impl VideoAnnexBStreamSendState { } } - - pub struct AudioPcmStreamSendState { config: AudioPcmStreamSendConfig, pending: Vec, @@ -552,9 +523,6 @@ impl AudioPcmStreamSendState { } } - - - pub fn packetize_video_access_unit( options: VideoFramePacketizeOptions<'_>, access_unit: &VideoAccessUnit, @@ -753,57 +721,6 @@ fn video_wire_records_with_parity( Ok(wire_records) } -pub fn packetize_video_annex_b_stream( - options: VideoAnnexBStreamPacketizeOptions<'_>, - input: &[u8], -) -> Result> { - if options.frame_duration_ticks == 0 { - return Err(anyhow!("frame_duration_ticks must be greater than zero")); - } - if options.deadline_delay_ticks < 0 { - return Err(anyhow!("deadline_delay_ticks must be non-negative")); - } - - let access_units = video_annex_b_access_units(options.codec, input)?; - let mut records = Vec::new(); - for (index, access_unit) in access_units.iter().enumerate() { - let frame_offset = u64::try_from(index).context("video frame index overflow")?; - let frame_id = options - .first_frame_id - .checked_add(frame_offset) - .ok_or_else(|| anyhow!("video frame_id overflow"))?; - let pts_offset = i64::try_from(index) - .ok() - .and_then(|index| index.checked_mul(i64::from(options.frame_duration_ticks))) - .ok_or_else(|| anyhow!("video pts_ticks overflow"))?; - let pts_ticks = options - .first_pts_ticks - .checked_add(pts_offset) - .ok_or_else(|| anyhow!("video pts_ticks overflow"))?; - let deadline_ticks = pts_ticks - .checked_add(options.deadline_delay_ticks) - .ok_or_else(|| anyhow!("video deadline_ticks overflow"))?; - - records.extend(packetize_video_access_unit( - VideoFramePacketizeOptions { - stream_id: options.stream_id, - session_id: options.session_id, - codec: options.codec, - frame_id, - pts_ticks, - duration_ticks: options.frame_duration_ticks, - timebase_num: options.timebase_num, - timebase_den: options.timebase_den, - deadline_ticks, - max_payload_bytes: options.max_payload_bytes, - }, - access_unit, - )?); - } - - Ok(records) -} - pub fn packetize_audio_packet( options: AudioPacketizeOptions<'_>, payload: &[u8], @@ -844,145 +761,6 @@ pub fn packetize_audio_packet( }) } -#[derive(Default)] -pub struct AudioPacketBuffer { - stream_id: Option, - session_id: Option, - codec: Option, - timebase_num: Option, - timebase_den: Option, - next_packet_id: Option, - emitted_any: bool, - packets: BTreeMap, -} - -impl AudioPacketBuffer { - pub fn insert(&mut self, packet: GameCultMediaAudioPacketRecord) -> Result<()> { - self.require_matching_audio_stream(&packet)?; - if packet.payload.is_empty() { - return Err(anyhow!("audio packet buffer payload must be non-empty")); - } - if self.packets.contains_key(&packet.packet_id) { - return Err(anyhow!( - "audio packet buffer has duplicate packet_id {}", - packet.packet_id - )); - } - match self.next_packet_id { - None => self.next_packet_id = Some(packet.packet_id), - Some(next_packet_id) if packet.packet_id < next_packet_id && !self.emitted_any => { - self.next_packet_id = Some(packet.packet_id); - } - Some(next_packet_id) if packet.packet_id < next_packet_id => { - return Err(anyhow!( - "audio packet buffer received stale packet_id {} before next expected {}", - packet.packet_id, - next_packet_id - )); - } - _ => {} - } - self.packets.insert(packet.packet_id, packet); - Ok(()) - } - - pub fn pop_ready_packets(&mut self) -> Vec { - let Some(mut next_packet_id) = self.next_packet_id else { - return Vec::new(); - }; - let mut ready = Vec::new(); - while let Some(packet) = self.packets.remove(&next_packet_id) { - ready.push(packet); - next_packet_id = next_packet_id.saturating_add(1); - } - if !ready.is_empty() { - self.emitted_any = true; - } - self.next_packet_id = Some(next_packet_id); - ready - } - - pub fn expire_late_packets(&mut self, now_ticks: i64) -> Vec { - let expired_ids = self - .packets - .iter() - .filter_map(|(packet_id, packet)| { - if packet.deadline_ticks <= now_ticks { - Some(*packet_id) - } else { - None - } - }) - .collect::>(); - for packet_id in &expired_ids { - self.packets.remove(packet_id); - } - if let Some(next_packet_id) = self.next_packet_id { - if expired_ids.contains(&next_packet_id) { - self.next_packet_id = self.packets.keys().next().copied(); - } - } - expired_ids - } - - pub fn pending_packet_count(&self) -> usize { - self.packets.len() - } - - fn require_matching_audio_stream( - &mut self, - packet: &GameCultMediaAudioPacketRecord, - ) -> Result<()> { - if packet.stream_id.is_empty() { - return Err(anyhow!("audio packet buffer stream_id must be non-empty")); - } - if packet.session_id.is_empty() { - return Err(anyhow!("audio packet buffer session_id must be non-empty")); - } - if packet.codec.is_empty() { - return Err(anyhow!("audio packet buffer codec must be non-empty")); - } - if packet.timebase_num == 0 || packet.timebase_den == 0 { - return Err(anyhow!("audio packet buffer timebase must be non-zero")); - } - - match ( - self.stream_id.as_deref(), - self.session_id.as_deref(), - self.codec.as_deref(), - self.timebase_num, - self.timebase_den, - ) { - (None, None, None, None, None) => { - self.stream_id = Some(packet.stream_id.clone()); - self.session_id = Some(packet.session_id.clone()); - self.codec = Some(packet.codec.clone()); - self.timebase_num = Some(packet.timebase_num); - self.timebase_den = Some(packet.timebase_den); - Ok(()) - } - ( - Some(stream_id), - Some(session_id), - Some(codec), - Some(timebase_num), - Some(timebase_den), - ) if stream_id == packet.stream_id - && session_id == packet.session_id - && codec == packet.codec - && timebase_num == packet.timebase_num - && timebase_den == packet.timebase_den => - { - Ok(()) - } - _ => Err(anyhow!( - "audio packet buffer received mixed stream metadata" - )), - } - } -} - - /// Muninn's producer name on the wire. The envelope in CultLib takes this as a /// parameter so it never has to know any producer; this is where Muninn says /// who it is, once. @@ -1014,27 +792,6 @@ pub fn encode_media_wire_records( .collect() } -pub fn encode_video_annex_b_stream_wire_records( - options: VideoAnnexBStreamWireOptions<'_>, - input: &[u8], -) -> Result>> { - let records = packetize_video_annex_b_stream(options.packetize, input)?; - let wire_records = video_wire_records_with_parity(&records)?; - encode_media_wire_records( - &wire_records, - options.stored_at, - options.source_runtime_id, - options.source_role, - ) -} - -pub fn video_annex_b_stream_send_payloads( - options: VideoAnnexBStreamWireOptions<'_>, - input: &[u8], -) -> Result> { - encode_video_annex_b_stream_wire_records(options, input).map(wire_payloads_to_media_send) -} - pub fn encode_audio_packet_wire_record( options: AudioPacketWireOptions<'_>, payload: &[u8], @@ -1076,27 +833,6 @@ fn wire_payload_to_media_send(payload: Vec) -> MuninnMediaSendPayload { } } - -fn encode_record_payload(record: &T) -> Result> { - rmp_serde::to_vec(record).map_err(Into::into) -} - - - - -fn decode_record_payload(payload: &[u8]) -> Result { - rmp_serde::from_slice(payload).map_err(Into::into) -} - - - - - - - - - - fn normalized_video_codec(codec: &str) -> Option<&'static str> { match codec.trim().to_ascii_lowercase().as_str() { "h264" | "h.264" | "avc" | "avc1" | "video/avc" => Some("h264"), @@ -1106,8 +842,6 @@ fn normalized_video_codec(codec: &str) -> Option<&'static str> { } } - - fn annex_b_nal_units<'a>( input: &'a [u8], codec_name: &str, @@ -1530,101 +1264,6 @@ mod tests { Ok(()) } - #[test] - fn packetizes_annex_b_stream_into_timed_video_records() -> Result<()> { - let mut stream = Vec::new(); - stream.extend_from_slice(&start_code()); - stream.extend_from_slice(&[0x65, 0x80]); - stream.extend_from_slice(&start_code()); - stream.extend_from_slice(&[0x41, 0x80]); - - let records = packetize_video_annex_b_stream( - VideoAnnexBStreamPacketizeOptions { - stream_id: "muninn.raven.av.rudp", - session_id: "session-1", - codec: "avc", - first_frame_id: 9, - first_pts_ticks: 27_000, - frame_duration_ticks: 3_000, - timebase_num: 1, - timebase_den: 90_000, - deadline_delay_ticks: 1_800, - max_payload_bytes: 16, - }, - &stream, - )?; - - assert_eq!(records.len(), 2); - assert_eq!(records[0].frame_id, 9); - assert_eq!(records[0].pts_ticks, 27_000); - assert_eq!(records[0].deadline_ticks, 28_800); - assert!(records[0].keyframe); - assert_eq!(records[1].frame_id, 10); - assert_eq!(records[1].pts_ticks, 30_000); - assert_eq!(records[1].deadline_ticks, 31_800); - assert_eq!(records[1].dependency_frame_id, Some(9)); - assert_eq!(records[1].codec, "avc"); - Ok(()) - } - - #[test] - fn packetizes_non_reference_h264_p_frames_without_dependency() -> Result<()> { - let mut stream = Vec::new(); - stream.extend_from_slice(&start_code()); - stream.extend_from_slice(&[0x65, 0x80]); - stream.extend_from_slice(&start_code()); - stream.extend_from_slice(&[0x01, 0x80]); - - let records = packetize_video_annex_b_stream( - VideoAnnexBStreamPacketizeOptions { - stream_id: "muninn.raven.av.rudp", - session_id: "session-1", - codec: "h264", - first_frame_id: 9, - first_pts_ticks: 27_000, - frame_duration_ticks: 3_000, - timebase_num: 1, - timebase_den: 90_000, - deadline_delay_ticks: 1_800, - max_payload_bytes: 16, - }, - &stream, - )?; - - assert_eq!(records.len(), 2); - assert!(records[0].keyframe); - assert_eq!(records[0].dependency_frame_id, None); - assert!(!records[1].keyframe); - assert_eq!(records[1].dependency_frame_id, None); - Ok(()) - } - - #[test] - fn rejects_negative_annex_b_stream_deadline_delay() { - let error = packetize_video_annex_b_stream( - VideoAnnexBStreamPacketizeOptions { - stream_id: "muninn.raven.av.rudp", - session_id: "session-1", - codec: "h264", - first_frame_id: 9, - first_pts_ticks: 27_000, - frame_duration_ticks: 3_000, - timebase_num: 1, - timebase_den: 90_000, - deadline_delay_ticks: -1, - max_payload_bytes: 4, - }, - &[0, 0, 0, 1, 0x65, 0x80], - ) - .unwrap_err(); - - assert!( - error - .to_string() - .contains("deadline_delay_ticks must be non-negative") - ); - } - #[test] fn rejects_zero_duration_video_access_units() { let access_unit = VideoAccessUnit { @@ -1799,123 +1438,6 @@ mod tests { ); } - fn audio_packet(packet_id: u64, deadline_ticks: i64) -> GameCultMediaAudioPacketRecord { - packetize_audio_packet( - AudioPacketizeOptions { - stream_id: "muninn.raven.av.rudp", - session_id: "session-1", - codec: "opus", - packet_id, - pts_ticks: (packet_id as i64) * 960, - duration_ticks: 960, - timebase_num: 1, - timebase_den: 48_000, - deadline_ticks, - }, - &[0xf8, 0xff, packet_id as u8], - ) - .unwrap() - } - - #[test] - fn audio_packet_buffer_emits_contiguous_packets_in_order() -> Result<()> { - let mut buffer = AudioPacketBuffer::default(); - - buffer.insert(audio_packet(7, 10_000))?; - buffer.insert(audio_packet(9, 12_000))?; - let ready = buffer.pop_ready_packets(); - - assert_eq!( - ready - .iter() - .map(|packet| packet.packet_id) - .collect::>(), - vec![7] - ); - assert_eq!(buffer.pending_packet_count(), 1); - - buffer.insert(audio_packet(8, 11_000))?; - let ready = buffer.pop_ready_packets(); - - assert_eq!( - ready - .iter() - .map(|packet| packet.packet_id) - .collect::>(), - vec![8, 9] - ); - assert_eq!(buffer.pending_packet_count(), 0); - Ok(()) - } - - #[test] - fn audio_packet_buffer_tracks_lowest_packet_before_first_emit() -> Result<()> { - let mut buffer = AudioPacketBuffer::default(); - - buffer.insert(audio_packet(9, 12_000))?; - buffer.insert(audio_packet(7, 10_000))?; - buffer.insert(audio_packet(8, 11_000))?; - - let ready = buffer.pop_ready_packets(); - - assert_eq!( - ready - .iter() - .map(|packet| packet.packet_id) - .collect::>(), - vec![7, 8, 9] - ); - assert_eq!(buffer.pending_packet_count(), 0); - Ok(()) - } - - #[test] - fn audio_packet_buffer_rejects_stale_packets_after_emit() -> Result<()> { - let mut buffer = AudioPacketBuffer::default(); - - buffer.insert(audio_packet(7, 10_000))?; - assert_eq!(buffer.pop_ready_packets().len(), 1); - let error = buffer.insert(audio_packet(6, 9_000)).unwrap_err(); - - assert!(error.to_string().contains("stale packet_id")); - Ok(()) - } - - #[test] - fn audio_packet_buffer_expires_late_packets() -> Result<()> { - let mut buffer = AudioPacketBuffer::default(); - - buffer.insert(audio_packet(7, 10_000))?; - buffer.insert(audio_packet(8, 12_000))?; - - let expired = buffer.expire_late_packets(10_000); - - assert_eq!(expired, vec![7]); - assert_eq!(buffer.pending_packet_count(), 1); - assert_eq!( - buffer - .pop_ready_packets() - .into_iter() - .map(|packet| packet.packet_id) - .collect::>(), - vec![8] - ); - Ok(()) - } - - #[test] - fn audio_packet_buffer_rejects_mixed_stream_metadata() -> Result<()> { - let mut buffer = AudioPacketBuffer::default(); - let mut mixed = audio_packet(8, 12_000); - mixed.session_id = "session-2".to_string(); - - buffer.insert(audio_packet(7, 10_000))?; - let error = buffer.insert(mixed).unwrap_err(); - - assert!(error.to_string().contains("mixed stream metadata")); - Ok(()) - } - #[test] fn media_wire_round_trips_video_document() -> Result<()> { let access_unit = VideoAccessUnit { @@ -1989,129 +1511,6 @@ mod tests { Ok(()) } - #[test] - fn media_wire_batches_annex_b_video_records() -> Result<()> { - let mut stream = Vec::new(); - stream.extend_from_slice(&start_code()); - stream.extend_from_slice(&[0x65, 0x80]); - stream.extend_from_slice(&start_code()); - stream.extend_from_slice(&[0x41, 0x80]); - let records = packetize_video_annex_b_stream( - VideoAnnexBStreamPacketizeOptions { - stream_id: "muninn.raven.av.rudp", - session_id: "session-1", - codec: "h264", - first_frame_id: 9, - first_pts_ticks: 27_000, - frame_duration_ticks: 3_000, - timebase_num: 1, - timebase_den: 90_000, - deadline_delay_ticks: 1_800, - max_payload_bytes: 16, - }, - &stream, - )?; - let wire_records = records - .iter() - .cloned() - .map(GameCultMediaWireRecord::Video) - .collect::>(); - - let wire = encode_media_wire_records( - &wire_records, - "2026-06-18T00:00:00Z", - "muninn-test", - "media-test", - )?; - - assert_eq!(wire.len(), 2); - assert_eq!(decode_media_wire_record(&wire[0])?, wire_records[0]); - assert_eq!(decode_media_wire_record(&wire[1])?, wire_records[1]); - Ok(()) - } - - #[test] - fn media_wire_encodes_annex_b_stream_for_sender() -> Result<()> { - let mut stream = Vec::new(); - stream.extend_from_slice(&start_code()); - stream.extend_from_slice(&[0x65, 0x80]); - stream.extend_from_slice(&start_code()); - stream.extend_from_slice(&[0x41, 0x80]); - - let wire = encode_video_annex_b_stream_wire_records( - VideoAnnexBStreamWireOptions { - packetize: VideoAnnexBStreamPacketizeOptions { - stream_id: "muninn.raven.av.rudp", - session_id: "session-1", - codec: "h264", - first_frame_id: 9, - first_pts_ticks: 27_000, - frame_duration_ticks: 3_000, - timebase_num: 1, - timebase_den: 90_000, - deadline_delay_ticks: 1_800, - max_payload_bytes: 16, - }, - stored_at: "2026-06-18T00:00:00Z", - source_runtime_id: "muninn-test", - source_role: "media-test", - }, - &stream, - )?; - - assert_eq!(wire.len(), 2); - let first = decode_media_wire_record(&wire[0])?; - let second = decode_media_wire_record(&wire[1])?; - let GameCultMediaWireRecord::Video(first) = first else { - panic!("expected video media record"); - }; - let GameCultMediaWireRecord::Video(second) = second else { - panic!("expected video media record"); - }; - assert_eq!(first.frame_id, 9); - assert_eq!(first.deadline_ticks, 28_800); - assert_eq!(second.frame_id, 10); - assert_eq!(second.deadline_ticks, 31_800); - Ok(()) - } - - #[test] - fn media_send_payloads_pin_video_to_media_channel() -> Result<()> { - let mut stream = Vec::new(); - stream.extend_from_slice(&start_code()); - stream.extend_from_slice(&[0x65, 0x80]); - - let payloads = video_annex_b_stream_send_payloads( - VideoAnnexBStreamWireOptions { - packetize: VideoAnnexBStreamPacketizeOptions { - stream_id: "muninn.raven.av.rudp", - session_id: "session-1", - codec: "h264", - first_frame_id: 9, - first_pts_ticks: 27_000, - frame_duration_ticks: 3_000, - timebase_num: 1, - timebase_den: 90_000, - deadline_delay_ticks: 1_800, - max_payload_bytes: 16, - }, - stored_at: "2026-06-18T00:00:00Z", - source_runtime_id: "muninn-test", - source_role: "media-test", - }, - &stream, - )?; - - assert_eq!(payloads.len(), 1); - assert_eq!(payloads[0].channel_id, MUNINN_MEDIA_RUDP_CHANNEL); - let GameCultMediaWireRecord::Video(record) = decode_media_wire_record(&payloads[0].payload)? - else { - panic!("expected video media record"); - }; - assert_eq!(record.frame_id, 9); - Ok(()) - } - fn stream_send_config() -> VideoAnnexBStreamSendConfig { VideoAnnexBStreamSendConfig { stream_id: "muninn.raven.av.rudp".to_string(), @@ -2130,6 +1529,109 @@ mod tests { } } + /// Push two access units through the sender and return every video data + /// record (parity excluded) it put on the wire, in order. + fn sent_video_records( + config: VideoAnnexBStreamSendConfig, + second_slice_header: u8, + ) -> Result> { + let mut stream = Vec::new(); + stream.extend_from_slice(&start_code()); + stream.extend_from_slice(&[0x65, 0x80]); + stream.extend_from_slice(&start_code()); + stream.extend_from_slice(&[second_slice_header, 0x80]); + let mut sender = VideoAnnexBStreamSendState::new(config)?; + let mut payloads = sender.push("2026-06-18T00:00:00Z", &stream)?; + payloads.extend(sender.finish("2026-06-18T00:00:00Z")?); + let mut records = Vec::new(); + for payload in &payloads { + if let GameCultMediaWireRecord::Video(record) = + decode_media_wire_record(&payload.payload)? + { + records.push(record); + } + } + Ok(records) + } + + #[test] + fn annex_b_stream_send_state_times_records_from_the_configured_clock() -> Result<()> { + let mut config = stream_send_config(); + config.codec = "avc".to_string(); + + let records = sent_video_records(config, 0x41)?; + + assert_eq!(records.len(), 2); + assert_eq!(records[0].frame_id, 9); + assert_eq!(records[0].pts_ticks, 27_000); + assert_eq!(records[0].deadline_ticks, 28_800); + assert!(records[0].keyframe); + assert_eq!(records[1].frame_id, 10); + assert_eq!(records[1].pts_ticks, 30_000); + assert_eq!(records[1].deadline_ticks, 31_800); + assert_eq!(records[1].dependency_frame_id, Some(9)); + assert_eq!(records[1].codec, "avc"); + Ok(()) + } + + #[test] + fn annex_b_stream_send_state_leaves_non_reference_p_frames_without_dependency() -> Result<()> { + let records = sent_video_records(stream_send_config(), 0x01)?; + + assert_eq!(records.len(), 2); + assert!(records[0].keyframe); + assert_eq!(records[0].dependency_frame_id, None); + assert!(!records[1].keyframe); + assert_eq!(records[1].dependency_frame_id, None); + Ok(()) + } + + #[test] + fn annex_b_stream_send_state_rejects_negative_deadline_delay() { + let mut config = stream_send_config(); + config.deadline_delay_ticks = -1; + + let error = match VideoAnnexBStreamSendState::new(config) { + Ok(_) => panic!("a negative deadline delay must not create a sender"), + Err(error) => error, + }; + + assert!( + error + .to_string() + .contains("deadline_delay_ticks must be non-negative") + ); + } + + #[test] + fn media_wire_batches_sender_records_and_reencodes_them_identically() -> Result<()> { + let mut stream = Vec::new(); + stream.extend_from_slice(&start_code()); + stream.extend_from_slice(&[0x65, 0x80]); + stream.extend_from_slice(&start_code()); + stream.extend_from_slice(&[0x41, 0x80]); + let mut sender = VideoAnnexBStreamSendState::new(stream_send_config())?; + let mut payloads = sender.push("2026-06-18T00:00:00Z", &stream)?; + payloads.extend(sender.finish("2026-06-18T00:00:00Z")?); + assert!(payloads.len() >= 2); + let wire_records = payloads + .iter() + .map(|payload| decode_media_wire_record(&payload.payload)) + .collect::>>()?; + + let wire = encode_media_wire_records( + &wire_records, + "2026-06-18T00:00:00Z", + "muninn-test", + "media-test", + )?; + + assert_eq!(wire.len(), payloads.len()); + for (encoded, original) in wire.iter().zip(&payloads) { + assert_eq!(encoded, &original.payload); + } + Ok(()) + } #[test] fn annex_b_stream_send_state_emits_only_completed_frames() -> Result<()> { @@ -2212,10 +1714,6 @@ mod tests { Ok(()) } - - - - #[test] fn media_wire_encodes_audio_packet_for_sender() -> Result<()> { let wire = encode_audio_packet_wire_record( @@ -2464,43 +1962,6 @@ mod tests { Ok(()) } - #[test] - fn feedback_payload_decodes_legacy_record_without_chunk_keys() -> Result<()> { - #[derive(serde::Serialize)] - struct LegacyFeedbackRecord { - stream_id: String, - session_id: String, - receiver_id: String, - highest_decodable_frame_id: Option, - missing_frame_ids: Vec, - late_frame_ids: Vec, - requested_keyframe: bool, - jitter_us: i64, - decode_queue_us: i64, - observed_at: String, - } - - let payload = encode_record_payload(&LegacyFeedbackRecord { - stream_id: "muninn.raven.av.rudp".to_string(), - session_id: "session-1".to_string(), - receiver_id: "starfire.obs".to_string(), - highest_decodable_frame_id: Some(41), - missing_frame_ids: vec![42], - late_frame_ids: vec![40], - requested_keyframe: true, - jitter_us: 750, - decode_queue_us: 2_000, - observed_at: "2026-06-18T00:00:00Z".to_string(), - })?; - - let decoded: GameCultMediaReceiverFeedbackRecord = decode_record_payload(&payload)?; - - assert_eq!(decoded.missing_video_chunk_keys, Vec::::new()); - assert_eq!(decoded.missing_frame_ids, vec![42]); - assert!(decoded.requested_keyframe); - Ok(()) - } - #[test] fn media_wire_rejects_mismatched_record_key() -> Result<()> { let record = packetize_audio_packet(