1//! Custom MoQ-style media objects used by `sigil/media/3`.2//!3//! This wire format borrows MoQ's group, object, publisher-priority, and4//! delivery-timeout semantics. It is intentionally not an implementation of5//! the IETF MoQ Transport protocol.67use std::sync::Arc;89use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};1011use crate::{12 FrameFlags, MAX_MEDIA_PAYLOAD_LEN, MAX_VIDEO_DIMENSION, MAX_VIDEO_PIXELS, MediaCodec,13 ProtocolError, Result,14};1516const MEDIA_OBJECT_V3_MAGIC: [u8; 4] = *b"SGO3";17const MEDIA_CONTROL_V3_MAGIC: [u8; 4] = *b"SGC3";18const MEDIA_V3_VERSION: u16 = 3;1920/// Size of a v3 media-object header in bytes.21pub const MEDIA_OBJECT_V3_HEADER_LEN: usize = 64;22const MEDIA_OBJECT_V3_HEADER_LEN_U16: u16 = MEDIA_OBJECT_V3_HEADER_LEN as u16;23/// Size of a v3 media-control request in bytes.24pub const MEDIA_CONTROL_REQUEST_V3_LEN: usize = 28;25const MEDIA_CONTROL_REQUEST_V3_LEN_U16: u16 = MEDIA_CONTROL_REQUEST_V3_LEN as u16;2627/// Lowest delivery timeout a v3 peer may place on an object.28pub const MIN_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS: u32 = 16;29/// Highest delivery timeout a v3 peer may place on an object.30pub const MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS: u32 = 1_000;31/// Highest object identifier in a v3 group.32pub const MAX_MEDIA_OBJECT_ID_V3: u32 = 255;33/// Maximum aggregate encoded payload retained or accepted for one v3 group.34pub const MAX_MEDIA_GROUP_BYTES_V3: usize = 32 * 1024 * 1024;3536/// Fixed-width header for one independently streamed v3 encoded access unit.37///38/// Integer fields use network byte order. A lower `publisher_priority` value39/// denotes a higher priority. `group_id` identifies one encoded GOP and40/// `object_id` identifies the access unit within that GOP.4142pub struct MediaObjectHeaderV3 {43 pub codec: MediaCodec,44 pub publisher_priority: u8,45 pub flags: FrameFlags,46 pub width: u16,47 pub height: u16,48 pub payload_len: u32,49 pub object_id: u32,50 pub group_id: u64,51 pub sequence: u64,52 pub capture_timestamp_us: u64,53 pub pts_us: i64,54 pub delivery_timeout_ms: u32,55}5657impl MediaObjectHeaderV3 {58 59 pub fn h264(60 width: u16,61 height: u16,62 payload_len: usize,63 publisher_priority: u8,64 flags: FrameFlags,65 object_id: u32,66 group_id: u64,67 sequence: u64,68 capture_timestamp_us: u64,69 pts_us: i64,70 delivery_timeout_ms: u32,71 ) -> Result<Self> {72 let payload_len =73 u32::try_from(payload_len).map_err(|_| ProtocolError::InvalidMediaPayloadLength {74 actual: payload_len,75 maximum: MAX_MEDIA_PAYLOAD_LEN,76 })?;77 let header = Self {78 codec: MediaCodec::H264,79 publisher_priority,80 flags,81 width,82 height,83 payload_len,84 object_id,85 group_id,86 sequence,87 capture_timestamp_us,88 pts_us,89 delivery_timeout_ms,90 };91 header.validate()?;92 Ok(header)93 }9495 pub fn validate(&self) -> Result<()> {96 let pixels = u32::from(self.width) * u32::from(self.height);97 if self.width == 098 || self.height == 099 || self.width > MAX_VIDEO_DIMENSION100 || self.height > MAX_VIDEO_DIMENSION101 || pixels > MAX_VIDEO_PIXELS102 {103 return Err(ProtocolError::InvalidDimensions {104 width: self.width,105 height: self.height,106 });107 }108109 let payload_len = self.payload_len as usize;110 if payload_len == 0 || payload_len > MAX_MEDIA_PAYLOAD_LEN {111 return Err(ProtocolError::InvalidMediaPayloadLength {112 actual: payload_len,113 maximum: MAX_MEDIA_PAYLOAD_LEN,114 });115 }116 if FrameFlags::from_bits(self.flags.bits()).is_none() {117 return Err(ProtocolError::UnsupportedFlags(self.flags.bits()));118 }119 if self.object_id > MAX_MEDIA_OBJECT_ID_V3 {120 return Err(ProtocolError::InvalidMessage {121 message_type: "v3 media object header",122 reason: "object id must be 0..=255",123 });124 }125 if !(MIN_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS..=MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS)126 .contains(&self.delivery_timeout_ms)127 {128 return Err(ProtocolError::InvalidMessage {129 message_type: "v3 media object header",130 reason: "delivery timeout must be 16..=1000 milliseconds",131 });132 }133134 let keyframe = self.flags.contains(FrameFlags::KEYFRAME);135 let codec_config = self.flags.contains(FrameFlags::CODEC_CONFIG);136 if self.object_id == 0 && !(keyframe && codec_config) {137 return Err(ProtocolError::InvalidMessage {138 message_type: "v3 media object header",139 reason: "object zero must be a keyframe carrying codec configuration",140 });141 }142 if self.object_id != 0143 && (keyframe || codec_config || self.flags.contains(FrameFlags::DISCONTINUITY))144 {145 return Err(ProtocolError::InvalidMessage {146 message_type: "v3 media object header",147 reason: "nonzero objects must not carry recovery-boundary flags",148 });149 }150 Ok(())151 }152153 pub fn encode(&self) -> Result<[u8; MEDIA_OBJECT_V3_HEADER_LEN]> {154 self.validate()?;155 let mut wire = [0_u8; MEDIA_OBJECT_V3_HEADER_LEN];156 wire[0..4].copy_from_slice(&MEDIA_OBJECT_V3_MAGIC);157 wire[4..6].copy_from_slice(&MEDIA_V3_VERSION.to_be_bytes());158 wire[6..8].copy_from_slice(&MEDIA_OBJECT_V3_HEADER_LEN_U16.to_be_bytes());159 wire[8] = self.codec as u8;160 wire[9] = self.publisher_priority;161 wire[10] = self.flags.bits();162 // Byte 11 is reserved and must remain zero.163 wire[12..14].copy_from_slice(&self.width.to_be_bytes());164 wire[14..16].copy_from_slice(&self.height.to_be_bytes());165 wire[16..20].copy_from_slice(&self.payload_len.to_be_bytes());166 wire[20..24].copy_from_slice(&self.object_id.to_be_bytes());167 wire[24..32].copy_from_slice(&self.group_id.to_be_bytes());168 wire[32..40].copy_from_slice(&self.sequence.to_be_bytes());169 wire[40..48].copy_from_slice(&self.capture_timestamp_us.to_be_bytes());170 wire[48..56].copy_from_slice(&self.pts_us.to_be_bytes());171 wire[56..60].copy_from_slice(&self.delivery_timeout_ms.to_be_bytes());172 // Bytes 60..64 are reserved and must remain zero.173 Ok(wire)174 }175176 pub fn decode(wire: &[u8; MEDIA_OBJECT_V3_HEADER_LEN]) -> Result<Self> {177 let magic: [u8; 4] = wire[0..4].try_into().expect("fixed slice length");178 if magic != MEDIA_OBJECT_V3_MAGIC {179 return Err(ProtocolError::InvalidMediaMagic(magic));180 }181 let version = u16::from_be_bytes(wire[4..6].try_into().expect("fixed slice length"));182 if version != MEDIA_V3_VERSION {183 return Err(ProtocolError::UnsupportedVersion {184 expected: MEDIA_V3_VERSION,185 actual: version,186 });187 }188 let header_len = u16::from_be_bytes(wire[6..8].try_into().expect("fixed slice length"));189 if header_len != MEDIA_OBJECT_V3_HEADER_LEN_U16 {190 return Err(ProtocolError::InvalidMessage {191 message_type: "v3 media object header",192 reason: "header length must be 64 bytes",193 });194 }195 if wire[11] != 0 || wire[60..64].iter().any(|byte| *byte != 0) {196 return Err(ProtocolError::InvalidMessage {197 message_type: "v3 media object header",198 reason: "reserved fields must be zero",199 });200 }201 let flags =202 FrameFlags::from_bits(wire[10]).ok_or(ProtocolError::UnsupportedFlags(wire[10]))?;203 let header = Self {204 codec: MediaCodec::try_from(wire[8])?,205 publisher_priority: wire[9],206 flags,207 width: u16::from_be_bytes(wire[12..14].try_into().expect("fixed slice length")),208 height: u16::from_be_bytes(wire[14..16].try_into().expect("fixed slice length")),209 payload_len: u32::from_be_bytes(wire[16..20].try_into().expect("fixed slice length")),210 object_id: u32::from_be_bytes(wire[20..24].try_into().expect("fixed slice length")),211 group_id: u64::from_be_bytes(wire[24..32].try_into().expect("fixed slice length")),212 sequence: u64::from_be_bytes(wire[32..40].try_into().expect("fixed slice length")),213 capture_timestamp_us: u64::from_be_bytes(214 wire[40..48].try_into().expect("fixed slice length"),215 ),216 pts_us: i64::from_be_bytes(wire[48..56].try_into().expect("fixed slice length")),217 delivery_timeout_ms: u32::from_be_bytes(218 wire[56..60].try_into().expect("fixed slice length"),219 ),220 };221 header.validate()?;222 Ok(header)223 }224}225226/// One complete v3 media object.227228pub struct MediaObjectV3 {229 pub header: MediaObjectHeaderV3,230 /// The compressed access unit. Held behind an `Arc` so the host can fan the231 /// same encoded object out to every connected client without copying the232 /// payload per connection.233 pub payload: Arc<[u8]>,234}235236impl MediaObjectV3 {237 pub fn new(header: MediaObjectHeaderV3, payload: impl Into<Arc<[u8]>>) -> Result<Self> {238 header.validate()?;239 let payload = payload.into();240 if payload.len() != header.payload_len as usize {241 return Err(ProtocolError::InvalidMessage {242 message_type: "v3 media object",243 reason: "payload does not match declared length",244 });245 }246 Ok(Self { header, payload })247 }248}249250/// Read exactly one complete v3 media object followed by a clean stream EOF.251pub async fn read_media_object_v3<R>(reader: &mut R) -> Result<MediaObjectV3>252where253 R: AsyncRead + Unpin,254{255 let mut wire = [0_u8; MEDIA_OBJECT_V3_HEADER_LEN];256 reader.read_exact(&mut wire).await?;257 let header = MediaObjectHeaderV3::decode(&wire)?;258259 // Header validation, including the peer-controlled payload bound, occurs260 // before this allocation.261 let mut payload = vec![0_u8; header.payload_len as usize];262 reader.read_exact(&mut payload).await?;263 let mut trailing = [0_u8; 1];264 if reader.read(&mut trailing).await? != 0 {265 return Err(ProtocolError::InvalidMessage {266 message_type: "v3 media object",267 reason: "trailing bytes after the media object",268 });269 }270 Ok(MediaObjectV3 {271 header,272 payload: payload.into(),273 })274}275276/// Validate and write one complete v3 media object.277pub async fn write_media_object_v3<W>(writer: &mut W, object: &MediaObjectV3) -> Result<()>278where279 W: AsyncWrite + Unpin,280{281 object.header.validate()?;282 if object.payload.len() != object.header.payload_len as usize {283 return Err(ProtocolError::InvalidMessage {284 message_type: "v3 media object",285 reason: "payload does not match declared length",286 });287 }288 writer.write_all(&object.header.encode()?).await?;289 writer.write_all(&object.payload).await?;290 writer.flush().await?;291 Ok(())292}293294/// V3 media-control request type.295296297pub enum MediaControlRequestTypeV3 {298 RequestKeyframe = 1,299}300301impl TryFrom<u8> for MediaControlRequestTypeV3 {302 type Error = ProtocolError;303304 fn try_from(value: u8) -> Result<Self> {305 match value {306 1 => Ok(Self::RequestKeyframe),307 _ => Err(ProtocolError::InvalidMessage {308 message_type: "v3 media control request",309 reason: "unknown request type",310 }),311 }312 }313}314315/// Why Portal is asking Sigil for a new independently decodable frame.316317318pub enum KeyframeRequestReasonV3 {319 Join = 1,320 TransportGap = 2,321 DeliveryTimeout = 3,322 DecoderReset = 4,323 FrontendBackpressure = 5,324}325326impl TryFrom<u8> for KeyframeRequestReasonV3 {327 type Error = ProtocolError;328329 fn try_from(value: u8) -> Result<Self> {330 match value {331 1 => Ok(Self::Join),332 2 => Ok(Self::TransportGap),333 3 => Ok(Self::DeliveryTimeout),334 4 => Ok(Self::DecoderReset),335 5 => Ok(Self::FrontendBackpressure),336 _ => Err(ProtocolError::InvalidMessage {337 message_type: "v3 media control request",338 reason: "unknown keyframe request reason",339 }),340 }341 }342}343344/// Fixed-width client-to-host keyframe request for the v3 control stream.345346pub struct MediaControlRequestV3 {347 pub request_type: MediaControlRequestTypeV3,348 pub reason: KeyframeRequestReasonV3,349 pub request_id: u64,350 /// Last globally sequenced object accepted by the client, or `None` when351 /// no object has been accepted yet.352 pub last_sequence: Option<u64>,353}354355impl MediaControlRequestV3 {356 pub const fn request_keyframe(357 request_id: u64,358 last_sequence: Option<u64>,359 reason: KeyframeRequestReasonV3,360 ) -> Self {361 Self {362 request_type: MediaControlRequestTypeV3::RequestKeyframe,363 reason,364 request_id,365 last_sequence,366 }367 }368369 pub fn validate(&self) -> Result<()> {370 if self.request_type != MediaControlRequestTypeV3::RequestKeyframe {371 return Err(ProtocolError::InvalidMessage {372 message_type: "v3 media control request",373 reason: "unknown request type",374 });375 }376 if self.last_sequence == Some(u64::MAX) {377 return Err(ProtocolError::InvalidMessage {378 message_type: "v3 media control request",379 reason: "the all-ones last sequence is reserved for none",380 });381 }382 Ok(())383 }384385 pub fn encode(&self) -> Result<[u8; MEDIA_CONTROL_REQUEST_V3_LEN]> {386 self.validate()?;387 let mut wire = [0_u8; MEDIA_CONTROL_REQUEST_V3_LEN];388 wire[0..4].copy_from_slice(&MEDIA_CONTROL_V3_MAGIC);389 wire[4..6].copy_from_slice(&MEDIA_V3_VERSION.to_be_bytes());390 wire[6..8].copy_from_slice(&MEDIA_CONTROL_REQUEST_V3_LEN_U16.to_be_bytes());391 wire[8] = self.request_type as u8;392 wire[9] = self.reason as u8;393 // Bytes 10..12 are reserved and must remain zero.394 wire[12..20].copy_from_slice(&self.request_id.to_be_bytes());395 wire[20..28].copy_from_slice(&self.last_sequence.unwrap_or(u64::MAX).to_be_bytes());396 Ok(wire)397 }398399 pub fn decode(wire: &[u8; MEDIA_CONTROL_REQUEST_V3_LEN]) -> Result<Self> {400 let magic: [u8; 4] = wire[0..4].try_into().expect("fixed slice length");401 if magic != MEDIA_CONTROL_V3_MAGIC {402 return Err(ProtocolError::InvalidMessage {403 message_type: "v3 media control request",404 reason: "invalid control magic",405 });406 }407 let version = u16::from_be_bytes(wire[4..6].try_into().expect("fixed slice length"));408 if version != MEDIA_V3_VERSION {409 return Err(ProtocolError::UnsupportedVersion {410 expected: MEDIA_V3_VERSION,411 actual: version,412 });413 }414 let request_len = u16::from_be_bytes(wire[6..8].try_into().expect("fixed slice length"));415 if request_len != MEDIA_CONTROL_REQUEST_V3_LEN_U16 {416 return Err(ProtocolError::InvalidMessage {417 message_type: "v3 media control request",418 reason: "request length must be 28 bytes",419 });420 }421 if wire[10..12].iter().any(|byte| *byte != 0) {422 return Err(ProtocolError::InvalidMessage {423 message_type: "v3 media control request",424 reason: "reserved fields must be zero",425 });426 }427 let last_sequence =428 u64::from_be_bytes(wire[20..28].try_into().expect("fixed slice length"));429 let request = Self {430 request_type: MediaControlRequestTypeV3::try_from(wire[8])?,431 reason: KeyframeRequestReasonV3::try_from(wire[9])?,432 request_id: u64::from_be_bytes(wire[12..20].try_into().expect("fixed slice length")),433 last_sequence: (last_sequence != u64::MAX).then_some(last_sequence),434 };435 request.validate()?;436 Ok(request)437 }438}439440/// Read one fixed-width v3 media-control request. Clean EOF returns `None`.441pub async fn read_media_control_request_v3<R>(442 reader: &mut R,443) -> Result<Option<MediaControlRequestV3>>444where445 R: AsyncRead + Unpin,446{447 let mut wire = [0_u8; MEDIA_CONTROL_REQUEST_V3_LEN];448 let read = reader.read(&mut wire[..1]).await?;449 if read == 0 {450 return Ok(None);451 }452 reader.read_exact(&mut wire[1..]).await?;453 MediaControlRequestV3::decode(&wire).map(Some)454}455456/// Validate and write one fixed-width v3 media-control request.457pub async fn write_media_control_request_v3<W>(458 writer: &mut W,459 request: &MediaControlRequestV3,460) -> Result<()>461where462 W: AsyncWrite + Unpin,463{464 writer.write_all(&request.encode()?).await?;465 writer.flush().await?;466 Ok(())467}468469470mod tests {471 use std::io::ErrorKind;472473 use tokio::io::{AsyncWriteExt, duplex};474475 use super::*;476477 fn keyframe_header() -> MediaObjectHeaderV3 {478 MediaObjectHeaderV3::h264(479 1280,480 800,481 4,482 32,483 FrameFlags::KEYFRAME484 .union(FrameFlags::CODEC_CONFIG)485 .union(FrameFlags::DISCONTINUITY),486 0,487 0x0102_0304_0506_0708,488 0x1112_1314_1516_1718,489 0x2122_2324_2526_2728,490 -2,491 100,492 )493 .unwrap()494 }495496 497 fn media_object_v3_header_golden_vector() {498 assert_eq!(499 keyframe_header().encode().unwrap(),500 [501 0x53, 0x47, 0x4f, 0x33, // SGO3502 0x00, 0x03, 0x00, 0x40, // version, header length503 0x01, 0x20, 0x07, 0x00, // H.264, priority, flags, reserved504 0x05, 0x00, 0x03, 0x20, // 1280x800505 0x00, 0x00, 0x00, 0x04, // payload length506 0x00, 0x00, 0x00, 0x00, // object id507 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, // group id508 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, // sequence509 0x21, 0x22, 0x23, 0x24, 0x25, 0x26, 0x27, 0x28, // capture timestamp510 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xfe, // PTS511 0x00, 0x00, 0x00, 0x64, // delivery timeout512 0x00, 0x00, 0x00, 0x00, // reserved513 ]514 );515 }516517 518 fn media_object_v3_header_round_trips() {519 let expected = keyframe_header();520 assert_eq!(521 MediaObjectHeaderV3::decode(&expected.encode().unwrap()).unwrap(),522 expected523 );524 }525526 527 fn media_object_v3_malformed_headers_fail_closed() {528 let valid = keyframe_header().encode().unwrap();529 let cases: &[(usize, u8)] = &[530 (0, 0), // magic531 (5, 2), // version532 (7, 63), // header length533 (8, 9), // codec534 (10, 128), // flags535 (11, 1), // reserved536 (12, 0), // zero width537 (59, 15), // delivery timeout below minimum538 (60, 1), // reserved539 ];540 for &(offset, value) in cases {541 let mut malformed = valid;542 malformed[offset] = value;543 assert!(544 MediaObjectHeaderV3::decode(&malformed).is_err(),545 "offset {offset}"546 );547 }548 }549550 551 fn media_object_v3_enforces_group_boundaries_and_limits() {552 let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);553 assert!(554 MediaObjectHeaderV3::h264(555 1280,556 800,557 1,558 0,559 FrameFlags::NONE,560 255,561 1,562 1,563 1,564 1,565 MIN_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS,566 )567 .is_ok()568 );569 assert!(570 MediaObjectHeaderV3::h264(1280, 800, 1, 0, FrameFlags::NONE, 0, 1, 1, 1, 1, 16)571 .is_err()572 );573 assert!(574 MediaObjectHeaderV3::h264(1280, 800, 1, 0, FrameFlags::KEYFRAME, 0, 1, 1, 1, 1, 16)575 .is_err()576 );577 assert!(MediaObjectHeaderV3::h264(1280, 800, 1, 0, keyframe, 256, 1, 1, 1, 1, 16).is_err());578 assert!(579 MediaObjectHeaderV3::h264(1280, 800, 1, 0, FrameFlags::NONE, 1, 1, 2, 2, 2, 16).is_ok()580 );581 assert!(582 MediaObjectHeaderV3::h264(583 1280,584 800,585 1,586 0,587 FrameFlags::DISCONTINUITY,588 1,589 1,590 2,591 2,592 2,593 16,594 )595 .is_err()596 );597 assert!(598 MediaObjectHeaderV3::h264(1280, 800, 1, 0, FrameFlags::NONE, 1, 1, 2, 2, 2, 15)599 .is_err()600 );601 assert!(602 MediaObjectHeaderV3::h264(1280, 800, 1, 0, FrameFlags::NONE, 1, 1, 2, 2, 2, 1_001)603 .is_err()604 );605 assert!(606 MediaObjectHeaderV3::h264(607 1280,608 800,609 1,610 u8::MAX,611 FrameFlags::NONE,612 1,613 1,614 2,615 2,616 2,617 MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS,618 )619 .is_ok()620 );621 assert!(MediaObjectHeaderV3::h264(1280, 800, 0, 0, keyframe, 0, 1, 1, 1, 1, 16).is_err());622 assert!(623 MediaObjectHeaderV3::h264(624 1280,625 800,626 MAX_MEDIA_PAYLOAD_LEN + 1,627 0,628 keyframe,629 0,630 1,631 1,632 1,633 1,634 16635 )636 .is_err()637 );638 }639640 641 async fn media_object_v3_round_trips_with_exact_eof() {642 let expected = MediaObjectV3::new(keyframe_header(), vec![0, 0, 0, 1]).unwrap();643 let (mut sender, mut receiver) = duplex(256);644 write_media_object_v3(&mut sender, &expected).await.unwrap();645 sender.shutdown().await.unwrap();646 assert_eq!(read_media_object_v3(&mut receiver).await.unwrap(), expected);647 }648649 650 async fn media_object_v3_rejects_empty_truncated_and_trailing_streams() {651 let (mut empty_sender, mut empty_receiver) = duplex(16);652 empty_sender.shutdown().await.unwrap();653 assert!(matches!(654 read_media_object_v3(&mut empty_receiver).await,655 Err(ProtocolError::Io(ref error)) if error.kind() == ErrorKind::UnexpectedEof656 ));657658 let mut header = keyframe_header();659 header.payload_len = 2;660 let (mut short_sender, mut short_receiver) = duplex(128);661 short_sender662 .write_all(&header.encode().unwrap())663 .await664 .unwrap();665 short_sender.write_all(&[1]).await.unwrap();666 short_sender.shutdown().await.unwrap();667 assert!(matches!(668 read_media_object_v3(&mut short_receiver).await,669 Err(ProtocolError::Io(ref error)) if error.kind() == ErrorKind::UnexpectedEof670 ));671672 let object = MediaObjectV3::new(keyframe_header(), vec![0, 0, 0, 1]).unwrap();673 let (mut sender, mut receiver) = duplex(256);674 write_media_object_v3(&mut sender, &object).await.unwrap();675 sender.write_all(&[0xff]).await.unwrap();676 sender.shutdown().await.unwrap();677 assert!(matches!(678 read_media_object_v3(&mut receiver).await,679 Err(ProtocolError::InvalidMessage {680 message_type: "v3 media object",681 ..682 })683 ));684 }685686 687 async fn media_object_v3_rejects_oversized_payload_before_body_read() {688 let mut wire = keyframe_header().encode().unwrap();689 wire[16..20].copy_from_slice(&((MAX_MEDIA_PAYLOAD_LEN as u32) + 1).to_be_bytes());690 let (mut sender, mut receiver) = duplex(128);691 sender.write_all(&wire).await.unwrap();692 assert!(matches!(693 read_media_object_v3(&mut receiver).await,694 Err(ProtocolError::InvalidMediaPayloadLength { .. })695 ));696 }697698 fn control_request(last_sequence: Option<u64>) -> MediaControlRequestV3 {699 MediaControlRequestV3::request_keyframe(700 0x0102_0304_0506_0708,701 last_sequence,702 KeyframeRequestReasonV3::FrontendBackpressure,703 )704 }705706 707 fn media_control_request_v3_golden_vector() {708 assert_eq!(709 control_request(Some(0x1112_1314_1516_1718))710 .encode()711 .unwrap(),712 [713 0x53, 0x47, 0x43, 0x33, // SGC3714 0x00, 0x03, 0x00, 0x1c, // version, request length715 0x01, 0x05, 0x00, 0x00, // type, reason, reserved716 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, // request id717 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, // last sequence718 ]719 );720 }721722 723 fn media_control_request_v3_round_trips_and_preserves_none() {724 for expected in [control_request(Some(42)), control_request(None)] {725 assert_eq!(726 MediaControlRequestV3::decode(&expected.encode().unwrap()).unwrap(),727 expected728 );729 }730 assert_eq!(&control_request(None).encode().unwrap()[20..28], &[0xff; 8]);731 assert!(732 MediaControlRequestV3::request_keyframe(733 1,734 Some(u64::MAX),735 KeyframeRequestReasonV3::Join,736 )737 .encode()738 .is_err()739 );740 }741742 743 fn media_control_request_v3_rejects_malformed_and_unknown_fields() {744 let valid = control_request(Some(1)).encode().unwrap();745 let cases: &[(usize, u8)] = &[746 (0, 0), // magic747 (5, 2), // version748 (7, 27), // request length749 (8, 2), // unknown type750 (9, 6), // unknown reason751 (10, 1), // reserved752 ];753 for &(offset, value) in cases {754 let mut malformed = valid;755 malformed[offset] = value;756 assert!(757 MediaControlRequestV3::decode(&malformed).is_err(),758 "offset {offset}"759 );760 }761 }762763 764 async fn media_control_request_v3_reads_fixed_messages_and_clean_eof() {765 let expected = control_request(Some(9));766 let (mut sender, mut receiver) = duplex(128);767 write_media_control_request_v3(&mut sender, &expected)768 .await769 .unwrap();770 sender.shutdown().await.unwrap();771 assert_eq!(772 read_media_control_request_v3(&mut receiver).await.unwrap(),773 Some(expected)774 );775 assert_eq!(776 read_media_control_request_v3(&mut receiver).await.unwrap(),777 None778 );779 }780781 782 async fn media_control_request_v3_rejects_truncation() {783 let wire = control_request(Some(9)).encode().unwrap();784 let (mut sender, mut receiver) = duplex(128);785 sender.write_all(&wire[..27]).await.unwrap();786 sender.shutdown().await.unwrap();787 assert!(matches!(788 read_media_control_request_v3(&mut receiver).await,789 Err(ProtocolError::Io(ref error)) if error.kind() == ErrorKind::UnexpectedEof790 ));791 }792}