1use bytes::{Buf, BufMut};
20
21use super::types::ObjectStatus;
22use crate::error::CodecError;
23use crate::object_properties::{ObjectProperty, PropertyValue};
24use crate::varint::VarInt;
25
26fn skip(buf: &mut impl Buf, len: u64) -> Result<(), CodecError> {
28 let len = usize::try_from(len).map_err(|_| CodecError::UnexpectedEnd)?;
29 if buf.remaining() < len {
30 return Err(CodecError::UnexpectedEnd);
31 }
32 buf.advance(len);
33 Ok(())
34}
35
36fn check_extensions_against_status(
57 status: Option<u64>,
58 extension_headers_len: u64,
59) -> Result<(), CodecError> {
60 if status == Some(ObjectStatus::ObjectDoesNotExist.as_u64()) && extension_headers_len != 0 {
61 return Err(CodecError::ExtensionsOnNonExistentObject(
66 usize::try_from(extension_headers_len).unwrap_or(usize::MAX),
67 ));
68 }
69 Ok(())
70}
71
72#[derive(Debug, Clone, Copy, PartialEq, Eq)]
87pub struct SubgroupStreamType(u8);
88
89impl SubgroupStreamType {
90 pub fn as_u8(self) -> u8 {
92 self.0
93 }
94
95 pub fn from_u8(v: u8) -> Option<Self> {
100 if (0x10..=0x15).contains(&v) || (0x18..=0x1D).contains(&v) {
101 Some(SubgroupStreamType(v))
102 } else {
103 None
104 }
105 }
106
107 pub fn from_flags(
113 subgroup_id_field_present: bool,
114 subgroup_id_is_first_object: bool,
115 extensions_present: bool,
116 end_of_group: bool,
117 ) -> Self {
118 let mut v: u8 = 0x10;
119 if extensions_present {
120 v |= 0x01;
121 }
122 if subgroup_id_field_present {
123 v |= 0x04;
124 } else if subgroup_id_is_first_object {
125 v |= 0x02;
126 }
127 if end_of_group {
128 v |= 0x08;
129 }
130 SubgroupStreamType(v)
131 }
132
133 pub fn has_subgroup_id_field(self) -> bool {
135 self.0 & 0x04 != 0
136 }
137
138 pub fn subgroup_id_is_first_object(self) -> bool {
142 !self.has_subgroup_id_field() && (self.0 & 0x02 != 0)
143 }
144
145 pub fn extensions_present(self) -> bool {
147 self.0 & 0x01 != 0
148 }
149
150 pub fn contains_end_of_group(self) -> bool {
153 self.0 & 0x08 != 0
154 }
155}
156
157#[derive(Debug, Clone, PartialEq, Eq)]
162pub struct SubgroupHeader {
163 pub stream_type: SubgroupStreamType,
165 pub track_alias: VarInt,
167 pub group_id: VarInt,
169 pub subgroup_id: Option<VarInt>,
174 pub publisher_priority: u8,
176}
177
178impl SubgroupHeader {
179 pub fn encode(&self, buf: &mut impl BufMut) {
189 VarInt::from_u64(self.stream_type.as_u8() as u64).unwrap().encode(buf);
190 self.track_alias.encode(buf);
191 self.group_id.encode(buf);
192 if self.stream_type.has_subgroup_id_field() {
193 let sg = self.subgroup_id.unwrap_or_else(|| VarInt::from_u64(0).unwrap());
194 sg.encode(buf);
195 }
196 buf.put_u8(self.publisher_priority);
197 }
198
199 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
218 if self.stream_type.has_subgroup_id_field() != self.subgroup_id.is_some() {
219 return Err(CodecError::InvalidField);
220 }
221 self.encode(buf);
222 Ok(())
223 }
224
225 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
232 let type_val = VarInt::decode(buf)?.into_inner();
233 if type_val > 0xFF {
234 return Err(stream_type_error(type_val));
235 }
236 let stream_type = SubgroupStreamType::from_u8(type_val as u8)
237 .ok_or_else(|| stream_type_error(type_val))?;
238 let track_alias = VarInt::decode(buf)?;
239 let group_id = VarInt::decode(buf)?;
240 let subgroup_id =
241 if stream_type.has_subgroup_id_field() { Some(VarInt::decode(buf)?) } else { None };
242 if buf.remaining() < 1 {
243 return Err(CodecError::UnexpectedEnd);
244 }
245 let publisher_priority = buf.get_u8();
246 Ok(SubgroupHeader { stream_type, track_alias, group_id, subgroup_id, publisher_priority })
247 }
248}
249
250#[derive(Debug, Clone, PartialEq, Eq)]
253pub struct SubgroupObject {
254 pub object_id: VarInt,
256 pub extension_headers: Vec<u8>,
262 pub status: Option<ObjectStatus>,
266 pub payload: Vec<u8>,
268}
269
270#[derive(Debug, Clone, Copy, PartialEq, Eq)]
275pub struct SubgroupObjectMeta {
276 pub object_id: u64,
278 pub extension_headers_len: u64,
281 pub payload_length: u64,
283 pub status: Option<u64>,
285 pub wire_len: u64,
287}
288
289#[derive(Debug, Clone, Copy, PartialEq, Eq)]
303pub enum PayloadPermission {
304 Permitted,
308 Forbidden,
311}
312
313impl PayloadPermission {
314 pub fn for_status(status: ObjectStatus) -> Self {
321 match status {
322 ObjectStatus::Normal => PayloadPermission::Permitted,
323 ObjectStatus::ObjectDoesNotExist => PayloadPermission::Forbidden,
324 ObjectStatus::EndOfGroup => PayloadPermission::Forbidden,
325 ObjectStatus::EndOfTrack => PayloadPermission::Forbidden,
326 }
327 }
328
329 pub fn permits(self) -> bool {
334 matches!(self, PayloadPermission::Permitted)
335 }
336}
337
338impl SubgroupObject {
339 pub fn status(&self) -> ObjectStatus {
344 if self.payload.is_empty() {
345 self.status.unwrap_or(ObjectStatus::Normal)
346 } else {
347 ObjectStatus::Normal
348 }
349 }
350
351 pub fn permits_payload(&self) -> bool {
358 PayloadPermission::for_status(self.status()).permits()
359 }
360}
361
362impl SubgroupObjectMeta {
363 pub fn payload_permission(&self) -> Option<PayloadPermission> {
384 match self.status {
385 None => Some(PayloadPermission::Permitted),
386 Some(code) => ObjectStatus::from_u64(code).map(PayloadPermission::for_status),
387 }
388 }
389}
390
391pub const PRIOR_GROUP_ID_GAP: u64 = 0x3C;
393pub const PRIOR_OBJECT_ID_GAP: u64 = 0x3E;
395
396#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
398pub struct PriorGaps {
399 pub group: Option<u64>,
401 pub object: Option<u64>,
403}
404
405impl PriorGaps {
406 pub fn from_properties(
415 properties: &[ObjectProperty<'_>],
416 group_id: u64,
417 object_id: u64,
418 ) -> Result<Self, CodecError> {
419 let mut gaps = Self::default();
420 collect_gaps(properties, &mut gaps)?;
421 if let Some(gap) = gaps.group.filter(|&gap| gap > group_id) {
422 return Err(CodecError::MalformedTrack {
423 detail: format!("Prior Group ID Gap {gap} is larger than Group ID {group_id}"),
424 });
425 }
426 if let Some(gap) = gaps.object.filter(|&gap| gap > object_id) {
427 return Err(CodecError::MalformedTrack {
428 detail: format!("Prior Object ID Gap {gap} is larger than Object ID {object_id}"),
429 });
430 }
431 Ok(gaps)
432 }
433}
434
435fn collect_gaps(properties: &[ObjectProperty<'_>], gaps: &mut PriorGaps) -> Result<(), CodecError> {
436 for property in properties {
437 let slot = match (&property.value, property.property_type) {
438 (PropertyValue::Immutable { properties, .. }, _) => {
439 collect_gaps(properties, gaps)?;
440 continue;
441 }
442 (PropertyValue::Varint(value), PRIOR_GROUP_ID_GAP) => {
443 (&mut gaps.group, *value, "Group")
444 }
445 (PropertyValue::Varint(value), PRIOR_OBJECT_ID_GAP) => {
446 (&mut gaps.object, *value, "Object")
447 }
448 _ => continue,
449 };
450 let (field, value, which) = slot;
451 if field.replace(value).is_some() {
452 return Err(CodecError::MalformedTrack {
453 detail: format!("an Object contains more than one Prior {which} ID Gap"),
454 });
455 }
456 }
457 Ok(())
458}
459
460#[derive(Debug, Clone, PartialEq, Eq)]
477pub struct FetchGaps(Box<crate::prior_gaps::FetchGapTracker>);
479
480impl Default for FetchGaps {
481 fn default() -> Self {
482 Self::new()
483 }
484}
485
486impl FetchGaps {
487 pub fn new() -> Self {
489 Self(Box::new(crate::prior_gaps::FetchGapTracker::new(
490 crate::prior_gaps::GapRules::Draft14To15,
491 )))
492 }
493
494 pub fn check(
500 &mut self,
501 group_id: u64,
502 object_id: u64,
503 extension_block: &[u8],
504 ) -> Result<PriorGaps, CodecError> {
505 let properties = crate::object_properties::parse_object_properties(
506 crate::version::DraftVersion::Draft14,
507 extension_block,
508 )?;
509 let gaps = PriorGaps::from_properties(&properties, group_id, object_id)?;
510 self.0.check(group_id, object_id, gaps.group.as_slice(), gaps.object)?;
511 Ok(gaps)
512 }
513}
514
515#[derive(Debug, Clone)]
522pub struct SubgroupObjectReader {
523 extensions_present: bool,
524 prev_object_id: Option<u64>,
525 group_id: u64,
526 group_gap: Option<u64>,
527}
528
529impl SubgroupObjectReader {
530 pub fn new(header: &SubgroupHeader) -> Self {
532 Self {
533 extensions_present: header.stream_type.extensions_present(),
534 prev_object_id: None,
535 group_id: header.group_id.into_inner(),
536 group_gap: None,
537 }
538 }
539
540 pub fn read_object(&mut self, buf: &mut impl Buf) -> Result<SubgroupObject, CodecError> {
554 let delta = VarInt::decode(buf)?.into_inner();
555 let prior_object_id = self.prev_object_id;
556 let object_id_val = match self.prev_object_id {
557 None => delta,
558 Some(prev) => prev
559 .checked_add(1)
560 .and_then(|v| v.checked_add(delta))
561 .ok_or(CodecError::InvalidField)?,
562 };
563 self.prev_object_id = Some(object_id_val);
564 let object_id = VarInt::from_u64(object_id_val).map_err(|_| CodecError::InvalidField)?;
565
566 let extension_headers = if self.extensions_present {
567 let ext_len = VarInt::decode(buf)?.into_inner() as usize;
568 crate::types::read_bytes(buf, ext_len)?
569 } else {
570 Vec::new()
571 };
572
573 let payload_length = VarInt::decode(buf)?.into_inner() as usize;
574 let (status, payload) = if payload_length == 0 {
575 let status_val = VarInt::decode(buf)?.into_inner();
576 let status = ObjectStatus::from_u64(status_val).ok_or(CodecError::InvalidField)?;
577 (Some(status), Vec::new())
578 } else {
579 let payload = crate::types::read_bytes(buf, payload_length)?;
580 (None, payload)
581 };
582 check_extensions_against_status(
583 status.map(|s| s.as_u64()),
584 extension_headers.len() as u64,
585 )?;
586
587 let properties = crate::object_properties::parse_object_properties(
588 crate::version::DraftVersion::Draft14,
589 &extension_headers,
590 )?;
591 let gaps = PriorGaps::from_properties(&properties, self.group_id, object_id_val)?;
592 self.check_stream_gaps(gaps, prior_object_id, object_id_val)?;
593 Ok(SubgroupObject { object_id, extension_headers, status, payload })
594 }
595
596 fn check_stream_gaps(
597 &mut self,
598 gaps: PriorGaps,
599 prior_object_id: Option<u64>,
600 object_id: u64,
601 ) -> Result<(), CodecError> {
602 if let Some(gap) = gaps.group {
603 match self.group_gap {
604 Some(seen) if seen != gap => {
605 return Err(CodecError::MalformedTrack {
606 detail: format!(
607 "two Objects in Group {} carry different Prior Group ID Gap values",
608 self.group_id
609 ),
610 });
611 }
612 _ => self.group_gap = Some(gap),
613 }
614 }
615 if let (Some(gap), Some(prior)) = (gaps.object, prior_object_id) {
618 if gap > 0 && prior >= object_id - gap {
619 return Err(CodecError::MalformedTrack {
620 detail: format!(
621 "Prior Object ID Gap {gap} of Object {object_id} covers Object {prior}"
622 ),
623 });
624 }
625 }
626 Ok(())
627 }
628
629 pub fn read_object_meta(
635 &mut self,
636 buf: &mut impl Buf,
637 ) -> Result<SubgroupObjectMeta, CodecError> {
638 let start = buf.remaining();
639 let delta = VarInt::decode(buf)?.into_inner();
640 let object_id_val = match self.prev_object_id {
641 None => delta,
642 Some(prev) => prev
643 .checked_add(1)
644 .and_then(|v| v.checked_add(delta))
645 .ok_or(CodecError::InvalidField)?,
646 };
647 self.prev_object_id = Some(object_id_val);
648 let object_id =
649 VarInt::from_u64(object_id_val).map_err(|_| CodecError::InvalidField)?.into_inner();
650
651 let extension_headers_len = if self.extensions_present {
652 let ext_len = VarInt::decode(buf)?.into_inner();
653 skip(buf, ext_len)?;
654 ext_len
655 } else {
656 0
657 };
658
659 let payload_length = VarInt::decode(buf)?.into_inner();
660 let status = if payload_length == 0 {
661 let status_val = VarInt::decode(buf)?.into_inner();
662 Some(ObjectStatus::from_u64(status_val).ok_or(CodecError::InvalidField)?.as_u64())
663 } else {
664 skip(buf, payload_length)?;
665 None
666 };
667 check_extensions_against_status(status, extension_headers_len)?;
668
669 Ok(SubgroupObjectMeta {
670 object_id,
671 extension_headers_len,
672 payload_length,
673 status,
674 wire_len: (start - buf.remaining()) as u64,
675 })
676 }
677
678 pub fn write_object(
704 &mut self,
705 object: &SubgroupObject,
706 buf: &mut impl BufMut,
707 ) -> Result<(), CodecError> {
708 let explicit_status = match object.status {
709 Some(status) if status != ObjectStatus::Normal => {
710 if !object.payload.is_empty() {
711 return Err(CodecError::InvalidField);
712 }
713 Some(status)
714 }
715 Some(ObjectStatus::Normal) if object.payload.is_empty() => Some(ObjectStatus::Normal),
716 _ => None,
717 };
718 if !self.extensions_present && !object.extension_headers.is_empty() {
719 return Err(CodecError::InvalidField);
720 }
721 check_extensions_against_status(
722 explicit_status.map(|s| s.as_u64()),
723 object.extension_headers.len() as u64,
724 )?;
725
726 let oid = object.object_id.into_inner();
727 let delta = match self.prev_object_id {
728 None => oid,
729 Some(prev) => oid
730 .checked_sub(prev)
731 .and_then(|v| v.checked_sub(1))
732 .ok_or(CodecError::InvalidField)?,
733 };
734 VarInt::from_u64(delta).map_err(|_| CodecError::InvalidField)?.encode(buf);
735 if self.extensions_present {
736 VarInt::from_u64(object.extension_headers.len() as u64)
737 .map_err(|_| CodecError::InvalidField)?
738 .encode(buf);
739 buf.put_slice(&object.extension_headers);
740 }
741 if let Some(status) = explicit_status {
742 VarInt::from_u64(0).unwrap().encode(buf);
743 VarInt::from_u64(status.as_u64()).unwrap().encode(buf);
744 } else {
745 VarInt::from_u64(object.payload.len() as u64)
746 .map_err(|_| CodecError::InvalidField)?
747 .encode(buf);
748 buf.put_slice(&object.payload);
749 }
750 self.prev_object_id = Some(oid);
751 Ok(())
752 }
753}
754
755fn stream_type_error(raw: u64) -> CodecError {
778 let assigned = raw == FETCH_STREAM_TYPE as u64
779 || (raw <= 0xFF && SubgroupStreamType::from_u8(raw as u8).is_some());
780 if assigned {
781 CodecError::InvalidField
782 } else {
783 CodecError::UnknownStreamType(raw)
784 }
785}
786
787pub const FETCH_STREAM_TYPE: u8 = 0x05;
789
790#[derive(Debug, Clone, PartialEq, Eq)]
792pub struct FetchHeader {
793 pub request_id: VarInt,
795}
796
797impl FetchHeader {
798 pub fn encode(&self, buf: &mut impl BufMut) {
800 VarInt::from_u64(FETCH_STREAM_TYPE as u64).unwrap().encode(buf);
801 self.request_id.encode(buf);
802 }
803
804 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
811 let type_val = VarInt::decode(buf)?.into_inner();
812 if type_val != FETCH_STREAM_TYPE as u64 {
813 return Err(stream_type_error(type_val));
814 }
815 let request_id = VarInt::decode(buf)?;
816 Ok(FetchHeader { request_id })
817 }
818}
819
820#[derive(Debug, Clone, PartialEq, Eq)]
826pub struct FetchObject {
827 pub group_id: VarInt,
829 pub subgroup_id: VarInt,
832 pub object_id: VarInt,
834 pub publisher_priority: u8,
836 pub extension_headers: Vec<u8>,
838 pub status: Option<ObjectStatus>,
840 pub payload: Vec<u8>,
842}
843
844#[derive(Debug, Clone, Copy, PartialEq, Eq)]
849pub struct FetchObjectMeta {
850 pub group_id: u64,
852 pub subgroup_id: u64,
854 pub object_id: u64,
856 pub publisher_priority: u8,
858 pub extension_headers_len: u64,
861 pub payload_length: u64,
863 pub status: Option<u64>,
865 pub wire_len: u64,
867}
868
869impl FetchObjectMeta {
870 pub fn payload_permission(&self) -> Option<PayloadPermission> {
877 match self.status {
878 None => Some(PayloadPermission::Permitted),
879 Some(code) => ObjectStatus::from_u64(code).map(PayloadPermission::for_status),
880 }
881 }
882}
883
884impl FetchObject {
885 pub fn status(&self) -> ObjectStatus {
890 if self.payload.is_empty() {
891 self.status.unwrap_or(ObjectStatus::Normal)
892 } else {
893 ObjectStatus::Normal
894 }
895 }
896
897 pub fn permits_payload(&self) -> bool {
905 PayloadPermission::for_status(self.status()).permits()
906 }
907
908 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
929 if matches!(self.status, Some(status) if status != ObjectStatus::Normal)
930 && !self.payload.is_empty()
931 {
932 return Err(CodecError::InvalidField);
933 }
934 check_extensions_against_status(
935 self.status.map(|s| s.as_u64()),
936 self.extension_headers.len() as u64,
937 )?;
938 self.encode(buf);
939 Ok(())
940 }
941
942 pub fn encode(&self, buf: &mut impl BufMut) {
949 self.group_id.encode(buf);
950 self.subgroup_id.encode(buf);
951 self.object_id.encode(buf);
952 buf.put_u8(self.publisher_priority);
953 VarInt::from_u64(self.extension_headers.len() as u64).unwrap().encode(buf);
954 buf.put_slice(&self.extension_headers);
955 if let Some(status) = self.status {
956 VarInt::from_u64(0).unwrap().encode(buf);
957 VarInt::from_u64(status.as_u64()).unwrap().encode(buf);
958 } else {
959 VarInt::from_u64(self.payload.len() as u64).unwrap().encode(buf);
960 buf.put_slice(&self.payload);
961 }
962 }
963
964 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
966 let group_id = VarInt::decode(buf)?;
967 let subgroup_id = VarInt::decode(buf)?;
968 let object_id = VarInt::decode(buf)?;
969 if buf.remaining() < 1 {
970 return Err(CodecError::UnexpectedEnd);
971 }
972 let publisher_priority = buf.get_u8();
973 let ext_len = VarInt::decode(buf)?.into_inner() as usize;
974 let extension_headers = crate::types::read_bytes(buf, ext_len)?;
975 let payload_length = VarInt::decode(buf)?.into_inner() as usize;
976 let (status, payload) = if payload_length == 0 {
977 let status_val = VarInt::decode(buf)?.into_inner();
978 let status = ObjectStatus::from_u64(status_val).ok_or(CodecError::InvalidField)?;
979 (Some(status), Vec::new())
980 } else {
981 (None, crate::types::read_bytes(buf, payload_length)?)
982 };
983 check_extensions_against_status(
984 status.map(|s| s.as_u64()),
985 extension_headers.len() as u64,
986 )?;
987 Ok(FetchObject {
988 group_id,
989 subgroup_id,
990 object_id,
991 publisher_priority,
992 extension_headers,
993 status,
994 payload,
995 })
996 }
997
998 pub fn decode_meta(buf: &mut impl Buf) -> Result<FetchObjectMeta, CodecError> {
1002 let start = buf.remaining();
1003 let group_id = VarInt::decode(buf)?.into_inner();
1004 let subgroup_id = VarInt::decode(buf)?.into_inner();
1005 let object_id = VarInt::decode(buf)?.into_inner();
1006 if buf.remaining() < 1 {
1007 return Err(CodecError::UnexpectedEnd);
1008 }
1009 let publisher_priority = buf.get_u8();
1010 let extension_headers_len = VarInt::decode(buf)?.into_inner();
1011 skip(buf, extension_headers_len)?;
1012 let payload_length = VarInt::decode(buf)?.into_inner();
1013 let status = if payload_length == 0 {
1014 let status_val = VarInt::decode(buf)?.into_inner();
1015 Some(ObjectStatus::from_u64(status_val).ok_or(CodecError::InvalidField)?.as_u64())
1016 } else {
1017 skip(buf, payload_length)?;
1018 None
1019 };
1020 check_extensions_against_status(status, extension_headers_len)?;
1021 Ok(FetchObjectMeta {
1022 group_id,
1023 subgroup_id,
1024 object_id,
1025 publisher_priority,
1026 extension_headers_len,
1027 payload_length,
1028 status,
1029 wire_len: (start - buf.remaining()) as u64,
1030 })
1031 }
1032}
1033
1034#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1049pub struct DatagramType(u8);
1050
1051impl DatagramType {
1052 pub fn as_u8(self) -> u8 {
1054 self.0
1055 }
1056
1057 pub fn from_u8(v: u8) -> Option<Self> {
1059 if (0x00..=0x07).contains(&v) || v == 0x20 || v == 0x21 {
1060 Some(DatagramType(v))
1061 } else {
1062 None
1063 }
1064 }
1065
1066 pub fn payload(object_id_present: bool, extensions_present: bool, end_of_group: bool) -> Self {
1068 let mut v: u8 = 0x00;
1069 if extensions_present {
1070 v |= 0x01;
1071 }
1072 if end_of_group {
1073 v |= 0x02;
1074 }
1075 if !object_id_present {
1076 v |= 0x04;
1077 }
1078 DatagramType(v)
1079 }
1080
1081 pub fn status(extensions_present: bool) -> Self {
1083 if extensions_present {
1084 DatagramType(0x21)
1085 } else {
1086 DatagramType(0x20)
1087 }
1088 }
1089
1090 pub fn is_status(self) -> bool {
1093 self.0 >= 0x20
1094 }
1095
1096 pub fn object_id_present(self) -> bool {
1098 if self.is_status() {
1101 true
1102 } else {
1103 self.0 & 0x04 == 0
1104 }
1105 }
1106
1107 pub fn end_of_group(self) -> bool {
1109 !self.is_status() && (self.0 & 0x02 != 0)
1110 }
1111
1112 pub fn extensions_present(self) -> bool {
1114 self.0 & 0x01 != 0
1115 }
1116}
1117
1118#[derive(Debug, Clone, PartialEq, Eq)]
1120pub struct DatagramObject {
1121 pub datagram_type: DatagramType,
1123 pub track_alias: VarInt,
1125 pub group_id: VarInt,
1127 pub object_id: VarInt,
1130 pub publisher_priority: u8,
1132 pub extension_headers: Vec<u8>,
1135 pub status: Option<ObjectStatus>,
1137 pub payload: Vec<u8>,
1139}
1140
1141impl DatagramObject {
1142 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1166 if self.datagram_type.is_status() {
1167 if !self.payload.is_empty() {
1168 return Err(CodecError::InvalidField);
1169 }
1170 } else if matches!(self.status, Some(status) if status != ObjectStatus::Normal) {
1171 return Err(CodecError::InvalidField);
1172 }
1173 if self.datagram_type.extensions_present() {
1179 if self.extension_headers.is_empty() {
1180 return Err(CodecError::InvalidField);
1181 }
1182 } else if !self.extension_headers.is_empty() {
1183 return Err(CodecError::InvalidField);
1184 }
1185 check_extensions_against_status(
1186 self.status.map(|s| s.as_u64()),
1187 self.extension_headers.len() as u64,
1188 )?;
1189 self.encode(buf);
1190 Ok(())
1191 }
1192
1193 pub fn encode(&self, buf: &mut impl BufMut) {
1202 VarInt::from_u64(self.datagram_type.as_u8() as u64).unwrap().encode(buf);
1203 self.track_alias.encode(buf);
1204 self.group_id.encode(buf);
1205 if self.datagram_type.object_id_present() {
1206 self.object_id.encode(buf);
1207 }
1208 buf.put_u8(self.publisher_priority);
1209 if self.datagram_type.extensions_present() {
1210 VarInt::from_u64(self.extension_headers.len() as u64).unwrap().encode(buf);
1211 buf.put_slice(&self.extension_headers);
1212 }
1213 if self.datagram_type.is_status() {
1214 let status = self.status.unwrap_or(ObjectStatus::Normal);
1215 VarInt::from_u64(status.as_u64()).unwrap().encode(buf);
1216 } else {
1217 buf.put_slice(&self.payload);
1218 }
1219 }
1220
1221 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1228 let type_val = VarInt::decode(buf)?.into_inner();
1229 if type_val > 0xFF {
1230 return Err(CodecError::UnknownDatagramType(type_val));
1231 }
1232 let datagram_type = DatagramType::from_u8(type_val as u8)
1233 .ok_or(CodecError::UnknownDatagramType(type_val))?;
1234 let track_alias = VarInt::decode(buf)?;
1235 let group_id = VarInt::decode(buf)?;
1236 let object_id = if datagram_type.object_id_present() {
1237 VarInt::decode(buf)?
1238 } else {
1239 VarInt::from_u64(0).unwrap()
1240 };
1241 if buf.remaining() < 1 {
1242 return Err(CodecError::UnexpectedEnd);
1243 }
1244 let publisher_priority = buf.get_u8();
1245 let extension_headers = if datagram_type.extensions_present() {
1246 let ext_len = VarInt::decode(buf)?.into_inner() as usize;
1247 if ext_len == 0 {
1258 return Err(CodecError::InvalidField);
1259 }
1260 crate::types::read_bytes(buf, ext_len)?
1261 } else {
1262 Vec::new()
1263 };
1264 let (status, payload) = if datagram_type.is_status() {
1265 let status_val = VarInt::decode(buf)?.into_inner();
1266 let status = ObjectStatus::from_u64(status_val).ok_or(CodecError::InvalidField)?;
1267 (Some(status), Vec::new())
1268 } else {
1269 let remaining = buf.remaining();
1270 (None, crate::types::read_bytes(buf, remaining)?)
1271 };
1272 check_extensions_against_status(
1273 status.map(|s| s.as_u64()),
1274 extension_headers.len() as u64,
1275 )?;
1276 Ok(DatagramObject {
1277 datagram_type,
1278 track_alias,
1279 group_id,
1280 object_id,
1281 publisher_priority,
1282 extension_headers,
1283 status,
1284 payload,
1285 })
1286 }
1287
1288 pub fn decode_object(buf: &mut impl Buf) -> Result<Self, CodecError> {
1304 let object = Self::decode(buf)?;
1305 if let Some(status) = object.status {
1306 if buf.has_remaining() {
1307 return Err(CodecError::PayloadNotPermitted {
1308 status: status.as_u64(),
1309 len: buf.remaining(),
1310 detail: "its type states a status in place of a payload",
1311 });
1312 }
1313 }
1314 let properties = crate::object_properties::parse_object_properties(
1315 crate::version::DraftVersion::Draft14,
1316 &object.extension_headers,
1317 )?;
1318 PriorGaps::from_properties(
1319 &properties,
1320 object.group_id.into_inner(),
1321 object.object_id.into_inner(),
1322 )?;
1323 Ok(object)
1324 }
1325}
1326
1327#[cfg(test)]
1328mod tests {
1329 use super::*;
1330
1331 fn vi(v: u64) -> VarInt {
1332 VarInt::from_u64(v).unwrap()
1333 }
1334
1335 #[test]
1338 fn subgroup_type_0x10_all_off() {
1339 let t = SubgroupStreamType::from_u8(0x10).unwrap();
1340 assert!(!t.has_subgroup_id_field());
1341 assert!(!t.subgroup_id_is_first_object());
1342 assert!(!t.extensions_present());
1343 assert!(!t.contains_end_of_group());
1344 }
1345
1346 #[test]
1347 fn subgroup_type_0x15_explicit_with_ext() {
1348 let t = SubgroupStreamType::from_u8(0x15).unwrap();
1349 assert!(t.has_subgroup_id_field());
1350 assert!(!t.subgroup_id_is_first_object());
1351 assert!(t.extensions_present());
1352 assert!(!t.contains_end_of_group());
1353 }
1354
1355 #[test]
1356 fn subgroup_type_0x1d_all_on() {
1357 let t = SubgroupStreamType::from_u8(0x1D).unwrap();
1358 assert!(t.has_subgroup_id_field());
1359 assert!(t.extensions_present());
1360 assert!(t.contains_end_of_group());
1361 }
1362
1363 #[test]
1364 fn subgroup_type_0x12_first_object() {
1365 let t = SubgroupStreamType::from_u8(0x12).unwrap();
1366 assert!(!t.has_subgroup_id_field());
1367 assert!(t.subgroup_id_is_first_object());
1368 assert!(!t.extensions_present());
1369 }
1370
1371 #[test]
1372 fn subgroup_type_rejects_undefined() {
1373 for bad in [0x00u8, 0x0F, 0x16, 0x17, 0x1E, 0x1F, 0x20] {
1374 assert!(SubgroupStreamType::from_u8(bad).is_none(), "0x{bad:02x} should be rejected");
1375 }
1376 }
1377
1378 #[test]
1379 fn subgroup_type_from_flags_roundtrip() {
1380 for &f_sg in &[false, true] {
1381 for &f_first in &[false, true] {
1382 for &f_ext in &[false, true] {
1383 for &f_eog in &[false, true] {
1384 let t = SubgroupStreamType::from_flags(f_sg, f_first, f_ext, f_eog);
1385 assert_eq!(t.has_subgroup_id_field(), f_sg);
1386 if !f_sg {
1389 assert_eq!(t.subgroup_id_is_first_object(), f_first);
1390 }
1391 assert_eq!(t.extensions_present(), f_ext);
1392 assert_eq!(t.contains_end_of_group(), f_eog);
1393 }
1394 }
1395 }
1396 }
1397 }
1398
1399 #[test]
1402 fn subgroup_header_roundtrip_0x10() {
1403 let h = SubgroupHeader {
1404 stream_type: SubgroupStreamType::from_u8(0x10).unwrap(),
1405 track_alias: vi(1),
1406 group_id: vi(0),
1407 subgroup_id: None,
1408 publisher_priority: 128,
1409 };
1410 let mut buf = Vec::new();
1411 h.encode(&mut buf);
1412 assert_eq!(buf[0], 0x10);
1413 let decoded = SubgroupHeader::decode(&mut &buf[..]).unwrap();
1414 assert_eq!(decoded, h);
1415 }
1416
1417 #[test]
1418 fn subgroup_header_roundtrip_explicit_subgroup() {
1419 let h = SubgroupHeader {
1420 stream_type: SubgroupStreamType::from_u8(0x14).unwrap(),
1421 track_alias: vi(5),
1422 group_id: vi(10),
1423 subgroup_id: Some(vi(2)),
1424 publisher_priority: 64,
1425 };
1426 let mut buf = Vec::new();
1427 h.encode(&mut buf);
1428 let decoded = SubgroupHeader::decode(&mut &buf[..]).unwrap();
1429 assert_eq!(decoded, h);
1430 }
1431
1432 #[test]
1433 fn subgroup_header_decode_rejects_bad_type() {
1434 let buf = [0x16u8, 0x01, 0x00, 0x80];
1440 let err = SubgroupHeader::decode(&mut &buf[..]).unwrap_err();
1441 assert!(
1442 matches!(err, CodecError::UnknownStreamType(0x16)),
1443 "an unassigned Type must be named as one, got {err:?}"
1444 );
1445 }
1446
1447 #[test]
1448 fn subgroup_header_decode_does_not_call_the_fetch_type_unknown() {
1449 let buf = [0x05u8, 0x01, 0x00, 0x80];
1453 let err = SubgroupHeader::decode(&mut &buf[..]).unwrap_err();
1454 assert!(
1455 matches!(err, CodecError::InvalidField),
1456 "a fetch stream at the subgroup reader must be refused without naming the \
1457 unknown-stream-type rule, got {err:?}"
1458 );
1459 }
1460
1461 #[test]
1464 fn subgroup_reader_delta_sequential_ids() {
1465 let header = SubgroupHeader {
1467 stream_type: SubgroupStreamType::from_u8(0x10).unwrap(),
1468 track_alias: vi(1),
1469 group_id: vi(0),
1470 subgroup_id: None,
1471 publisher_priority: 0,
1472 };
1473
1474 let mut write = SubgroupObjectReader::new(&header);
1475 let mut buf = Vec::new();
1476 for i in 0..3u64 {
1477 let obj = SubgroupObject {
1478 object_id: vi(i),
1479 extension_headers: vec![],
1480 status: None,
1481 payload: vec![0xAA + i as u8; 4],
1482 };
1483 write.write_object(&obj, &mut buf).unwrap();
1484 }
1485
1486 let mut read = SubgroupObjectReader::new(&header);
1487 let mut cursor = &buf[..];
1488 let o0 = read.read_object(&mut cursor).unwrap();
1489 assert_eq!(o0.object_id.into_inner(), 0);
1490 assert_eq!(o0.payload, vec![0xAA; 4]);
1491 let o1 = read.read_object(&mut cursor).unwrap();
1492 assert_eq!(o1.object_id.into_inner(), 1);
1493 let o2 = read.read_object(&mut cursor).unwrap();
1494 assert_eq!(o2.object_id.into_inner(), 2);
1495 }
1496
1497 #[test]
1498 fn subgroup_reader_delta_sparse_ids() {
1499 let header = SubgroupHeader {
1501 stream_type: SubgroupStreamType::from_u8(0x10).unwrap(),
1502 track_alias: vi(1),
1503 group_id: vi(0),
1504 subgroup_id: None,
1505 publisher_priority: 0,
1506 };
1507 let mut write = SubgroupObjectReader::new(&header);
1508 let mut buf = Vec::new();
1509 for &id in &[5u64, 10, 11] {
1510 write
1511 .write_object(
1512 &SubgroupObject {
1513 object_id: vi(id),
1514 extension_headers: vec![],
1515 status: None,
1516 payload: vec![1, 2, 3],
1517 },
1518 &mut buf,
1519 )
1520 .unwrap();
1521 }
1522 let mut read = SubgroupObjectReader::new(&header);
1523 let mut cursor = &buf[..];
1524 assert_eq!(read.read_object(&mut cursor).unwrap().object_id.into_inner(), 5);
1525 assert_eq!(read.read_object(&mut cursor).unwrap().object_id.into_inner(), 10);
1526 assert_eq!(read.read_object(&mut cursor).unwrap().object_id.into_inner(), 11);
1527 }
1528
1529 #[test]
1530 fn subgroup_reader_with_extensions() {
1531 let header = SubgroupHeader {
1533 stream_type: SubgroupStreamType::from_u8(0x11).unwrap(),
1534 track_alias: vi(1),
1535 group_id: vi(5),
1536 subgroup_id: None,
1537 publisher_priority: 0,
1538 };
1539 let mut write = SubgroupObjectReader::new(&header);
1540 let mut buf = Vec::new();
1541 write
1542 .write_object(
1543 &SubgroupObject {
1544 object_id: vi(0),
1545 extension_headers: vec![0x3c, 0x02],
1546 status: None,
1547 payload: vec![0xFF],
1548 },
1549 &mut buf,
1550 )
1551 .unwrap();
1552 let mut read = SubgroupObjectReader::new(&header);
1553 let o = read.read_object(&mut &buf[..]).unwrap();
1554 assert_eq!(o.extension_headers, vec![0x3c, 0x02]);
1555 assert_eq!(o.payload, vec![0xFF]);
1556 }
1557
1558 #[test]
1559 fn subgroup_reader_status_object() {
1560 let header = SubgroupHeader {
1561 stream_type: SubgroupStreamType::from_u8(0x10).unwrap(),
1562 track_alias: vi(1),
1563 group_id: vi(0),
1564 subgroup_id: None,
1565 publisher_priority: 0,
1566 };
1567 let mut write = SubgroupObjectReader::new(&header);
1568 let mut buf = Vec::new();
1569 write
1570 .write_object(
1571 &SubgroupObject {
1572 object_id: vi(7),
1573 extension_headers: vec![],
1574 status: Some(ObjectStatus::EndOfGroup),
1575 payload: vec![],
1576 },
1577 &mut buf,
1578 )
1579 .unwrap();
1580 let mut read = SubgroupObjectReader::new(&header);
1581 let o = read.read_object(&mut &buf[..]).unwrap();
1582 assert_eq!(o.object_id.into_inner(), 7);
1583 assert_eq!(o.status, Some(ObjectStatus::EndOfGroup));
1584 assert!(o.payload.is_empty());
1585 }
1586
1587 #[test]
1588 fn subgroup_reader_meta_matches_read_object() {
1589 let header = SubgroupHeader {
1591 stream_type: SubgroupStreamType::from_u8(0x11).unwrap(),
1592 track_alias: vi(1),
1593 group_id: vi(0),
1594 subgroup_id: None,
1595 publisher_priority: 0,
1596 };
1597 let mut write = SubgroupObjectReader::new(&header);
1598 let mut buf = Vec::new();
1599 for (id, status) in
1600 [(0u64, None), (4, Some(ObjectStatus::EndOfGroup)), (9, None)].into_iter()
1601 {
1602 write
1603 .write_object(
1604 &SubgroupObject {
1605 object_id: vi(id),
1606 extension_headers: vec![0x0A, 0x0B],
1607 status,
1608 payload: if status.is_some() { vec![] } else { vec![0xEE; 3] },
1609 },
1610 &mut buf,
1611 )
1612 .unwrap();
1613 }
1614
1615 let mut full = SubgroupObjectReader::new(&header);
1616 let mut meta = SubgroupObjectReader::new(&header);
1617 let mut full_cursor = &buf[..];
1618 let mut meta_cursor = &buf[..];
1619 for _ in 0..3 {
1620 let before = meta_cursor.remaining();
1621 let o = full.read_object(&mut full_cursor).unwrap();
1622 let m = meta.read_object_meta(&mut meta_cursor).unwrap();
1623 assert_eq!(m.object_id, o.object_id.into_inner());
1624 assert_eq!(m.extension_headers_len, o.extension_headers.len() as u64);
1625 assert_eq!(m.payload_length, o.payload.len() as u64);
1626 assert_eq!(m.status, o.status.map(|s| s.as_u64()));
1627 assert_eq!(m.wire_len, (before - meta_cursor.remaining()) as u64);
1628 assert_eq!(full_cursor.remaining(), meta_cursor.remaining());
1629 }
1630 assert!(meta_cursor.is_empty());
1631 }
1632
1633 #[test]
1634 fn subgroup_reader_meta_short_buffer_is_unexpected_end() {
1635 let header = SubgroupHeader {
1636 stream_type: SubgroupStreamType::from_u8(0x10).unwrap(),
1637 track_alias: vi(1),
1638 group_id: vi(0),
1639 subgroup_id: None,
1640 publisher_priority: 0,
1641 };
1642 let mut write = SubgroupObjectReader::new(&header);
1643 let mut buf = Vec::new();
1644 write
1645 .write_object(
1646 &SubgroupObject {
1647 object_id: vi(0),
1648 extension_headers: vec![],
1649 status: None,
1650 payload: vec![1, 2, 3, 4],
1651 },
1652 &mut buf,
1653 )
1654 .unwrap();
1655
1656 for cut in 1..buf.len() {
1657 let mut read = SubgroupObjectReader::new(&header);
1658 let err = read.read_object_meta(&mut &buf[..cut]).unwrap_err();
1659 assert!(
1660 matches!(err, CodecError::UnexpectedEnd | CodecError::VarInt(_)),
1661 "cut {cut} gave {err:?}"
1662 );
1663 }
1664 }
1665
1666 #[test]
1669 fn fetch_header_roundtrip() {
1670 let h = FetchHeader { request_id: vi(99) };
1671 let mut buf = Vec::new();
1672 h.encode(&mut buf);
1673 assert_eq!(buf[0], 0x05);
1674 assert_eq!(FetchHeader::decode(&mut &buf[..]).unwrap(), h);
1675 }
1676
1677 #[test]
1678 fn fetch_header_rejects_wrong_type() {
1679 let buf = [0x10u8, 0x05];
1680 assert!(FetchHeader::decode(&mut &buf[..]).is_err());
1681 }
1682
1683 #[test]
1684 fn fetch_object_roundtrip_with_payload() {
1685 let obj = FetchObject {
1686 group_id: vi(3),
1687 subgroup_id: vi(1),
1688 object_id: vi(7),
1689 publisher_priority: 200,
1690 extension_headers: vec![0xAA, 0xBB],
1691 status: None,
1692 payload: vec![1, 2, 3, 4],
1693 };
1694 let mut buf = Vec::new();
1695 obj.encode(&mut buf);
1696 assert_eq!(FetchObject::decode(&mut &buf[..]).unwrap(), obj);
1697 }
1698
1699 #[test]
1700 fn fetch_object_roundtrip_status() {
1701 let obj = FetchObject {
1702 group_id: vi(3),
1703 subgroup_id: vi(1),
1704 object_id: vi(8),
1705 publisher_priority: 200,
1706 extension_headers: vec![],
1707 status: Some(ObjectStatus::ObjectDoesNotExist),
1708 payload: vec![],
1709 };
1710 let mut buf = Vec::new();
1711 obj.encode(&mut buf);
1712 assert_eq!(FetchObject::decode(&mut &buf[..]).unwrap(), obj);
1713 }
1714
1715 #[test]
1716 fn fetch_object_meta_matches_decode() {
1717 for obj in [
1718 FetchObject {
1719 group_id: vi(3),
1720 subgroup_id: vi(1),
1721 object_id: vi(7),
1722 publisher_priority: 200,
1723 extension_headers: vec![0xAA, 0xBB],
1724 status: None,
1725 payload: vec![1, 2, 3, 4],
1726 },
1727 FetchObject {
1728 group_id: vi(4),
1729 subgroup_id: vi(0),
1730 object_id: vi(8),
1731 publisher_priority: 1,
1732 extension_headers: vec![],
1733 status: Some(ObjectStatus::EndOfTrack),
1734 payload: vec![],
1735 },
1736 ] {
1737 let mut buf = Vec::new();
1738 obj.encode(&mut buf);
1739 let mut decode_cursor = &buf[..];
1740 let mut meta_cursor = &buf[..];
1741 let decoded = FetchObject::decode(&mut decode_cursor).unwrap();
1742 let meta = FetchObject::decode_meta(&mut meta_cursor).unwrap();
1743 assert_eq!(meta.group_id, decoded.group_id.into_inner());
1744 assert_eq!(meta.subgroup_id, decoded.subgroup_id.into_inner());
1745 assert_eq!(meta.object_id, decoded.object_id.into_inner());
1746 assert_eq!(meta.publisher_priority, decoded.publisher_priority);
1747 assert_eq!(meta.extension_headers_len, decoded.extension_headers.len() as u64);
1748 assert_eq!(meta.payload_length, decoded.payload.len() as u64);
1749 assert_eq!(meta.status, decoded.status.map(|s| s.as_u64()));
1750 assert_eq!(meta.wire_len, buf.len() as u64);
1751 assert!(meta_cursor.is_empty());
1752 assert_eq!(decode_cursor.remaining(), meta_cursor.remaining());
1753 }
1754 }
1755
1756 #[test]
1759 fn datagram_type_variants() {
1760 let t0 = DatagramType::from_u8(0x00).unwrap();
1761 assert!(t0.object_id_present());
1762 assert!(!t0.extensions_present());
1763 assert!(!t0.end_of_group());
1764 assert!(!t0.is_status());
1765
1766 let t7 = DatagramType::from_u8(0x07).unwrap();
1767 assert!(!t7.object_id_present()); assert!(t7.extensions_present());
1769 assert!(t7.end_of_group());
1770 assert!(!t7.is_status());
1771
1772 let t20 = DatagramType::from_u8(0x20).unwrap();
1773 assert!(t20.is_status());
1774 assert!(!t20.extensions_present());
1775 assert!(t20.object_id_present());
1777
1778 let t21 = DatagramType::from_u8(0x21).unwrap();
1779 assert!(t21.is_status());
1780 assert!(t21.extensions_present());
1781 }
1782
1783 #[test]
1784 fn datagram_type_rejects_undefined() {
1785 for bad in [0x08u8, 0x10, 0x1F, 0x22, 0x80] {
1786 assert!(DatagramType::from_u8(bad).is_none(), "0x{bad:02x}");
1787 }
1788 }
1789
1790 #[test]
1793 fn datagram_object_0x00_roundtrip() {
1794 let d = DatagramObject {
1795 datagram_type: DatagramType::from_u8(0x00).unwrap(),
1796 track_alias: vi(1),
1797 group_id: vi(2),
1798 object_id: vi(3),
1799 publisher_priority: 100,
1800 extension_headers: vec![],
1801 status: None,
1802 payload: vec![0xDE, 0xAD, 0xBE, 0xEF],
1803 };
1804 let mut buf = Vec::new();
1805 d.encode(&mut buf);
1806 assert_eq!(DatagramObject::decode(&mut &buf[..]).unwrap(), d);
1807 }
1808
1809 #[test]
1810 fn datagram_object_0x04_no_object_id() {
1811 let d = DatagramObject {
1813 datagram_type: DatagramType::from_u8(0x04).unwrap(),
1814 track_alias: vi(1),
1815 group_id: vi(2),
1816 object_id: vi(0),
1817 publisher_priority: 100,
1818 extension_headers: vec![],
1819 status: None,
1820 payload: vec![0xAA],
1821 };
1822 let mut buf = Vec::new();
1823 d.encode(&mut buf);
1824 let decoded = DatagramObject::decode(&mut &buf[..]).unwrap();
1825 assert_eq!(decoded, d);
1826 }
1827
1828 #[test]
1829 fn datagram_object_0x21_status_with_extensions() {
1830 let d = DatagramObject {
1831 datagram_type: DatagramType::from_u8(0x21).unwrap(),
1832 track_alias: vi(9),
1833 group_id: vi(4),
1834 object_id: vi(11),
1835 publisher_priority: 50,
1836 extension_headers: vec![0xCA, 0xFE],
1837 status: Some(ObjectStatus::EndOfTrack),
1838 payload: vec![],
1839 };
1840 let mut buf = Vec::new();
1841 d.encode(&mut buf);
1842 assert_eq!(DatagramObject::decode(&mut &buf[..]).unwrap(), d);
1843 }
1844
1845 fn datagram(type_byte: u8, status: Option<ObjectStatus>, payload: Vec<u8>) -> DatagramObject {
1847 DatagramObject {
1848 datagram_type: DatagramType::from_u8(type_byte).unwrap(),
1849 track_alias: vi(1),
1850 group_id: vi(0),
1851 object_id: vi(0),
1852 publisher_priority: 128,
1853 extension_headers: vec![],
1854 status,
1855 payload,
1856 }
1857 }
1858
1859 #[test]
1880 fn encode_checked_refuses_the_field_the_type_byte_cannot_carry() {
1881 for &status in ObjectStatus::ALL {
1882 if status == ObjectStatus::Normal {
1883 continue;
1884 }
1885
1886 let object = datagram(0x00, Some(status), vec![0xDE, 0xAD]);
1888 let mut refused = Vec::new();
1889 let result = object.encode_checked(&mut refused);
1890 assert!(
1891 matches!(result, Err(CodecError::InvalidField)),
1892 "encode_checked must refuse {status:?} under a payload type; got {result:?}"
1893 );
1894 assert!(refused.is_empty(), "a refused {status:?} datagram still wrote {refused:?}");
1895
1896 let mut dropped = Vec::new();
1897 object.encode(&mut dropped);
1898 let decoded = DatagramObject::decode(&mut &dropped[..])
1899 .unwrap_or_else(|e| panic!("the lossy encoding of {status:?} must parse: {e:?}"));
1900 assert_eq!(decoded.status, None, "{status:?} is exactly what `encode` loses here");
1901 assert_eq!(decoded.payload, vec![0xDE, 0xAD]);
1902
1903 let mut carried = Vec::new();
1906 datagram(0x20, Some(status), vec![]).encode_checked(&mut carried).unwrap_or_else(|e| {
1907 panic!("encode_checked refused a status-type {status:?}: {e:?}")
1908 });
1909 let decoded = DatagramObject::decode(&mut &carried[..]).unwrap();
1910 assert_eq!(decoded.status, Some(status), "{status:?} lost its status");
1911 }
1912
1913 let object = datagram(0x20, Some(ObjectStatus::EndOfGroup), vec![0xDE, 0xAD]);
1915 let mut refused = Vec::new();
1916 let result = object.encode_checked(&mut refused);
1917 assert!(
1918 matches!(result, Err(CodecError::InvalidField)),
1919 "encode_checked must refuse a payload under a status type; got {result:?}"
1920 );
1921 assert!(refused.is_empty(), "a refused datagram still wrote {refused:?}");
1922
1923 let mut dropped = Vec::new();
1924 object.encode(&mut dropped);
1925 let decoded = DatagramObject::decode(&mut &dropped[..]).unwrap();
1926 assert!(decoded.payload.is_empty(), "the payload is exactly what `encode` loses here");
1927 assert_eq!(decoded.status, Some(ObjectStatus::EndOfGroup));
1928
1929 let mut named = Vec::new();
1932 datagram(0x00, Some(ObjectStatus::Normal), vec![0xDE, 0xAD])
1933 .encode_checked(&mut named)
1934 .unwrap_or_else(|e| {
1935 panic!("encode_checked refused a Normal status under a payload type: {e:?}")
1936 });
1937 let mut unnamed = Vec::new();
1938 datagram(0x00, None, vec![0xDE, 0xAD]).encode_checked(&mut unnamed).unwrap();
1939 assert_eq!(named, unnamed, "naming Normal must ask for the bytes leaving it out asks for");
1940 }
1941}
1942
1943#[cfg(test)]
1944mod datagram_gap_tests {
1945 use super::*;
1946
1947 fn datagram(group: u64, object: u64, block: &[u8]) -> Vec<u8> {
1948 let mut bytes = Vec::new();
1949 DatagramObject {
1950 datagram_type: DatagramType::payload(true, true, false),
1951 track_alias: VarInt::from_u64(1).unwrap(),
1952 group_id: VarInt::from_u64(group).unwrap(),
1953 object_id: VarInt::from_u64(object).unwrap(),
1954 publisher_priority: 0x80,
1955 extension_headers: block.to_vec(),
1956 status: None,
1957 payload: vec![0xAA],
1958 }
1959 .encode(&mut bytes);
1960 bytes
1961 }
1962
1963 fn is_malformed_track(bytes: &[u8]) -> bool {
1964 matches!(
1965 DatagramObject::decode_object(&mut &bytes[..]),
1966 Err(CodecError::MalformedTrack { .. })
1967 )
1968 }
1969
1970 #[test]
1971 fn decode_object_refuses_the_gap_conditions_one_datagram_shows() {
1972 for bytes in [
1973 datagram(1, 0, &[0x3C, 2]),
1974 datagram(5, 0, &[0x3E, 1]),
1975 datagram(5, 5, &[0x3C, 1, 0x3C, 1]),
1976 ] {
1977 assert!(is_malformed_track(&bytes));
1978 assert!(DatagramObject::decode(&mut &bytes[..]).is_ok());
1979 }
1980 let wire = datagram(2, 1, &[0x3C, 2, 0x3E, 1]);
1981 assert!(DatagramObject::decode_object(&mut &wire[..]).is_ok());
1982 }
1983
1984 #[test]
1985 fn decode_object_refuses_a_payload_after_a_status_datagram() {
1986 let mut wire = Vec::new();
1987 DatagramObject {
1988 datagram_type: DatagramType::from_u8(0x20).unwrap(),
1989 track_alias: VarInt::from_u64(1).unwrap(),
1990 group_id: VarInt::from_u64(0).unwrap(),
1991 object_id: VarInt::from_u64(0).unwrap(),
1992 publisher_priority: 0x80,
1993 extension_headers: Vec::new(),
1994 status: Some(ObjectStatus::Normal),
1995 payload: Vec::new(),
1996 }
1997 .encode(&mut wire);
1998 assert!(DatagramObject::decode_object(&mut &wire[..]).is_ok());
1999 wire.push(0xAA);
2000 assert!(DatagramObject::decode(&mut &wire[..]).is_ok());
2001 assert!(matches!(
2002 DatagramObject::decode_object(&mut &wire[..]),
2003 Err(CodecError::PayloadNotPermitted { len: 1, .. })
2004 ));
2005 }
2006}