1use bytes::{Buf, BufMut};
37
38use crate::dispatch::{AnyFetchHeader, AnySubgroupHeader};
39use crate::error::CodecError;
40use crate::varint::VarInt;
41use crate::version::DraftVersion;
42
43#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct AnySubgroupObject {
53 pub object_id: u64,
56 pub extension_headers: Vec<u8>,
69 pub extension_count: Option<u64>,
74 pub status: Option<u64>,
86 pub payload: Vec<u8>,
88}
89
90#[derive(Debug, Clone, Copy, PartialEq, Eq)]
95pub struct AnySubgroupObjectMeta {
96 pub object_id: u64,
98 pub payload_length: u64,
100 pub status: Option<u64>,
102 pub extension_headers_len: u64,
107 pub wire_len: u64,
110}
111
112#[derive(Debug, Clone, Copy, PartialEq, Eq)]
125pub enum AnyFetchEndOfRange {
126 NonExistent,
128 Unknown,
130 TimedOut,
139 Gone,
144 EndOfGroup,
150 EndOfTrack,
155}
156
157#[derive(Debug, Clone, Copy, PartialEq, Eq)]
168pub enum AnyFetchGroupOrder {
169 Ascending,
171 Descending,
173}
174
175#[derive(Debug, Clone, PartialEq, Eq)]
181pub struct AnyFetchObject {
182 pub group_id: u64,
186 pub subgroup_id: u64,
189 pub has_subgroup_id: bool,
202 pub object_id: u64,
204 pub publisher_priority: u8,
212 pub extension_headers: Vec<u8>,
215 pub extension_count: Option<u64>,
218 pub status: Option<u64>,
226 pub end_of_range: Option<AnyFetchEndOfRange>,
233 pub payload: Vec<u8>,
235}
236
237#[derive(Debug, Clone, Copy, PartialEq, Eq)]
241pub struct AnyFetchObjectMeta {
242 pub group_id: u64,
244 pub subgroup_id: u64,
246 pub has_subgroup_id: bool,
249 pub object_id: u64,
251 pub publisher_priority: u8,
254 pub payload_length: u64,
256 pub status: Option<u64>,
258 pub end_of_range: Option<AnyFetchEndOfRange>,
260 pub extension_headers_len: u64,
262 pub wire_len: u64,
264}
265
266#[cfg(any(
290 feature = "draft16",
291 feature = "draft17",
292 feature = "draft18",
293 feature = "draft19",
294 feature = "draft20",
295 feature = "draft21",
296 feature = "draft22",
297 feature = "draft23"
298))]
299const DEFAULT_PUBLISHER_PRIORITY: u8 = 128;
300
301#[allow(dead_code)]
304mod conv {
305 use super::{AnySubgroupObject, Buf, CodecError};
306 use crate::varint::VarInt;
307
308 pub fn skip(buf: &mut impl Buf, len: u64) -> Result<(), CodecError> {
310 let len = usize::try_from(len).map_err(|_| CodecError::UnexpectedEnd)?;
311 if buf.remaining() < len {
312 return Err(CodecError::UnexpectedEnd);
313 }
314 buf.advance(len);
315 Ok(())
316 }
317
318 pub fn take(buf: &mut impl Buf, len: u64) -> Result<Vec<u8>, CodecError> {
320 let len = usize::try_from(len).map_err(|_| CodecError::UnexpectedEnd)?;
321 crate::types::read_bytes(buf, len)
322 }
323
324 pub fn varint(v: u64) -> Result<VarInt, CodecError> {
326 VarInt::from_u64(v).map_err(|_| CodecError::InvalidField)
327 }
328
329 pub fn status_to_write(object: &AnySubgroupObject) -> Result<Option<u64>, CodecError> {
344 match (object.status, object.payload.is_empty()) {
345 (Some(0), false) => Ok(None),
346 (Some(_), false) => Err(CodecError::InvalidField),
347 (Some(code), true) => Ok(Some(code)),
348 (None, true) => Ok(Some(0)),
349 (None, false) => Ok(None),
350 }
351 }
352}
353
354macro_rules! legacy_subgroup_glue {
364 (no_extensions $name:ident, $feat:literal, $draft:ident) => {
365 #[cfg(feature = $feat)]
366 mod $name {
367 use super::conv;
368 use super::{AnySubgroupObject, AnySubgroupObjectMeta};
369 use crate::error::CodecError;
370 use crate::$draft::data_stream::ObjectHeader;
371 use crate::$draft::types::ObjectStatus;
372 use bytes::{Buf, BufMut};
373
374 pub fn read_object(buf: &mut impl Buf) -> Result<AnySubgroupObject, CodecError> {
375 let header = ObjectHeader::decode(buf)?;
376 let payload_length = header.payload_length.into_inner();
377 let (status, payload) = if payload_length == 0 {
378 (Some(header.object_status as u64), Vec::new())
379 } else {
380 (None, conv::take(buf, payload_length)?)
381 };
382 Ok(AnySubgroupObject {
383 object_id: header.object_id.into_inner(),
384 extension_headers: Vec::new(),
385 extension_count: None,
386 status,
387 payload,
388 })
389 }
390
391 pub fn read_object_meta(
392 buf: &mut impl Buf,
393 ) -> Result<AnySubgroupObjectMeta, CodecError> {
394 let start = buf.remaining();
395 let header = ObjectHeader::decode(buf)?;
396 let payload_length = header.payload_length.into_inner();
397 let status = if payload_length == 0 {
398 Some(header.object_status as u64)
399 } else {
400 conv::skip(buf, payload_length)?;
401 None
402 };
403 Ok(AnySubgroupObjectMeta {
404 object_id: header.object_id.into_inner(),
405 payload_length,
406 status,
407 extension_headers_len: 0,
408 wire_len: (start - buf.remaining()) as u64,
409 })
410 }
411
412 pub fn write_object(
413 object: &AnySubgroupObject,
414 buf: &mut impl BufMut,
415 ) -> Result<(), CodecError> {
416 if !object.extension_headers.is_empty() {
417 return Err(CodecError::InvalidField);
418 }
419 let object_status = match conv::status_to_write(object)? {
420 Some(code) => ObjectStatus::from_u64(code).ok_or(CodecError::InvalidField)?,
421 None => ObjectStatus::Normal,
422 };
423 ObjectHeader {
424 object_id: conv::varint(object.object_id)?,
425 payload_length: conv::varint(object.payload.len() as u64)?,
426 object_status,
427 }
428 .encode(buf);
429 buf.put_slice(&object.payload);
430 Ok(())
431 }
432 }
433 };
434
435 (count_extensions $name:ident, $feat:literal, $draft:ident) => {
436 #[cfg(feature = $feat)]
437 mod $name {
438 use super::conv;
439 use super::{AnySubgroupObject, AnySubgroupObjectMeta};
440 use crate::error::CodecError;
441 use crate::$draft::data_stream::ObjectHeader;
442 use crate::$draft::types::ObjectStatus;
443 use bytes::{Buf, BufMut};
444
445 pub fn read_object(buf: &mut impl Buf) -> Result<AnySubgroupObject, CodecError> {
446 let header = ObjectHeader::decode(buf)?;
447 let payload_length = header.payload_length.into_inner();
448 let (status, payload) = if payload_length == 0 {
449 (Some(header.object_status as u64), Vec::new())
450 } else {
451 (None, conv::take(buf, payload_length)?)
452 };
453 Ok(AnySubgroupObject {
454 object_id: header.object_id.into_inner(),
455 extension_headers: header.extensions,
456 extension_count: Some(header.extension_count.into_inner()),
457 status,
458 payload,
459 })
460 }
461
462 pub fn read_object_meta(
463 buf: &mut impl Buf,
464 ) -> Result<AnySubgroupObjectMeta, CodecError> {
465 let start = buf.remaining();
466 let header = ObjectHeader::decode(buf)?;
467 let payload_length = header.payload_length.into_inner();
468 let status = if payload_length == 0 {
469 Some(header.object_status as u64)
470 } else {
471 conv::skip(buf, payload_length)?;
472 None
473 };
474 Ok(AnySubgroupObjectMeta {
475 object_id: header.object_id.into_inner(),
476 payload_length,
477 status,
478 extension_headers_len: header.extensions.len() as u64,
479 wire_len: (start - buf.remaining()) as u64,
480 })
481 }
482
483 pub fn write_object(
484 object: &AnySubgroupObject,
485 buf: &mut impl BufMut,
486 ) -> Result<(), CodecError> {
487 let extension_count = match object.extension_count {
488 Some(count) => count,
489 None if object.extension_headers.is_empty() => 0,
490 None => return Err(CodecError::InvalidField),
491 };
492 let object_status = match conv::status_to_write(object)? {
493 Some(code) => ObjectStatus::from_u64(code).ok_or(CodecError::InvalidField)?,
494 None => ObjectStatus::Normal,
495 };
496 ObjectHeader {
497 object_id: conv::varint(object.object_id)?,
498 extension_count: conv::varint(extension_count)?,
499 extensions: object.extension_headers.clone(),
500 payload_length: conv::varint(object.payload.len() as u64)?,
501 object_status,
502 }
503 .encode(buf);
504 buf.put_slice(&object.payload);
505 Ok(())
506 }
507 }
508 };
509
510 (length_extensions $name:ident, $feat:literal, $draft:ident, $version:ident) => {
511 #[cfg(feature = $feat)]
512 mod $name {
513 use super::conv;
514 use super::{AnySubgroupObject, AnySubgroupObjectMeta};
515 use crate::error::CodecError;
516 use crate::$draft::data_stream::ObjectHeader;
517 use crate::$draft::types::ObjectStatus;
518 use bytes::{Buf, BufMut};
519
520 pub fn read_object(buf: &mut impl Buf) -> Result<AnySubgroupObject, CodecError> {
521 let header = ObjectHeader::decode(buf)?;
522 let payload_length = header.payload_length.into_inner();
523 let (status, payload) = if payload_length == 0 {
524 (Some(header.object_status as u64), Vec::new())
525 } else {
526 (None, conv::take(buf, payload_length)?)
527 };
528 crate::object_properties::parse_object_properties(
529 crate::version::DraftVersion::$version,
530 &header.extensions,
531 )?;
532 Ok(AnySubgroupObject {
533 object_id: header.object_id.into_inner(),
534 extension_headers: header.extensions,
535 extension_count: None,
536 status,
537 payload,
538 })
539 }
540
541 pub fn read_object_meta(
542 buf: &mut impl Buf,
543 ) -> Result<AnySubgroupObjectMeta, CodecError> {
544 let start = buf.remaining();
545 let header = ObjectHeader::decode(buf)?;
546 let payload_length = header.payload_length.into_inner();
547 let status = if payload_length == 0 {
548 Some(header.object_status as u64)
549 } else {
550 conv::skip(buf, payload_length)?;
551 None
552 };
553 Ok(AnySubgroupObjectMeta {
554 object_id: header.object_id.into_inner(),
555 payload_length,
556 status,
557 extension_headers_len: header.extension_headers_length.into_inner(),
558 wire_len: (start - buf.remaining()) as u64,
559 })
560 }
561
562 pub fn write_object(
563 object: &AnySubgroupObject,
564 buf: &mut impl BufMut,
565 ) -> Result<(), CodecError> {
566 let object_status = match conv::status_to_write(object)? {
567 Some(code) => ObjectStatus::from_u64(code).ok_or(CodecError::InvalidField)?,
568 None => ObjectStatus::Normal,
569 };
570 ObjectHeader {
571 object_id: conv::varint(object.object_id)?,
572 extension_headers_length: conv::varint(object.extension_headers.len() as u64)?,
573 extensions: object.extension_headers.clone(),
574 payload_length: conv::varint(object.payload.len() as u64)?,
575 object_status,
576 }
577 .encode(buf);
578 buf.put_slice(&object.payload);
579 Ok(())
580 }
581 }
582 };
583
584 (gated_extensions $name:ident, $feat:literal, $draft:ident) => {
585 #[cfg(feature = $feat)]
586 mod $name {
587 use super::conv;
588 use super::{AnySubgroupObject, AnySubgroupObjectMeta};
589 use crate::error::CodecError;
590 use crate::$draft::data_stream::{ObjectHeader, SubgroupGaps};
591 use crate::$draft::types::ObjectStatus;
592 use bytes::{Buf, BufMut};
593
594 pub fn read_object(
595 extensions: bool,
596 gaps: &mut SubgroupGaps,
597 buf: &mut impl Buf,
598 ) -> Result<AnySubgroupObject, CodecError> {
599 let header = ObjectHeader::decode_with_extensions(extensions, buf)?;
600 let payload_length = header.payload_length.into_inner();
601 let (status, payload) = if payload_length == 0 {
602 (Some(header.object_status as u64), Vec::new())
603 } else {
604 (None, conv::take(buf, payload_length)?)
605 };
606 gaps.check(&header)?;
607 Ok(AnySubgroupObject {
608 object_id: header.object_id.into_inner(),
609 extension_headers: header.extensions,
610 extension_count: None,
611 status,
612 payload,
613 })
614 }
615
616 pub fn read_object_meta(
617 extensions: bool,
618 buf: &mut impl Buf,
619 ) -> Result<AnySubgroupObjectMeta, CodecError> {
620 let start = buf.remaining();
621 let header = ObjectHeader::decode_with_extensions(extensions, buf)?;
622 let payload_length = header.payload_length.into_inner();
623 let status = if payload_length == 0 {
624 Some(header.object_status as u64)
625 } else {
626 conv::skip(buf, payload_length)?;
627 None
628 };
629 Ok(AnySubgroupObjectMeta {
630 object_id: header.object_id.into_inner(),
631 payload_length,
632 status,
633 extension_headers_len: header.extension_headers_length.into_inner(),
634 wire_len: (start - buf.remaining()) as u64,
635 })
636 }
637
638 pub fn write_object(
639 extensions: bool,
640 object: &AnySubgroupObject,
641 buf: &mut impl BufMut,
642 ) -> Result<(), CodecError> {
643 if !extensions && !object.extension_headers.is_empty() {
644 return Err(CodecError::InvalidField);
645 }
646 let object_status = match conv::status_to_write(object)? {
647 Some(code) => ObjectStatus::from_u64(code).ok_or(CodecError::InvalidField)?,
648 None => ObjectStatus::Normal,
649 };
650 ObjectHeader {
651 object_id: conv::varint(object.object_id)?,
652 extension_headers_length: conv::varint(object.extension_headers.len() as u64)?,
653 extensions: object.extension_headers.clone(),
654 payload_length: conv::varint(object.payload.len() as u64)?,
655 object_status,
656 }
657 .encode_with_extensions(extensions, buf);
658 buf.put_slice(&object.payload);
659 Ok(())
660 }
661 }
662 };
663}
664
665legacy_subgroup_glue!(no_extensions sg07, "draft07", draft07);
666legacy_subgroup_glue!(count_extensions sg08, "draft08", draft08);
667legacy_subgroup_glue!(length_extensions sg09, "draft09", draft09, Draft09);
668legacy_subgroup_glue!(length_extensions sg10, "draft10", draft10, Draft10);
669legacy_subgroup_glue!(gated_extensions sg11, "draft11", draft11);
670legacy_subgroup_glue!(gated_extensions sg12, "draft12", draft12);
671legacy_subgroup_glue!(gated_extensions sg13, "draft13", draft13);
672
673macro_rules! modern_subgroup_glue {
687 (derived_length $name:ident, $feat:literal, $draft:ident) => {
688 #[cfg(feature = $feat)]
689 mod $name {
690 use super::conv;
691 use super::{AnySubgroupObject, AnySubgroupObjectMeta};
692 use crate::error::CodecError;
693 use crate::$draft::data_stream::{SubgroupObject, SubgroupObjectReader};
694 use crate::$draft::types::ObjectStatus;
695 use bytes::{Buf, BufMut};
696
697 pub fn read_object(
698 reader: &mut SubgroupObjectReader,
699 buf: &mut impl Buf,
700 ) -> Result<AnySubgroupObject, CodecError> {
701 let object = reader.read_object(buf)?;
702 Ok(AnySubgroupObject {
703 object_id: object.object_id.into_inner(),
704 extension_headers: object.extension_headers,
705 extension_count: None,
706 status: object.status.map(ObjectStatus::as_u64),
707 payload: object.payload,
708 })
709 }
710
711 pub fn read_object_meta(
712 reader: &mut SubgroupObjectReader,
713 buf: &mut impl Buf,
714 ) -> Result<AnySubgroupObjectMeta, CodecError> {
715 let meta = reader.read_object_meta(buf)?;
716 Ok(AnySubgroupObjectMeta {
717 object_id: meta.object_id,
718 payload_length: meta.payload_length,
719 status: meta.status,
720 extension_headers_len: meta.extension_headers_len,
721 wire_len: meta.wire_len,
722 })
723 }
724
725 pub fn write_object(
726 writer: &mut SubgroupObjectReader,
727 object: &AnySubgroupObject,
728 buf: &mut impl BufMut,
729 ) -> Result<(), CodecError> {
730 let status = match conv::status_to_write(object)? {
731 Some(code) => {
732 Some(ObjectStatus::from_u64(code).ok_or(CodecError::InvalidField)?)
733 }
734 None => None,
735 };
736 writer.write_object(
737 &SubgroupObject {
738 object_id: conv::varint(object.object_id)?,
739 extension_headers: object.extension_headers.clone(),
740 status,
741 payload: object.payload.clone(),
742 },
743 buf,
744 )
745 }
746 }
747 };
748
749 (explicit_length $name:ident, $feat:literal, $draft:ident) => {
750 #[cfg(feature = $feat)]
751 mod $name {
752 use super::conv;
753 use super::{AnySubgroupObject, AnySubgroupObjectMeta};
754 use crate::error::CodecError;
755 use crate::$draft::data_stream::{SubgroupObject, SubgroupObjectReader};
756 use crate::$draft::types::ObjectStatus;
757 use bytes::{Buf, BufMut};
758
759 pub fn read_object(
760 reader: &mut SubgroupObjectReader,
761 buf: &mut impl Buf,
762 ) -> Result<AnySubgroupObject, CodecError> {
763 let object = reader.read_object(buf)?;
764 Ok(AnySubgroupObject {
765 object_id: object.object_id.into_inner(),
766 extension_headers: object.extension_headers,
767 extension_count: None,
768 status: object.object_status.map(ObjectStatus::as_u64),
769 payload: object.payload,
770 })
771 }
772
773 pub fn read_object_meta(
774 reader: &mut SubgroupObjectReader,
775 buf: &mut impl Buf,
776 ) -> Result<AnySubgroupObjectMeta, CodecError> {
777 let meta = reader.read_object_meta(buf)?;
778 Ok(AnySubgroupObjectMeta {
779 object_id: meta.object_id,
780 payload_length: meta.payload_length,
781 status: meta.status,
782 extension_headers_len: meta.extension_headers_len,
783 wire_len: meta.wire_len,
784 })
785 }
786
787 pub fn write_object(
788 writer: &mut SubgroupObjectReader,
789 object: &AnySubgroupObject,
790 buf: &mut impl BufMut,
791 ) -> Result<(), CodecError> {
792 let object_status = match conv::status_to_write(object)? {
793 Some(code) => {
794 Some(ObjectStatus::from_u64(code).ok_or(CodecError::InvalidField)?)
795 }
796 None => None,
797 };
798 writer.write_object(
799 &SubgroupObject {
800 object_id: conv::varint(object.object_id)?,
801 extension_headers: object.extension_headers.clone(),
802 payload_length: conv::varint(object.payload.len() as u64)?,
803 object_status,
804 payload: object.payload.clone(),
805 },
806 buf,
807 )
808 }
809 }
810 };
811}
812
813modern_subgroup_glue!(derived_length sg14, "draft14", draft14);
814modern_subgroup_glue!(explicit_length sg15, "draft15", draft15);
815modern_subgroup_glue!(explicit_length sg16, "draft16", draft16);
816modern_subgroup_glue!(explicit_length sg17, "draft17", draft17);
817modern_subgroup_glue!(explicit_length sg18, "draft18", draft18);
818modern_subgroup_glue!(explicit_length sg19, "draft19", draft19);
819modern_subgroup_glue!(explicit_length sg20, "draft20", draft20);
820modern_subgroup_glue!(explicit_length sg21, "draft21", draft21);
821modern_subgroup_glue!(explicit_length sg22, "draft22", draft22);
822modern_subgroup_glue!(explicit_length sg23, "draft23", draft23);
823
824macro_rules! fetch_glue {
831 (no_extensions $name:ident, $feat:literal, $draft:ident) => {
832 #[cfg(feature = $feat)]
833 mod $name {
834 use super::conv;
835 use super::{AnyFetchObject, AnyFetchObjectMeta};
836 use crate::error::CodecError;
837 use crate::$draft::data_stream::FetchObjectHeader;
838 use bytes::Buf;
839
840 pub fn read_object(buf: &mut impl Buf) -> Result<AnyFetchObject, CodecError> {
841 let header = FetchObjectHeader::decode(buf)?;
842 let payload_length = header.payload_length.into_inner();
843 let (status, payload) = if payload_length == 0 {
844 (Some(header.object_status as u64), Vec::new())
845 } else {
846 (None, conv::take(buf, payload_length)?)
847 };
848 Ok(AnyFetchObject {
849 group_id: header.group_id.into_inner(),
850 subgroup_id: header.subgroup_id.into_inner(),
851 has_subgroup_id: true,
852 object_id: header.object_id.into_inner(),
853 publisher_priority: header.publisher_priority,
854 extension_headers: Vec::new(),
855 extension_count: None,
856 status,
857 end_of_range: None,
858 payload,
859 })
860 }
861
862 pub fn read_object_meta(buf: &mut impl Buf) -> Result<AnyFetchObjectMeta, CodecError> {
863 let start = buf.remaining();
864 let header = FetchObjectHeader::decode(buf)?;
865 let payload_length = header.payload_length.into_inner();
866 let status = if payload_length == 0 {
867 Some(header.object_status as u64)
868 } else {
869 conv::skip(buf, payload_length)?;
870 None
871 };
872 Ok(AnyFetchObjectMeta {
873 group_id: header.group_id.into_inner(),
874 subgroup_id: header.subgroup_id.into_inner(),
875 has_subgroup_id: true,
876 object_id: header.object_id.into_inner(),
877 publisher_priority: header.publisher_priority,
878 payload_length,
879 status,
880 end_of_range: None,
881 extension_headers_len: 0,
882 wire_len: (start - buf.remaining()) as u64,
883 })
884 }
885 }
886 };
887
888 (count_extensions $name:ident, $feat:literal, $draft:ident) => {
889 #[cfg(feature = $feat)]
890 mod $name {
891 use super::conv;
892 use super::{AnyFetchObject, AnyFetchObjectMeta};
893 use crate::error::CodecError;
894 use crate::$draft::data_stream::FetchObjectHeader;
895 use bytes::Buf;
896
897 pub fn read_object(buf: &mut impl Buf) -> Result<AnyFetchObject, CodecError> {
898 let header = FetchObjectHeader::decode(buf)?;
899 let payload_length = header.payload_length.into_inner();
900 let (status, payload) = if payload_length == 0 {
901 (Some(header.object_status as u64), Vec::new())
902 } else {
903 (None, conv::take(buf, payload_length)?)
904 };
905 Ok(AnyFetchObject {
906 group_id: header.group_id.into_inner(),
907 subgroup_id: header.subgroup_id.into_inner(),
908 has_subgroup_id: true,
909 object_id: header.object_id.into_inner(),
910 publisher_priority: header.publisher_priority,
911 extension_headers: header.extensions,
912 extension_count: Some(header.extension_count.into_inner()),
913 status,
914 end_of_range: None,
915 payload,
916 })
917 }
918
919 pub fn read_object_meta(buf: &mut impl Buf) -> Result<AnyFetchObjectMeta, CodecError> {
920 let start = buf.remaining();
921 let header = FetchObjectHeader::decode(buf)?;
922 let payload_length = header.payload_length.into_inner();
923 let status = if payload_length == 0 {
924 Some(header.object_status as u64)
925 } else {
926 conv::skip(buf, payload_length)?;
927 None
928 };
929 Ok(AnyFetchObjectMeta {
930 group_id: header.group_id.into_inner(),
931 subgroup_id: header.subgroup_id.into_inner(),
932 has_subgroup_id: true,
933 object_id: header.object_id.into_inner(),
934 publisher_priority: header.publisher_priority,
935 payload_length,
936 status,
937 end_of_range: None,
938 extension_headers_len: header.extensions.len() as u64,
939 wire_len: (start - buf.remaining()) as u64,
940 })
941 }
942 }
943 };
944
945 (length_extensions $name:ident, $feat:literal, $draft:ident) => {
946 #[cfg(feature = $feat)]
947 mod $name {
948 use super::conv;
949 use super::{AnyFetchObject, AnyFetchObjectMeta};
950 use crate::error::CodecError;
951 use crate::$draft::data_stream::FetchObjectHeader;
952 use bytes::Buf;
953
954 pub fn read_object(buf: &mut impl Buf) -> Result<AnyFetchObject, CodecError> {
955 let header = FetchObjectHeader::decode(buf)?;
956 let payload_length = header.payload_length.into_inner();
957 let (status, payload) = if payload_length == 0 {
958 (Some(header.object_status as u64), Vec::new())
959 } else {
960 (None, conv::take(buf, payload_length)?)
961 };
962 Ok(AnyFetchObject {
963 group_id: header.group_id.into_inner(),
964 subgroup_id: header.subgroup_id.into_inner(),
965 has_subgroup_id: true,
966 object_id: header.object_id.into_inner(),
967 publisher_priority: header.publisher_priority,
968 extension_headers: header.extensions,
969 extension_count: None,
970 status,
971 end_of_range: None,
972 payload,
973 })
974 }
975
976 pub fn read_object_meta(buf: &mut impl Buf) -> Result<AnyFetchObjectMeta, CodecError> {
977 let start = buf.remaining();
978 let header = FetchObjectHeader::decode(buf)?;
979 let payload_length = header.payload_length.into_inner();
980 let status = if payload_length == 0 {
981 Some(header.object_status as u64)
982 } else {
983 conv::skip(buf, payload_length)?;
984 None
985 };
986 Ok(AnyFetchObjectMeta {
987 group_id: header.group_id.into_inner(),
988 subgroup_id: header.subgroup_id.into_inner(),
989 has_subgroup_id: true,
990 object_id: header.object_id.into_inner(),
991 publisher_priority: header.publisher_priority,
992 payload_length,
993 status,
994 end_of_range: None,
995 extension_headers_len: header.extension_headers_length.into_inner(),
996 wire_len: (start - buf.remaining()) as u64,
997 })
998 }
999 }
1000 };
1001}
1002
1003fetch_glue!(no_extensions fo07, "draft07", draft07);
1004fetch_glue!(count_extensions fo08, "draft08", draft08);
1005fetch_glue!(length_extensions fo09, "draft09", draft09);
1006fetch_glue!(length_extensions fo10, "draft10", draft10);
1007fetch_glue!(length_extensions fo11, "draft11", draft11);
1008fetch_glue!(length_extensions fo12, "draft12", draft12);
1009fetch_glue!(length_extensions fo13, "draft13", draft13);
1010
1011#[cfg(feature = "draft14")]
1012mod fo14 {
1013 use super::{AnyFetchObject, AnyFetchObjectMeta};
1014 use crate::draft14::data_stream::FetchObject;
1015 use crate::draft14::types::ObjectStatus;
1016 use crate::error::CodecError;
1017 use bytes::Buf;
1018
1019 pub fn read_object(buf: &mut impl Buf) -> Result<AnyFetchObject, CodecError> {
1020 let object = FetchObject::decode(buf)?;
1021 Ok(AnyFetchObject {
1022 group_id: object.group_id.into_inner(),
1023 subgroup_id: object.subgroup_id.into_inner(),
1024 has_subgroup_id: true,
1025 object_id: object.object_id.into_inner(),
1026 publisher_priority: object.publisher_priority,
1027 extension_headers: object.extension_headers,
1028 extension_count: None,
1029 status: object.status.map(ObjectStatus::as_u64),
1030 end_of_range: None,
1031 payload: object.payload,
1032 })
1033 }
1034
1035 pub fn read_object_meta(buf: &mut impl Buf) -> Result<AnyFetchObjectMeta, CodecError> {
1036 let meta = FetchObject::decode_meta(buf)?;
1037 Ok(AnyFetchObjectMeta {
1038 group_id: meta.group_id,
1039 subgroup_id: meta.subgroup_id,
1040 has_subgroup_id: true,
1041 object_id: meta.object_id,
1042 publisher_priority: meta.publisher_priority,
1043 payload_length: meta.payload_length,
1044 status: meta.status,
1045 end_of_range: None,
1046 extension_headers_len: meta.extension_headers_len,
1047 wire_len: meta.wire_len,
1048 })
1049 }
1050}
1051
1052#[cfg(feature = "draft15")]
1053mod fo15 {
1054 use super::{conv, AnyFetchObject, AnyFetchObjectMeta};
1055 use crate::draft15::data_stream::FetchObjectReader;
1056 use crate::draft15::types::ObjectStatus;
1057 use crate::error::CodecError;
1058 use bytes::Buf;
1059
1060 pub fn read_object(
1061 reader: &mut FetchObjectReader,
1062 buf: &mut impl Buf,
1063 ) -> Result<AnyFetchObject, CodecError> {
1064 let header = reader.read_object_header(buf)?;
1065 let payload = conv::take(buf, header.payload_length.into_inner())?;
1069 Ok(AnyFetchObject {
1070 group_id: header.group_id.into_inner(),
1071 subgroup_id: header.subgroup_id.into_inner(),
1072 has_subgroup_id: true,
1073 object_id: header.object_id.into_inner(),
1074 publisher_priority: header.publisher_priority,
1075 extension_headers: header.extension_headers,
1076 extension_count: None,
1077 status: header.object_status.map(ObjectStatus::as_u64),
1080 end_of_range: None,
1081 payload,
1082 })
1083 }
1084
1085 pub fn read_object_frame(
1086 reader: &mut FetchObjectReader,
1087 buf: &mut impl Buf,
1088 ) -> Result<super::AnyFetchFrame, CodecError> {
1089 let start = buf.remaining();
1090 let header = reader.read_frame_header(buf)?;
1091 let payload_length = header.payload_length.into_inner();
1092 conv::skip(buf, payload_length)?;
1093 let meta = AnyFetchObjectMeta {
1094 group_id: header.group_id.into_inner(),
1095 subgroup_id: header.subgroup_id.into_inner(),
1096 has_subgroup_id: true,
1097 object_id: header.object_id.into_inner(),
1098 publisher_priority: header.publisher_priority,
1099 payload_length,
1100 status: header.object_status.map(ObjectStatus::as_u64),
1101 end_of_range: None,
1102 extension_headers_len: header.extension_headers.len() as u64,
1103 wire_len: (start - buf.remaining()) as u64,
1104 };
1105 Ok(super::AnyFetchFrame {
1106 meta,
1107 draft: crate::version::DraftVersion::Draft15,
1108 shape: super::FetchFrameShape::Draft15(header),
1109 })
1110 }
1111
1112 pub fn read_object_meta(
1113 reader: &mut FetchObjectReader,
1114 buf: &mut impl Buf,
1115 ) -> Result<AnyFetchObjectMeta, CodecError> {
1116 read_object_frame(reader, buf).map(|frame| frame.meta)
1117 }
1118}
1119
1120#[cfg(feature = "draft16")]
1121mod fo16 {
1122 use super::DEFAULT_PUBLISHER_PRIORITY;
1123 use super::{conv, AnyFetchEndOfRange, AnyFetchObject, AnyFetchObjectMeta};
1124 use crate::draft16::data_stream::{
1125 FetchEndOfRange, FetchObjectHeader, FetchObjectLocation, FetchObjectReader,
1126 };
1127 use crate::error::CodecError;
1128 use bytes::Buf;
1129
1130 fn parts(
1134 reader: &mut FetchObjectReader,
1135 buf: &mut impl Buf,
1136 checked: bool,
1137 ) -> Result<(FetchObjectHeader, FetchObjectLocation), CodecError> {
1138 let header = FetchObjectHeader::decode(buf)?;
1139 let location =
1140 if checked { reader.resolve(&header)? } else { reader.resolve_frame(&header)? };
1141 Ok((header, location))
1142 }
1143
1144 fn resolved(location: &FetchObjectLocation) -> super::Resolved {
1145 let end_of_range = location.end_of_range;
1146 super::Resolved {
1147 group_id: location.group_id,
1148 subgroup_id: location.subgroup_id.filter(|_| end_of_range.is_none()),
1155 object_id: location.object_id,
1156 publisher_priority: location.publisher_priority.unwrap_or(DEFAULT_PUBLISHER_PRIORITY),
1157 end_of_range: end_of_range.map(|r| match r {
1158 FetchEndOfRange::NonExistent => AnyFetchEndOfRange::NonExistent,
1159 FetchEndOfRange::Unknown => AnyFetchEndOfRange::Unknown,
1160 }),
1161 }
1162 }
1163
1164 pub fn read_object(
1165 reader: &mut FetchObjectReader,
1166 buf: &mut impl Buf,
1167 ) -> Result<AnyFetchObject, CodecError> {
1168 let (header, location) = parts(reader, buf, true)?;
1169 let payload = conv::take(buf, header.payload_length.into_inner())?;
1170 Ok(resolved(&location).into_object(header.extensions.unwrap_or_default(), payload))
1171 }
1172
1173 pub fn read_object_frame(
1174 reader: &mut FetchObjectReader,
1175 buf: &mut impl Buf,
1176 ) -> Result<super::AnyFetchFrame, CodecError> {
1177 let start = buf.remaining();
1178 let (header, location) = parts(reader, buf, false)?;
1179 let payload_length = header.payload_length.into_inner();
1180 conv::skip(buf, payload_length)?;
1181 let meta = resolved(&location).into_meta(
1182 header.extensions.as_ref().map_or(0, |e| e.len() as u64),
1183 payload_length,
1184 (start - buf.remaining()) as u64,
1185 );
1186 Ok(super::AnyFetchFrame {
1187 meta,
1188 draft: crate::version::DraftVersion::Draft16,
1189 shape: super::FetchFrameShape::Draft16(header, location),
1190 })
1191 }
1192
1193 pub fn read_object_meta(
1194 reader: &mut FetchObjectReader,
1195 buf: &mut impl Buf,
1196 ) -> Result<AnyFetchObjectMeta, CodecError> {
1197 read_object_frame(reader, buf).map(|frame| frame.meta)
1198 }
1199}
1200
1201#[cfg(feature = "draft17")]
1202mod fo17 {
1203 use super::{conv, AnyFetchEndOfRange, AnyFetchObject, AnyFetchObjectMeta};
1204 use crate::draft17::data_stream::{EndOfRange, FetchObject, FetchObjectReader};
1205 use crate::error::CodecError;
1206 use bytes::Buf;
1207
1208 fn resolved(object: &FetchObject) -> super::Resolved {
1209 super::Resolved {
1210 group_id: object.group_id,
1211 subgroup_id: object.subgroup_id,
1212 object_id: object.object_id,
1213 publisher_priority: object
1214 .publisher_priority
1215 .unwrap_or(super::DEFAULT_PUBLISHER_PRIORITY),
1216 end_of_range: object.header.end_of_range().map(|r| match r {
1217 EndOfRange::NonExistent => AnyFetchEndOfRange::NonExistent,
1218 EndOfRange::Unknown => AnyFetchEndOfRange::Unknown,
1219 }),
1220 }
1221 }
1222
1223 pub fn read_object(
1224 reader: &mut FetchObjectReader,
1225 buf: &mut impl Buf,
1226 ) -> Result<AnyFetchObject, CodecError> {
1227 let object = reader.read_object_header(buf)?;
1228 let resolved = resolved(&object);
1229 let payload = conv::take(buf, object.header.payload_length.into_inner())?;
1230 Ok(resolved.into_object(object.header.properties, payload))
1231 }
1232
1233 pub fn read_object_frame(
1234 reader: &mut FetchObjectReader,
1235 buf: &mut impl Buf,
1236 ) -> Result<super::AnyFetchFrame, CodecError> {
1237 let start = buf.remaining();
1238 let object = reader.read_frame_header(buf)?;
1239 let resolved = resolved(&object);
1240 let payload_length = object.header.payload_length.into_inner();
1241 conv::skip(buf, payload_length)?;
1242 let meta = resolved.into_meta(
1243 object.header.properties.len() as u64,
1244 payload_length,
1245 (start - buf.remaining()) as u64,
1246 );
1247 Ok(super::AnyFetchFrame {
1248 meta,
1249 draft: crate::version::DraftVersion::Draft17,
1250 shape: super::FetchFrameShape::Draft17(object),
1251 })
1252 }
1253
1254 pub fn read_object_meta(
1255 reader: &mut FetchObjectReader,
1256 buf: &mut impl Buf,
1257 ) -> Result<AnyFetchObjectMeta, CodecError> {
1258 read_object_frame(reader, buf).map(|frame| frame.meta)
1259 }
1260}
1261
1262#[cfg(feature = "draft18")]
1263mod fo18 {
1264 use super::{conv, AnyFetchEndOfRange, AnyFetchObject, AnyFetchObjectMeta};
1265 use crate::draft18::data_stream::{EndOfRange, FetchObject, FetchObjectReader};
1266 use crate::error::CodecError;
1267 use bytes::Buf;
1268
1269 fn resolved(object: &FetchObject) -> super::Resolved {
1270 super::Resolved {
1271 group_id: object.group_id,
1272 subgroup_id: object.subgroup_id,
1273 object_id: object.object_id,
1274 publisher_priority: object
1275 .publisher_priority
1276 .unwrap_or(super::DEFAULT_PUBLISHER_PRIORITY),
1277 end_of_range: object.header.end_of_range().map(|r| match r {
1278 EndOfRange::NonExistent => AnyFetchEndOfRange::NonExistent,
1279 EndOfRange::Unknown => AnyFetchEndOfRange::Unknown,
1280 }),
1281 }
1282 }
1283
1284 pub fn read_object(
1285 reader: &mut FetchObjectReader,
1286 buf: &mut impl Buf,
1287 ) -> Result<AnyFetchObject, CodecError> {
1288 let object = reader.read_object_header(buf)?;
1289 let resolved = resolved(&object);
1290 let payload = conv::take(buf, object.header.payload_length.into_inner())?;
1291 Ok(resolved.into_object(object.header.properties, payload))
1292 }
1293
1294 pub fn read_object_frame(
1295 reader: &mut FetchObjectReader,
1296 buf: &mut impl Buf,
1297 ) -> Result<super::AnyFetchFrame, CodecError> {
1298 let start = buf.remaining();
1299 let object = reader.read_frame_header(buf)?;
1300 let resolved = resolved(&object);
1301 let payload_length = object.header.payload_length.into_inner();
1302 conv::skip(buf, payload_length)?;
1303 let meta = resolved.into_meta(
1304 object.header.properties.len() as u64,
1305 payload_length,
1306 (start - buf.remaining()) as u64,
1307 );
1308 Ok(super::AnyFetchFrame {
1309 meta,
1310 draft: crate::version::DraftVersion::Draft18,
1311 shape: super::FetchFrameShape::Draft18(object),
1312 })
1313 }
1314
1315 pub fn read_object_meta(
1316 reader: &mut FetchObjectReader,
1317 buf: &mut impl Buf,
1318 ) -> Result<AnyFetchObjectMeta, CodecError> {
1319 read_object_frame(reader, buf).map(|frame| frame.meta)
1320 }
1321}
1322
1323#[cfg(feature = "draft19")]
1324mod fo19 {
1325 use super::{conv, AnyFetchEndOfRange, AnyFetchObject, AnyFetchObjectMeta};
1326 use crate::draft19::data_stream::{FetchEndOfRange, FetchObject, FetchObjectReader};
1327 use crate::error::CodecError;
1328 use bytes::Buf;
1329
1330 fn resolved(object: &FetchObject) -> super::Resolved {
1331 super::Resolved {
1332 group_id: object.group_id,
1333 subgroup_id: object.subgroup_id,
1334 object_id: object.object_id,
1335 publisher_priority: object
1336 .publisher_priority
1337 .unwrap_or(super::DEFAULT_PUBLISHER_PRIORITY),
1338 end_of_range: object.header.end_of_range().map(|r| match r {
1339 FetchEndOfRange::NonExistent => AnyFetchEndOfRange::NonExistent,
1340 FetchEndOfRange::Unknown => AnyFetchEndOfRange::Unknown,
1341 }),
1342 }
1343 }
1344
1345 pub fn read_object(
1346 reader: &mut FetchObjectReader,
1347 buf: &mut impl Buf,
1348 ) -> Result<AnyFetchObject, CodecError> {
1349 let object = reader.read_object_header(buf)?;
1350 let resolved = resolved(&object);
1351 let payload = conv::take(buf, object.header.payload_length.into_inner())?;
1352 Ok(resolved.into_object(object.header.properties.unwrap_or_default(), payload))
1353 }
1354
1355 pub fn read_object_frame(
1356 reader: &mut FetchObjectReader,
1357 buf: &mut impl Buf,
1358 ) -> Result<super::AnyFetchFrame, CodecError> {
1359 let start = buf.remaining();
1360 let object = reader.read_frame_header(buf)?;
1361 let resolved = resolved(&object);
1362 let payload_length = object.header.payload_length.into_inner();
1363 conv::skip(buf, payload_length)?;
1364 let meta = resolved.into_meta(
1365 object.header.properties.as_ref().map_or(0, |p| p.len() as u64),
1366 payload_length,
1367 (start - buf.remaining()) as u64,
1368 );
1369 Ok(super::AnyFetchFrame {
1370 meta,
1371 draft: crate::version::DraftVersion::Draft19,
1372 shape: super::FetchFrameShape::Draft19(object),
1373 })
1374 }
1375
1376 pub fn read_object_meta(
1377 reader: &mut FetchObjectReader,
1378 buf: &mut impl Buf,
1379 ) -> Result<AnyFetchObjectMeta, CodecError> {
1380 read_object_frame(reader, buf).map(|frame| frame.meta)
1381 }
1382}
1383
1384#[cfg(feature = "draft20")]
1385mod fo20 {
1386 use super::{conv, AnyFetchEndOfRange, AnyFetchObject, AnyFetchObjectMeta};
1387 use crate::draft20::data_stream::{FetchEndOfRange, FetchObject, FetchObjectReader};
1388 use crate::error::CodecError;
1389 use bytes::Buf;
1390
1391 fn resolved(object: &FetchObject) -> super::Resolved {
1392 super::Resolved {
1393 group_id: object.group_id,
1394 subgroup_id: object.subgroup_id,
1395 object_id: object.object_id,
1396 publisher_priority: object
1397 .publisher_priority
1398 .unwrap_or(super::DEFAULT_PUBLISHER_PRIORITY),
1399 end_of_range: object.header.end_of_range().map(|r| match r {
1400 FetchEndOfRange::NonExistent => AnyFetchEndOfRange::NonExistent,
1401 FetchEndOfRange::Unknown => AnyFetchEndOfRange::Unknown,
1402 FetchEndOfRange::TimedOut => AnyFetchEndOfRange::TimedOut,
1405 }),
1406 }
1407 }
1408
1409 pub fn read_object(
1410 reader: &mut FetchObjectReader,
1411 buf: &mut impl Buf,
1412 ) -> Result<AnyFetchObject, CodecError> {
1413 let object = reader.read_object_header(buf)?;
1414 let resolved = resolved(&object);
1415 let payload = conv::take(buf, object.header.payload_length.into_inner())?;
1416 Ok(resolved.into_object(object.header.properties.unwrap_or_default(), payload))
1417 }
1418
1419 pub fn read_object_frame(
1420 reader: &mut FetchObjectReader,
1421 buf: &mut impl Buf,
1422 ) -> Result<super::AnyFetchFrame, CodecError> {
1423 let start = buf.remaining();
1424 let object = reader.read_frame_header(buf)?;
1425 let resolved = resolved(&object);
1426 let payload_length = object.header.payload_length.into_inner();
1427 conv::skip(buf, payload_length)?;
1428 let meta = resolved.into_meta(
1429 object.header.properties.as_ref().map_or(0, |p| p.len() as u64),
1430 payload_length,
1431 (start - buf.remaining()) as u64,
1432 );
1433 Ok(super::AnyFetchFrame {
1434 meta,
1435 draft: crate::version::DraftVersion::Draft20,
1436 shape: super::FetchFrameShape::Draft20(object),
1437 })
1438 }
1439
1440 pub fn read_object_meta(
1441 reader: &mut FetchObjectReader,
1442 buf: &mut impl Buf,
1443 ) -> Result<AnyFetchObjectMeta, CodecError> {
1444 read_object_frame(reader, buf).map(|frame| frame.meta)
1445 }
1446}
1447#[cfg(feature = "draft21")]
1448mod fo21 {
1449 use super::{conv, AnyFetchEndOfRange, AnyFetchObject, AnyFetchObjectMeta};
1450 use crate::draft21::data_stream::{FetchEndOfRange, FetchObject, FetchObjectReader};
1451 use crate::error::CodecError;
1452 use bytes::Buf;
1453
1454 fn resolved(object: &FetchObject) -> super::Resolved {
1455 super::Resolved {
1456 group_id: object.group_id,
1457 subgroup_id: object.subgroup_id,
1458 object_id: object.object_id,
1459 publisher_priority: object
1460 .publisher_priority
1461 .unwrap_or(super::DEFAULT_PUBLISHER_PRIORITY),
1462 end_of_range: object.header.end_of_range().map(|r| match r {
1463 FetchEndOfRange::NonExistent => AnyFetchEndOfRange::NonExistent,
1464 FetchEndOfRange::Unknown => AnyFetchEndOfRange::Unknown,
1465 FetchEndOfRange::TimedOut => AnyFetchEndOfRange::TimedOut,
1468 }),
1469 }
1470 }
1471
1472 pub fn read_object(
1473 reader: &mut FetchObjectReader,
1474 buf: &mut impl Buf,
1475 ) -> Result<AnyFetchObject, CodecError> {
1476 let object = reader.read_object_header(buf)?;
1477 let resolved = resolved(&object);
1478 let payload = conv::take(buf, object.header.payload_length.into_inner())?;
1479 Ok(resolved.into_object(object.header.properties.unwrap_or_default(), payload))
1480 }
1481
1482 pub fn read_object_frame(
1483 reader: &mut FetchObjectReader,
1484 buf: &mut impl Buf,
1485 ) -> Result<super::AnyFetchFrame, CodecError> {
1486 let start = buf.remaining();
1487 let object = reader.read_frame_header(buf)?;
1488 let resolved = resolved(&object);
1489 let payload_length = object.header.payload_length.into_inner();
1490 conv::skip(buf, payload_length)?;
1491 let meta = resolved.into_meta(
1492 object.header.properties.as_ref().map_or(0, |p| p.len() as u64),
1493 payload_length,
1494 (start - buf.remaining()) as u64,
1495 );
1496 Ok(super::AnyFetchFrame {
1497 meta,
1498 draft: crate::version::DraftVersion::Draft21,
1499 shape: super::FetchFrameShape::Draft21(object),
1500 })
1501 }
1502
1503 pub fn read_object_meta(
1504 reader: &mut FetchObjectReader,
1505 buf: &mut impl Buf,
1506 ) -> Result<AnyFetchObjectMeta, CodecError> {
1507 read_object_frame(reader, buf).map(|frame| frame.meta)
1508 }
1509}
1510
1511#[cfg(feature = "draft22")]
1512mod fo22 {
1513 use super::{conv, AnyFetchEndOfRange, AnyFetchObject, AnyFetchObjectMeta};
1514 use crate::draft22::data_stream::{FetchEndOfRange, FetchObject, FetchObjectReader};
1515 use crate::error::CodecError;
1516 use bytes::Buf;
1517
1518 fn resolved(object: &FetchObject) -> super::Resolved {
1519 super::Resolved {
1520 group_id: object.group_id,
1521 subgroup_id: object.subgroup_id,
1522 object_id: object.object_id,
1523 publisher_priority: object
1524 .publisher_priority
1525 .unwrap_or(super::DEFAULT_PUBLISHER_PRIORITY),
1526 end_of_range: object.header.end_of_range().map(|r| match r {
1527 FetchEndOfRange::NonExistent => AnyFetchEndOfRange::NonExistent,
1528 FetchEndOfRange::Unknown => AnyFetchEndOfRange::Unknown,
1529 FetchEndOfRange::TimedOut => AnyFetchEndOfRange::TimedOut,
1532 }),
1533 }
1534 }
1535
1536 pub fn read_object(
1537 reader: &mut FetchObjectReader,
1538 buf: &mut impl Buf,
1539 ) -> Result<AnyFetchObject, CodecError> {
1540 let object = reader.read_object_header(buf)?;
1541 let resolved = resolved(&object);
1542 let payload = conv::take(buf, object.header.payload_length.into_inner())?;
1543 Ok(resolved.into_object(object.header.properties.unwrap_or_default(), payload))
1544 }
1545
1546 pub fn read_object_frame(
1547 reader: &mut FetchObjectReader,
1548 buf: &mut impl Buf,
1549 ) -> Result<super::AnyFetchFrame, CodecError> {
1550 let start = buf.remaining();
1551 let object = reader.read_frame_header(buf)?;
1552 let resolved = resolved(&object);
1553 let payload_length = object.header.payload_length.into_inner();
1554 conv::skip(buf, payload_length)?;
1555 let meta = resolved.into_meta(
1556 object.header.properties.as_ref().map_or(0, |p| p.len() as u64),
1557 payload_length,
1558 (start - buf.remaining()) as u64,
1559 );
1560 Ok(super::AnyFetchFrame {
1561 meta,
1562 draft: crate::version::DraftVersion::Draft22,
1563 shape: super::FetchFrameShape::Draft22(object),
1564 })
1565 }
1566
1567 pub fn read_object_meta(
1568 reader: &mut FetchObjectReader,
1569 buf: &mut impl Buf,
1570 ) -> Result<AnyFetchObjectMeta, CodecError> {
1571 read_object_frame(reader, buf).map(|frame| frame.meta)
1572 }
1573}
1574#[cfg(feature = "draft23")]
1575mod fo23 {
1576 use super::{conv, AnyFetchEndOfRange, AnyFetchObject, AnyFetchObjectMeta};
1577 use crate::draft23::data_stream::{
1578 FetchEndMarker, FetchEndOfRange, FetchObject, FetchObjectReader,
1579 };
1580 use crate::error::CodecError;
1581 use bytes::Buf;
1582
1583 fn resolved(object: &FetchObject) -> super::Resolved {
1584 let range = object.header.end_of_range().map(|r| match r {
1585 FetchEndOfRange::NonExistent => AnyFetchEndOfRange::NonExistent,
1586 FetchEndOfRange::Unknown => AnyFetchEndOfRange::Unknown,
1587 FetchEndOfRange::TimedOut => AnyFetchEndOfRange::TimedOut,
1588 FetchEndOfRange::Gone => AnyFetchEndOfRange::Gone,
1589 });
1590 let marker = object.header.end_marker().map(|m| match m {
1591 FetchEndMarker::EndOfGroup => AnyFetchEndOfRange::EndOfGroup,
1592 FetchEndMarker::EndOfTrack => AnyFetchEndOfRange::EndOfTrack,
1593 });
1594 super::Resolved {
1595 group_id: object.group_id,
1596 subgroup_id: object.subgroup_id,
1597 object_id: object.object_id,
1598 publisher_priority: object
1599 .publisher_priority
1600 .unwrap_or(super::DEFAULT_PUBLISHER_PRIORITY),
1601 end_of_range: range.or(marker),
1602 }
1603 }
1604
1605 pub fn read_object(
1606 reader: &mut FetchObjectReader,
1607 buf: &mut impl Buf,
1608 ) -> Result<AnyFetchObject, CodecError> {
1609 let object = reader.read_object_header(buf)?;
1610 let resolved = resolved(&object);
1611 let payload = conv::take(buf, object.header.payload_length.into_inner())?;
1612 Ok(resolved.into_object(object.header.properties.unwrap_or_default(), payload))
1613 }
1614
1615 pub fn read_object_frame(
1616 reader: &mut FetchObjectReader,
1617 buf: &mut impl Buf,
1618 ) -> Result<super::AnyFetchFrame, CodecError> {
1619 let start = buf.remaining();
1620 let object = reader.read_frame_header(buf)?;
1621 let resolved = resolved(&object);
1622 let payload_length = object.header.payload_length.into_inner();
1623 conv::skip(buf, payload_length)?;
1624 let meta = resolved.into_meta(
1625 object.header.properties.as_ref().map_or(0, |p| p.len() as u64),
1626 payload_length,
1627 (start - buf.remaining()) as u64,
1628 );
1629 Ok(super::AnyFetchFrame {
1630 meta,
1631 draft: crate::version::DraftVersion::Draft23,
1632 shape: super::FetchFrameShape::Draft23(object),
1633 })
1634 }
1635
1636 pub fn read_object_meta(
1637 reader: &mut FetchObjectReader,
1638 buf: &mut impl Buf,
1639 ) -> Result<AnyFetchObjectMeta, CodecError> {
1640 read_object_frame(reader, buf).map(|frame| frame.meta)
1641 }
1642}
1643
1644#[cfg(any(
1653 feature = "draft16",
1654 feature = "draft17",
1655 feature = "draft18",
1656 feature = "draft19",
1657 feature = "draft20",
1658 feature = "draft21",
1659 feature = "draft22",
1660 feature = "draft23"
1661))]
1662struct Resolved {
1663 group_id: u64,
1664 subgroup_id: Option<u64>,
1665 object_id: u64,
1666 publisher_priority: u8,
1667 end_of_range: Option<AnyFetchEndOfRange>,
1668}
1669
1670#[cfg(any(
1671 feature = "draft16",
1672 feature = "draft17",
1673 feature = "draft18",
1674 feature = "draft19",
1675 feature = "draft20",
1676 feature = "draft21",
1677 feature = "draft22",
1678 feature = "draft23"
1679))]
1680impl Resolved {
1681 fn into_object(self, extension_headers: Vec<u8>, payload: Vec<u8>) -> AnyFetchObject {
1682 AnyFetchObject {
1683 group_id: self.group_id,
1684 subgroup_id: self.subgroup_id.unwrap_or(0),
1685 has_subgroup_id: self.subgroup_id.is_some(),
1686 object_id: self.object_id,
1687 publisher_priority: self.publisher_priority,
1688 extension_headers,
1689 extension_count: None,
1690 status: None,
1693 end_of_range: self.end_of_range,
1694 payload,
1695 }
1696 }
1697
1698 fn into_meta(
1699 self,
1700 extension_headers_len: u64,
1701 payload_length: u64,
1702 wire_len: u64,
1703 ) -> AnyFetchObjectMeta {
1704 AnyFetchObjectMeta {
1705 group_id: self.group_id,
1706 subgroup_id: self.subgroup_id.unwrap_or(0),
1707 has_subgroup_id: self.subgroup_id.is_some(),
1708 object_id: self.object_id,
1709 publisher_priority: self.publisher_priority,
1710 payload_length,
1711 status: None,
1712 end_of_range: self.end_of_range,
1713 extension_headers_len,
1714 wire_len,
1715 }
1716 }
1717}
1718
1719#[derive(Debug, Clone)]
1726enum SubgroupReaderState {
1727 #[cfg(feature = "draft07")]
1728 Draft07,
1729 #[cfg(feature = "draft08")]
1730 Draft08,
1731 #[cfg(feature = "draft09")]
1732 Draft09,
1733 #[cfg(feature = "draft10")]
1734 Draft10,
1735 #[cfg(feature = "draft11")]
1736 Draft11 { extensions: bool, gaps: crate::draft11::data_stream::SubgroupGaps },
1737 #[cfg(feature = "draft12")]
1738 Draft12 { extensions: bool, gaps: crate::draft12::data_stream::SubgroupGaps },
1739 #[cfg(feature = "draft13")]
1740 Draft13 { extensions: bool, gaps: crate::draft13::data_stream::SubgroupGaps },
1741 #[cfg(feature = "draft14")]
1742 Draft14(crate::draft14::data_stream::SubgroupObjectReader),
1743 #[cfg(feature = "draft15")]
1744 Draft15(crate::draft15::data_stream::SubgroupObjectReader),
1745 #[cfg(feature = "draft16")]
1746 Draft16(crate::draft16::data_stream::SubgroupObjectReader),
1747 #[cfg(feature = "draft17")]
1748 Draft17(crate::draft17::data_stream::SubgroupObjectReader),
1749 #[cfg(feature = "draft18")]
1750 Draft18(crate::draft18::data_stream::SubgroupObjectReader),
1751 #[cfg(feature = "draft19")]
1752 Draft19(crate::draft19::data_stream::SubgroupObjectReader),
1753 #[cfg(feature = "draft20")]
1754 Draft20(crate::draft20::data_stream::SubgroupObjectReader),
1755 #[cfg(feature = "draft21")]
1756 Draft21(crate::draft21::data_stream::SubgroupObjectReader),
1757 #[cfg(feature = "draft22")]
1758 Draft22(crate::draft22::data_stream::SubgroupObjectReader),
1759 #[cfg(feature = "draft23")]
1760 Draft23(crate::draft23::data_stream::SubgroupObjectReader),
1761}
1762
1763#[derive(Debug, Clone)]
1774pub struct AnySubgroupObjectReader {
1775 state: SubgroupReaderState,
1776}
1777
1778impl AnySubgroupObjectReader {
1779 #[allow(unused_variables, unreachable_code)]
1785 pub fn new(header: &AnySubgroupHeader) -> Result<Self, CodecError> {
1786 let state = match header {
1787 #[cfg(feature = "draft07")]
1788 AnySubgroupHeader::Draft07(_) => SubgroupReaderState::Draft07,
1789 #[cfg(feature = "draft08")]
1790 AnySubgroupHeader::Draft08(_) => SubgroupReaderState::Draft08,
1791 #[cfg(feature = "draft09")]
1792 AnySubgroupHeader::Draft09(_) => SubgroupReaderState::Draft09,
1793 #[cfg(feature = "draft10")]
1794 AnySubgroupHeader::Draft10(_) => SubgroupReaderState::Draft10,
1795 #[cfg(feature = "draft11")]
1796 AnySubgroupHeader::Draft11(h) => SubgroupReaderState::Draft11 {
1797 extensions: subgroup_extensions_11(h)?,
1798 gaps: crate::draft11::data_stream::SubgroupGaps::new(h),
1799 },
1800 #[cfg(feature = "draft12")]
1801 AnySubgroupHeader::Draft12(h) => SubgroupReaderState::Draft12 {
1802 extensions: subgroup_extensions_12(h)?,
1803 gaps: crate::draft12::data_stream::SubgroupGaps::new(h),
1804 },
1805 #[cfg(feature = "draft13")]
1806 AnySubgroupHeader::Draft13(h) => SubgroupReaderState::Draft13 {
1807 extensions: subgroup_extensions_13(h)?,
1808 gaps: crate::draft13::data_stream::SubgroupGaps::new(h),
1809 },
1810 #[cfg(feature = "draft14")]
1811 AnySubgroupHeader::Draft14(h) => SubgroupReaderState::Draft14(
1812 crate::draft14::data_stream::SubgroupObjectReader::new(h),
1813 ),
1814 #[cfg(feature = "draft15")]
1815 AnySubgroupHeader::Draft15(h) => SubgroupReaderState::Draft15(
1816 crate::draft15::data_stream::SubgroupObjectReader::new(h),
1817 ),
1818 #[cfg(feature = "draft16")]
1819 AnySubgroupHeader::Draft16(h) => SubgroupReaderState::Draft16(
1820 crate::draft16::data_stream::SubgroupObjectReader::new(h),
1821 ),
1822 #[cfg(feature = "draft17")]
1823 AnySubgroupHeader::Draft17(h) => SubgroupReaderState::Draft17(
1824 crate::draft17::data_stream::SubgroupObjectReader::new(h),
1825 ),
1826 #[cfg(feature = "draft18")]
1827 AnySubgroupHeader::Draft18(h) => SubgroupReaderState::Draft18(
1828 crate::draft18::data_stream::SubgroupObjectReader::new(h),
1829 ),
1830 #[cfg(feature = "draft19")]
1831 AnySubgroupHeader::Draft19(h) => SubgroupReaderState::Draft19(
1832 crate::draft19::data_stream::SubgroupObjectReader::new(h),
1833 ),
1834 #[cfg(feature = "draft20")]
1835 AnySubgroupHeader::Draft20(h) => SubgroupReaderState::Draft20(
1836 crate::draft20::data_stream::SubgroupObjectReader::new(h),
1837 ),
1838 #[cfg(feature = "draft21")]
1839 AnySubgroupHeader::Draft21(h) => SubgroupReaderState::Draft21(
1840 crate::draft21::data_stream::SubgroupObjectReader::new(h),
1841 ),
1842 #[cfg(feature = "draft22")]
1843 AnySubgroupHeader::Draft22(h) => SubgroupReaderState::Draft22(
1844 crate::draft22::data_stream::SubgroupObjectReader::new(h),
1845 ),
1846 #[cfg(feature = "draft23")]
1847 AnySubgroupHeader::Draft23(h) => SubgroupReaderState::Draft23(
1848 crate::draft23::data_stream::SubgroupObjectReader::new(h),
1849 ),
1850 #[allow(unreachable_patterns)]
1851 _ => {
1852 return Err(CodecError::UnsupportedDraft(format!(
1853 "draft {:?} not enabled via feature flag",
1854 header.draft()
1855 )));
1856 }
1857 };
1858 Ok(Self { state })
1859 }
1860
1861 #[allow(unreachable_code)]
1863 pub fn draft(&self) -> DraftVersion {
1864 match &self.state {
1865 #[cfg(feature = "draft07")]
1866 SubgroupReaderState::Draft07 => DraftVersion::Draft07,
1867 #[cfg(feature = "draft08")]
1868 SubgroupReaderState::Draft08 => DraftVersion::Draft08,
1869 #[cfg(feature = "draft09")]
1870 SubgroupReaderState::Draft09 => DraftVersion::Draft09,
1871 #[cfg(feature = "draft10")]
1872 SubgroupReaderState::Draft10 => DraftVersion::Draft10,
1873 #[cfg(feature = "draft11")]
1874 SubgroupReaderState::Draft11 { .. } => DraftVersion::Draft11,
1875 #[cfg(feature = "draft12")]
1876 SubgroupReaderState::Draft12 { .. } => DraftVersion::Draft12,
1877 #[cfg(feature = "draft13")]
1878 SubgroupReaderState::Draft13 { .. } => DraftVersion::Draft13,
1879 #[cfg(feature = "draft14")]
1880 SubgroupReaderState::Draft14(_) => DraftVersion::Draft14,
1881 #[cfg(feature = "draft15")]
1882 SubgroupReaderState::Draft15(_) => DraftVersion::Draft15,
1883 #[cfg(feature = "draft16")]
1884 SubgroupReaderState::Draft16(_) => DraftVersion::Draft16,
1885 #[cfg(feature = "draft17")]
1886 SubgroupReaderState::Draft17(_) => DraftVersion::Draft17,
1887 #[cfg(feature = "draft18")]
1888 SubgroupReaderState::Draft18(_) => DraftVersion::Draft18,
1889 #[cfg(feature = "draft19")]
1890 SubgroupReaderState::Draft19(_) => DraftVersion::Draft19,
1891 #[cfg(feature = "draft20")]
1892 SubgroupReaderState::Draft20(_) => DraftVersion::Draft20,
1893 #[cfg(feature = "draft21")]
1894 SubgroupReaderState::Draft21(_) => DraftVersion::Draft21,
1895 #[cfg(feature = "draft22")]
1896 SubgroupReaderState::Draft22(_) => DraftVersion::Draft22,
1897 #[cfg(feature = "draft23")]
1898 SubgroupReaderState::Draft23(_) => DraftVersion::Draft23,
1899 #[allow(unreachable_patterns)]
1900 _ => unreachable!("AnySubgroupObjectReader has no enabled variants"),
1901 }
1902 }
1903
1904 #[allow(unused_variables, unreachable_code)]
1917 pub fn read_object(&mut self, buf: &mut impl Buf) -> Result<AnySubgroupObject, CodecError> {
1918 match &mut self.state {
1919 #[cfg(feature = "draft07")]
1920 SubgroupReaderState::Draft07 => sg07::read_object(buf),
1921 #[cfg(feature = "draft08")]
1922 SubgroupReaderState::Draft08 => sg08::read_object(buf),
1923 #[cfg(feature = "draft09")]
1924 SubgroupReaderState::Draft09 => sg09::read_object(buf),
1925 #[cfg(feature = "draft10")]
1926 SubgroupReaderState::Draft10 => sg10::read_object(buf),
1927 #[cfg(feature = "draft11")]
1928 SubgroupReaderState::Draft11 { extensions, gaps } => {
1929 sg11::read_object(*extensions, gaps, buf)
1930 }
1931 #[cfg(feature = "draft12")]
1932 SubgroupReaderState::Draft12 { extensions, gaps } => {
1933 sg12::read_object(*extensions, gaps, buf)
1934 }
1935 #[cfg(feature = "draft13")]
1936 SubgroupReaderState::Draft13 { extensions, gaps } => {
1937 sg13::read_object(*extensions, gaps, buf)
1938 }
1939 #[cfg(feature = "draft14")]
1940 SubgroupReaderState::Draft14(inner) => sg14::read_object(inner, buf),
1941 #[cfg(feature = "draft15")]
1942 SubgroupReaderState::Draft15(inner) => sg15::read_object(inner, buf),
1943 #[cfg(feature = "draft16")]
1944 SubgroupReaderState::Draft16(inner) => sg16::read_object(inner, buf),
1945 #[cfg(feature = "draft17")]
1946 SubgroupReaderState::Draft17(inner) => sg17::read_object(inner, buf),
1947 #[cfg(feature = "draft18")]
1948 SubgroupReaderState::Draft18(inner) => sg18::read_object(inner, buf),
1949 #[cfg(feature = "draft19")]
1950 SubgroupReaderState::Draft19(inner) => sg19::read_object(inner, buf),
1951 #[cfg(feature = "draft20")]
1952 SubgroupReaderState::Draft20(inner) => sg20::read_object(inner, buf),
1953 #[cfg(feature = "draft21")]
1954 SubgroupReaderState::Draft21(inner) => sg21::read_object(inner, buf),
1955 #[cfg(feature = "draft22")]
1956 SubgroupReaderState::Draft22(inner) => sg22::read_object(inner, buf),
1957 #[cfg(feature = "draft23")]
1958 SubgroupReaderState::Draft23(inner) => sg23::read_object(inner, buf),
1959 #[allow(unreachable_patterns)]
1960 _ => unreachable!("AnySubgroupObjectReader has no enabled variants"),
1961 }
1962 }
1963
1964 #[allow(unused_variables, unreachable_code)]
1971 pub fn read_object_meta(
1972 &mut self,
1973 buf: &mut impl Buf,
1974 ) -> Result<AnySubgroupObjectMeta, CodecError> {
1975 match &mut self.state {
1976 #[cfg(feature = "draft07")]
1977 SubgroupReaderState::Draft07 => sg07::read_object_meta(buf),
1978 #[cfg(feature = "draft08")]
1979 SubgroupReaderState::Draft08 => sg08::read_object_meta(buf),
1980 #[cfg(feature = "draft09")]
1981 SubgroupReaderState::Draft09 => sg09::read_object_meta(buf),
1982 #[cfg(feature = "draft10")]
1983 SubgroupReaderState::Draft10 => sg10::read_object_meta(buf),
1984 #[cfg(feature = "draft11")]
1985 SubgroupReaderState::Draft11 { extensions, .. } => {
1986 sg11::read_object_meta(*extensions, buf)
1987 }
1988 #[cfg(feature = "draft12")]
1989 SubgroupReaderState::Draft12 { extensions, .. } => {
1990 sg12::read_object_meta(*extensions, buf)
1991 }
1992 #[cfg(feature = "draft13")]
1993 SubgroupReaderState::Draft13 { extensions, .. } => {
1994 sg13::read_object_meta(*extensions, buf)
1995 }
1996 #[cfg(feature = "draft14")]
1997 SubgroupReaderState::Draft14(inner) => sg14::read_object_meta(inner, buf),
1998 #[cfg(feature = "draft15")]
1999 SubgroupReaderState::Draft15(inner) => sg15::read_object_meta(inner, buf),
2000 #[cfg(feature = "draft16")]
2001 SubgroupReaderState::Draft16(inner) => sg16::read_object_meta(inner, buf),
2002 #[cfg(feature = "draft17")]
2003 SubgroupReaderState::Draft17(inner) => sg17::read_object_meta(inner, buf),
2004 #[cfg(feature = "draft18")]
2005 SubgroupReaderState::Draft18(inner) => sg18::read_object_meta(inner, buf),
2006 #[cfg(feature = "draft19")]
2007 SubgroupReaderState::Draft19(inner) => sg19::read_object_meta(inner, buf),
2008 #[cfg(feature = "draft20")]
2009 SubgroupReaderState::Draft20(inner) => sg20::read_object_meta(inner, buf),
2010 #[cfg(feature = "draft21")]
2011 SubgroupReaderState::Draft21(inner) => sg21::read_object_meta(inner, buf),
2012 #[cfg(feature = "draft22")]
2013 SubgroupReaderState::Draft22(inner) => sg22::read_object_meta(inner, buf),
2014 #[cfg(feature = "draft23")]
2015 SubgroupReaderState::Draft23(inner) => sg23::read_object_meta(inner, buf),
2016 #[allow(unreachable_patterns)]
2017 _ => unreachable!("AnySubgroupObjectReader has no enabled variants"),
2018 }
2019 }
2020}
2021
2022#[derive(Debug, Clone)]
2030enum SubgroupWriterState {
2031 #[cfg(feature = "draft07")]
2032 Draft07 { prev_object_id: Option<u64> },
2033 #[cfg(feature = "draft08")]
2034 Draft08 { prev_object_id: Option<u64> },
2035 #[cfg(feature = "draft09")]
2036 Draft09 { prev_object_id: Option<u64> },
2037 #[cfg(feature = "draft10")]
2038 Draft10 { prev_object_id: Option<u64> },
2039 #[cfg(feature = "draft11")]
2040 Draft11 { extensions: bool, prev_object_id: Option<u64> },
2041 #[cfg(feature = "draft12")]
2042 Draft12 { extensions: bool, prev_object_id: Option<u64> },
2043 #[cfg(feature = "draft13")]
2044 Draft13 { extensions: bool, prev_object_id: Option<u64> },
2045 #[cfg(feature = "draft14")]
2046 Draft14 { inner: crate::draft14::data_stream::SubgroupObjectReader, extensions: bool },
2047 #[cfg(feature = "draft15")]
2048 Draft15 { inner: crate::draft15::data_stream::SubgroupObjectReader, extensions: bool },
2049 #[cfg(feature = "draft16")]
2050 Draft16 { inner: crate::draft16::data_stream::SubgroupObjectReader, extensions: bool },
2051 #[cfg(feature = "draft17")]
2052 Draft17 { inner: crate::draft17::data_stream::SubgroupObjectReader, extensions: bool },
2053 #[cfg(feature = "draft18")]
2054 Draft18 { inner: crate::draft18::data_stream::SubgroupObjectReader, extensions: bool },
2055 #[cfg(feature = "draft19")]
2056 Draft19 { inner: crate::draft19::data_stream::SubgroupObjectReader, extensions: bool },
2057 #[cfg(feature = "draft20")]
2058 Draft20 { inner: crate::draft20::data_stream::SubgroupObjectReader, extensions: bool },
2059 #[cfg(feature = "draft21")]
2060 Draft21 { inner: crate::draft21::data_stream::SubgroupObjectReader, extensions: bool },
2061 #[cfg(feature = "draft22")]
2062 Draft22 { inner: crate::draft22::data_stream::SubgroupObjectReader, extensions: bool },
2063 #[cfg(feature = "draft23")]
2064 Draft23 { inner: crate::draft23::data_stream::SubgroupObjectReader, extensions: bool },
2065}
2066
2067#[derive(Debug, Clone)]
2083pub struct AnySubgroupObjectWriter {
2084 state: SubgroupWriterState,
2085}
2086
2087impl AnySubgroupObjectWriter {
2088 #[allow(unused_variables, unreachable_code)]
2094 pub fn new(header: &AnySubgroupHeader) -> Result<Self, CodecError> {
2095 let state = match header {
2096 #[cfg(feature = "draft07")]
2097 AnySubgroupHeader::Draft07(_) => SubgroupWriterState::Draft07 { prev_object_id: None },
2098 #[cfg(feature = "draft08")]
2099 AnySubgroupHeader::Draft08(_) => SubgroupWriterState::Draft08 { prev_object_id: None },
2100 #[cfg(feature = "draft09")]
2101 AnySubgroupHeader::Draft09(_) => SubgroupWriterState::Draft09 { prev_object_id: None },
2102 #[cfg(feature = "draft10")]
2103 AnySubgroupHeader::Draft10(_) => SubgroupWriterState::Draft10 { prev_object_id: None },
2104 #[cfg(feature = "draft11")]
2105 AnySubgroupHeader::Draft11(h) => SubgroupWriterState::Draft11 {
2106 extensions: subgroup_extensions_11(h)?,
2107 prev_object_id: None,
2108 },
2109 #[cfg(feature = "draft12")]
2110 AnySubgroupHeader::Draft12(h) => SubgroupWriterState::Draft12 {
2111 extensions: subgroup_extensions_12(h)?,
2112 prev_object_id: None,
2113 },
2114 #[cfg(feature = "draft13")]
2115 AnySubgroupHeader::Draft13(h) => SubgroupWriterState::Draft13 {
2116 extensions: subgroup_extensions_13(h)?,
2117 prev_object_id: None,
2118 },
2119 #[cfg(feature = "draft14")]
2120 AnySubgroupHeader::Draft14(h) => SubgroupWriterState::Draft14 {
2121 inner: crate::draft14::data_stream::SubgroupObjectReader::new(h),
2122 extensions: h.stream_type.extensions_present(),
2123 },
2124 #[cfg(feature = "draft15")]
2125 AnySubgroupHeader::Draft15(h) => SubgroupWriterState::Draft15 {
2126 inner: crate::draft15::data_stream::SubgroupObjectReader::new(h),
2127 extensions: h.has_extensions(),
2128 },
2129 #[cfg(feature = "draft16")]
2130 AnySubgroupHeader::Draft16(h) => SubgroupWriterState::Draft16 {
2131 inner: crate::draft16::data_stream::SubgroupObjectReader::new(h),
2132 extensions: h.has_extensions(),
2133 },
2134 #[cfg(feature = "draft17")]
2135 AnySubgroupHeader::Draft17(h) => SubgroupWriterState::Draft17 {
2136 inner: crate::draft17::data_stream::SubgroupObjectReader::new(h),
2137 extensions: h.has_properties(),
2138 },
2139 #[cfg(feature = "draft18")]
2140 AnySubgroupHeader::Draft18(h) => SubgroupWriterState::Draft18 {
2141 inner: crate::draft18::data_stream::SubgroupObjectReader::new(h),
2142 extensions: h.has_properties(),
2143 },
2144 #[cfg(feature = "draft19")]
2145 AnySubgroupHeader::Draft19(h) => SubgroupWriterState::Draft19 {
2146 inner: crate::draft19::data_stream::SubgroupObjectReader::new(h),
2147 extensions: h.has_properties(),
2148 },
2149 #[cfg(feature = "draft20")]
2150 AnySubgroupHeader::Draft20(h) => SubgroupWriterState::Draft20 {
2151 inner: crate::draft20::data_stream::SubgroupObjectReader::new(h),
2152 extensions: h.has_properties(),
2153 },
2154 #[cfg(feature = "draft21")]
2155 AnySubgroupHeader::Draft21(h) => SubgroupWriterState::Draft21 {
2156 inner: crate::draft21::data_stream::SubgroupObjectReader::new(h),
2157 extensions: h.has_properties(),
2158 },
2159 #[cfg(feature = "draft22")]
2160 AnySubgroupHeader::Draft22(h) => SubgroupWriterState::Draft22 {
2161 inner: crate::draft22::data_stream::SubgroupObjectReader::new(h),
2162 extensions: h.has_properties(),
2163 },
2164 #[cfg(feature = "draft23")]
2165 AnySubgroupHeader::Draft23(h) => SubgroupWriterState::Draft23 {
2166 inner: crate::draft23::data_stream::SubgroupObjectReader::new(h),
2167 extensions: h.has_properties(),
2168 },
2169 #[allow(unreachable_patterns)]
2170 _ => {
2171 return Err(CodecError::UnsupportedDraft(format!(
2172 "draft {:?} not enabled via feature flag",
2173 header.draft()
2174 )));
2175 }
2176 };
2177 Ok(Self { state })
2178 }
2179
2180 #[allow(unreachable_code)]
2182 pub fn draft(&self) -> DraftVersion {
2183 match &self.state {
2184 #[cfg(feature = "draft07")]
2185 SubgroupWriterState::Draft07 { .. } => DraftVersion::Draft07,
2186 #[cfg(feature = "draft08")]
2187 SubgroupWriterState::Draft08 { .. } => DraftVersion::Draft08,
2188 #[cfg(feature = "draft09")]
2189 SubgroupWriterState::Draft09 { .. } => DraftVersion::Draft09,
2190 #[cfg(feature = "draft10")]
2191 SubgroupWriterState::Draft10 { .. } => DraftVersion::Draft10,
2192 #[cfg(feature = "draft11")]
2193 SubgroupWriterState::Draft11 { .. } => DraftVersion::Draft11,
2194 #[cfg(feature = "draft12")]
2195 SubgroupWriterState::Draft12 { .. } => DraftVersion::Draft12,
2196 #[cfg(feature = "draft13")]
2197 SubgroupWriterState::Draft13 { .. } => DraftVersion::Draft13,
2198 #[cfg(feature = "draft14")]
2199 SubgroupWriterState::Draft14 { .. } => DraftVersion::Draft14,
2200 #[cfg(feature = "draft15")]
2201 SubgroupWriterState::Draft15 { .. } => DraftVersion::Draft15,
2202 #[cfg(feature = "draft16")]
2203 SubgroupWriterState::Draft16 { .. } => DraftVersion::Draft16,
2204 #[cfg(feature = "draft17")]
2205 SubgroupWriterState::Draft17 { .. } => DraftVersion::Draft17,
2206 #[cfg(feature = "draft18")]
2207 SubgroupWriterState::Draft18 { .. } => DraftVersion::Draft18,
2208 #[cfg(feature = "draft19")]
2209 SubgroupWriterState::Draft19 { .. } => DraftVersion::Draft19,
2210 #[cfg(feature = "draft20")]
2211 SubgroupWriterState::Draft20 { .. } => DraftVersion::Draft20,
2212 #[cfg(feature = "draft21")]
2213 SubgroupWriterState::Draft21 { .. } => DraftVersion::Draft21,
2214 #[cfg(feature = "draft22")]
2215 SubgroupWriterState::Draft22 { .. } => DraftVersion::Draft22,
2216 #[cfg(feature = "draft23")]
2217 SubgroupWriterState::Draft23 { .. } => DraftVersion::Draft23,
2218 #[allow(unreachable_patterns)]
2219 _ => unreachable!("AnySubgroupObjectWriter has no enabled variants"),
2220 }
2221 }
2222
2223 #[allow(unused_variables, unreachable_code)]
2252 pub fn write_object(
2253 &mut self,
2254 object: &AnySubgroupObject,
2255 buf: &mut impl BufMut,
2256 ) -> Result<(), CodecError> {
2257 match &mut self.state {
2258 #[cfg(feature = "draft07")]
2259 SubgroupWriterState::Draft07 { prev_object_id } => {
2260 advance_absolute_id(prev_object_id, object, |o| sg07::write_object(o, buf))
2261 }
2262 #[cfg(feature = "draft08")]
2263 SubgroupWriterState::Draft08 { prev_object_id } => {
2264 advance_absolute_id(prev_object_id, object, |o| sg08::write_object(o, buf))
2265 }
2266 #[cfg(feature = "draft09")]
2267 SubgroupWriterState::Draft09 { prev_object_id } => {
2268 advance_absolute_id(prev_object_id, object, |o| sg09::write_object(o, buf))
2269 }
2270 #[cfg(feature = "draft10")]
2271 SubgroupWriterState::Draft10 { prev_object_id } => {
2272 advance_absolute_id(prev_object_id, object, |o| sg10::write_object(o, buf))
2273 }
2274 #[cfg(feature = "draft11")]
2275 SubgroupWriterState::Draft11 { extensions, prev_object_id } => {
2276 let extensions = *extensions;
2277 advance_absolute_id(prev_object_id, object, |o| {
2278 sg11::write_object(extensions, o, buf)
2279 })
2280 }
2281 #[cfg(feature = "draft12")]
2282 SubgroupWriterState::Draft12 { extensions, prev_object_id } => {
2283 let extensions = *extensions;
2284 advance_absolute_id(prev_object_id, object, |o| {
2285 sg12::write_object(extensions, o, buf)
2286 })
2287 }
2288 #[cfg(feature = "draft13")]
2289 SubgroupWriterState::Draft13 { extensions, prev_object_id } => {
2290 let extensions = *extensions;
2291 advance_absolute_id(prev_object_id, object, |o| {
2292 sg13::write_object(extensions, o, buf)
2293 })
2294 }
2295 #[cfg(feature = "draft14")]
2296 SubgroupWriterState::Draft14 { inner, extensions } => {
2297 reject_unrepresentable_extensions(*extensions, object)?;
2298 sg14::write_object(inner, object, buf)
2299 }
2300 #[cfg(feature = "draft15")]
2301 SubgroupWriterState::Draft15 { inner, extensions } => {
2302 reject_unrepresentable_extensions(*extensions, object)?;
2303 sg15::write_object(inner, object, buf)
2304 }
2305 #[cfg(feature = "draft16")]
2306 SubgroupWriterState::Draft16 { inner, extensions } => {
2307 reject_unrepresentable_extensions(*extensions, object)?;
2308 sg16::write_object(inner, object, buf)
2309 }
2310 #[cfg(feature = "draft17")]
2311 SubgroupWriterState::Draft17 { inner, extensions } => {
2312 reject_unrepresentable_extensions(*extensions, object)?;
2313 sg17::write_object(inner, object, buf)
2314 }
2315 #[cfg(feature = "draft18")]
2316 SubgroupWriterState::Draft18 { inner, extensions } => {
2317 reject_unrepresentable_extensions(*extensions, object)?;
2318 sg18::write_object(inner, object, buf)
2319 }
2320 #[cfg(feature = "draft19")]
2321 SubgroupWriterState::Draft19 { inner, extensions } => {
2322 reject_unrepresentable_extensions(*extensions, object)?;
2323 sg19::write_object(inner, object, buf)
2324 }
2325 #[cfg(feature = "draft20")]
2326 SubgroupWriterState::Draft20 { inner, extensions } => {
2327 reject_unrepresentable_extensions(*extensions, object)?;
2328 sg20::write_object(inner, object, buf)
2329 }
2330 #[cfg(feature = "draft21")]
2331 SubgroupWriterState::Draft21 { inner, extensions } => {
2332 reject_unrepresentable_extensions(*extensions, object)?;
2333 sg21::write_object(inner, object, buf)
2334 }
2335 #[cfg(feature = "draft22")]
2336 SubgroupWriterState::Draft22 { inner, extensions } => {
2337 reject_unrepresentable_extensions(*extensions, object)?;
2338 sg22::write_object(inner, object, buf)
2339 }
2340 #[cfg(feature = "draft23")]
2341 SubgroupWriterState::Draft23 { inner, extensions } => {
2342 reject_unrepresentable_extensions(*extensions, object)?;
2343 sg23::write_object(inner, object, buf)
2344 }
2345 #[allow(unreachable_patterns)]
2346 _ => unreachable!("AnySubgroupObjectWriter has no enabled variants"),
2347 }
2348 }
2349}
2350
2351#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2355pub enum Reemit {
2356 Verbatim,
2358 Reencoded {
2360 id_bytes_before: usize,
2362 id_bytes_after: usize,
2364 },
2365}
2366
2367pub fn reemit_subgroup_object(
2429 draft: DraftVersion,
2430 prev_forwarded: Option<u64>,
2431 object_id: u64,
2432 raw: &[u8],
2433 out: &mut impl BufMut,
2434) -> Result<Reemit, CodecError> {
2435 if matches!(prev_forwarded, Some(prev) if object_id <= prev) {
2436 return Err(CodecError::InvalidField);
2437 }
2438
2439 let mut cursor: &[u8] = raw;
2443 draft.decode_varint(&mut cursor).map_err(|_| CodecError::InvalidField)?;
2444 let id_bytes_before = raw.len() - cursor.len();
2445
2446 if !delta_encodes_object_ids(draft) {
2449 out.put_slice(raw);
2450 return Ok(Reemit::Verbatim);
2451 }
2452
2453 let delta = match prev_forwarded {
2454 None => object_id,
2455 Some(prev) => object_id
2456 .checked_sub(prev)
2457 .and_then(|v| v.checked_sub(1))
2458 .ok_or(CodecError::InvalidField)?,
2459 };
2460
2461 let field = if draft.uses_moqt_varint() {
2464 VarInt::from_u64_moqt(delta)
2465 } else {
2466 VarInt::from_u64(delta).map_err(|_| CodecError::InvalidField)?
2467 };
2468
2469 let mut encoded = [0u8; 9];
2471 let mut slot: &mut [u8] = &mut encoded;
2472 draft.encode_varint(field, &mut slot);
2473 let id_bytes_after = 9 - slot.len();
2474 let encoded = &encoded[..id_bytes_after];
2475
2476 if encoded == &raw[..id_bytes_before] {
2477 out.put_slice(raw);
2478 return Ok(Reemit::Verbatim);
2479 }
2480
2481 out.put_slice(encoded);
2482 out.put_slice(&raw[id_bytes_before..]);
2483 Ok(Reemit::Reencoded { id_bytes_before, id_bytes_after })
2484}
2485
2486fn delta_encodes_object_ids(draft: DraftVersion) -> bool {
2492 match draft {
2493 DraftVersion::Draft07
2494 | DraftVersion::Draft08
2495 | DraftVersion::Draft09
2496 | DraftVersion::Draft10
2497 | DraftVersion::Draft11
2498 | DraftVersion::Draft12
2499 | DraftVersion::Draft13 => false,
2500 DraftVersion::Draft14
2501 | DraftVersion::Draft15
2502 | DraftVersion::Draft16
2503 | DraftVersion::Draft17
2504 | DraftVersion::Draft18
2505 | DraftVersion::Draft19
2506 | DraftVersion::Draft20
2507 | DraftVersion::Draft21
2508 | DraftVersion::Draft22
2509 | DraftVersion::Draft23 => true,
2510 }
2511}
2512
2513#[cfg(any(
2523 feature = "draft07",
2524 feature = "draft08",
2525 feature = "draft09",
2526 feature = "draft10",
2527 feature = "draft11",
2528 feature = "draft12",
2529 feature = "draft13"
2530))]
2531fn advance_absolute_id(
2532 prev_object_id: &mut Option<u64>,
2533 object: &AnySubgroupObject,
2534 write: impl FnOnce(&AnySubgroupObject) -> Result<(), CodecError>,
2535) -> Result<(), CodecError> {
2536 if matches!(*prev_object_id, Some(prev) if object.object_id <= prev) {
2537 return Err(CodecError::InvalidField);
2538 }
2539 write(object)?;
2540 *prev_object_id = Some(object.object_id);
2541 Ok(())
2542}
2543
2544#[cfg(any(
2547 feature = "draft14",
2548 feature = "draft15",
2549 feature = "draft16",
2550 feature = "draft17",
2551 feature = "draft18",
2552 feature = "draft19",
2553 feature = "draft20",
2554 feature = "draft21",
2555 feature = "draft22",
2556 feature = "draft23"
2557))]
2558fn reject_unrepresentable_extensions(
2559 extensions: bool,
2560 object: &AnySubgroupObject,
2561) -> Result<(), CodecError> {
2562 if !extensions && !object.extension_headers.is_empty() {
2563 return Err(CodecError::InvalidField);
2564 }
2565 Ok(())
2566}
2567
2568#[derive(Debug, Clone)]
2574enum FetchReaderState {
2575 #[cfg(feature = "draft07")]
2576 Draft07,
2577 #[cfg(feature = "draft08")]
2578 Draft08,
2579 #[cfg(feature = "draft09")]
2580 Draft09,
2581 #[cfg(feature = "draft10")]
2582 Draft10,
2583 #[cfg(feature = "draft11")]
2584 Draft11,
2585 #[cfg(feature = "draft12")]
2586 Draft12,
2587 #[cfg(feature = "draft13")]
2588 Draft13,
2589 #[cfg(feature = "draft14")]
2590 Draft14,
2591 #[cfg(feature = "draft15")]
2592 Draft15(crate::draft15::data_stream::FetchObjectReader),
2593 #[cfg(feature = "draft16")]
2594 Draft16(crate::draft16::data_stream::FetchObjectReader),
2595 #[cfg(feature = "draft17")]
2596 Draft17(crate::draft17::data_stream::FetchObjectReader),
2597 #[cfg(feature = "draft18")]
2598 Draft18(crate::draft18::data_stream::FetchObjectReader),
2599 #[cfg(feature = "draft19")]
2600 Draft19(crate::draft19::data_stream::FetchObjectReader),
2601 #[cfg(feature = "draft20")]
2602 Draft20(crate::draft20::data_stream::FetchObjectReader),
2603 #[cfg(feature = "draft21")]
2604 Draft21(crate::draft21::data_stream::FetchObjectReader),
2605 #[cfg(feature = "draft22")]
2606 Draft22(crate::draft22::data_stream::FetchObjectReader),
2607 #[cfg(feature = "draft23")]
2608 Draft23(crate::draft23::data_stream::FetchObjectReader),
2609}
2610
2611#[derive(Debug, Clone)]
2616enum FetchGapsState {
2617 None,
2618 #[cfg(feature = "draft11")]
2619 Draft11(crate::draft11::data_stream::FetchGaps),
2620 #[cfg(feature = "draft12")]
2621 Draft12(crate::draft12::data_stream::FetchGaps),
2622 #[cfg(feature = "draft13")]
2623 Draft13(crate::draft13::data_stream::FetchGaps),
2624 #[cfg(feature = "draft14")]
2625 Draft14(crate::draft14::data_stream::FetchGaps),
2626}
2627
2628impl FetchGapsState {
2629 fn for_draft(draft: DraftVersion) -> Self {
2630 match draft {
2631 #[cfg(feature = "draft11")]
2632 DraftVersion::Draft11 => Self::Draft11(crate::draft11::data_stream::FetchGaps::new()),
2633 #[cfg(feature = "draft12")]
2634 DraftVersion::Draft12 => Self::Draft12(crate::draft12::data_stream::FetchGaps::new()),
2635 #[cfg(feature = "draft13")]
2636 DraftVersion::Draft13 => Self::Draft13(crate::draft13::data_stream::FetchGaps::new()),
2637 #[cfg(feature = "draft14")]
2638 DraftVersion::Draft14 => Self::Draft14(crate::draft14::data_stream::FetchGaps::new()),
2639 #[allow(unreachable_patterns)]
2640 _ => Self::None,
2641 }
2642 }
2643
2644 #[allow(unused_variables)]
2645 fn check(&mut self, group_id: u64, object_id: u64, block: &[u8]) -> Result<(), CodecError> {
2646 match self {
2647 Self::None => Ok(()),
2648 #[cfg(feature = "draft11")]
2649 Self::Draft11(gaps) => gaps.check(group_id, object_id, block).map(|_| ()),
2650 #[cfg(feature = "draft12")]
2651 Self::Draft12(gaps) => gaps.check(group_id, object_id, block).map(|_| ()),
2652 #[cfg(feature = "draft13")]
2653 Self::Draft13(gaps) => gaps.check(group_id, object_id, block).map(|_| ()),
2654 #[cfg(feature = "draft14")]
2655 Self::Draft14(gaps) => gaps.check(group_id, object_id, block).map(|_| ()),
2656 }
2657 }
2658}
2659
2660#[derive(Debug, Clone)]
2685pub struct AnyFetchObjectReader {
2686 state: FetchReaderState,
2687 gaps: FetchGapsState,
2688}
2689
2690impl AnyFetchObjectReader {
2691 #[allow(unused_variables, unreachable_code)]
2709 pub fn new(
2710 header: &AnyFetchHeader,
2711 group_order: AnyFetchGroupOrder,
2712 ) -> Result<Self, CodecError> {
2713 let state = match header {
2714 #[cfg(feature = "draft07")]
2715 AnyFetchHeader::Draft07(_) => FetchReaderState::Draft07,
2716 #[cfg(feature = "draft08")]
2717 AnyFetchHeader::Draft08(_) => FetchReaderState::Draft08,
2718 #[cfg(feature = "draft09")]
2719 AnyFetchHeader::Draft09(_) => FetchReaderState::Draft09,
2720 #[cfg(feature = "draft10")]
2721 AnyFetchHeader::Draft10(_) => FetchReaderState::Draft10,
2722 #[cfg(feature = "draft11")]
2723 AnyFetchHeader::Draft11(_) => FetchReaderState::Draft11,
2724 #[cfg(feature = "draft12")]
2725 AnyFetchHeader::Draft12(_) => FetchReaderState::Draft12,
2726 #[cfg(feature = "draft13")]
2727 AnyFetchHeader::Draft13(_) => FetchReaderState::Draft13,
2728 #[cfg(feature = "draft14")]
2729 AnyFetchHeader::Draft14(_) => FetchReaderState::Draft14,
2730 #[cfg(feature = "draft15")]
2733 AnyFetchHeader::Draft15(_) => {
2734 FetchReaderState::Draft15(crate::draft15::data_stream::FetchObjectReader::new())
2735 }
2736 #[cfg(feature = "draft16")]
2737 AnyFetchHeader::Draft16(_) => {
2738 FetchReaderState::Draft16(crate::draft16::data_stream::FetchObjectReader::new())
2739 }
2740 #[cfg(feature = "draft17")]
2741 AnyFetchHeader::Draft17(_) => {
2742 FetchReaderState::Draft17(crate::draft17::data_stream::FetchObjectReader::new())
2743 }
2744 #[cfg(feature = "draft18")]
2745 AnyFetchHeader::Draft18(_) => FetchReaderState::Draft18(
2746 crate::draft18::data_stream::FetchObjectReader::new(match group_order {
2747 AnyFetchGroupOrder::Ascending => {
2748 crate::draft18::data_stream::GroupOrder::Ascending
2749 }
2750 AnyFetchGroupOrder::Descending => {
2751 crate::draft18::data_stream::GroupOrder::Descending
2752 }
2753 }),
2754 ),
2755 #[cfg(feature = "draft19")]
2756 AnyFetchHeader::Draft19(_) => FetchReaderState::Draft19(
2757 crate::draft19::data_stream::FetchObjectReader::new(match group_order {
2758 AnyFetchGroupOrder::Ascending => {
2759 crate::draft19::data_stream::GroupOrder::Ascending
2760 }
2761 AnyFetchGroupOrder::Descending => {
2762 crate::draft19::data_stream::GroupOrder::Descending
2763 }
2764 }),
2765 ),
2766 #[cfg(feature = "draft20")]
2767 AnyFetchHeader::Draft20(_) => FetchReaderState::Draft20(
2768 crate::draft20::data_stream::FetchObjectReader::new(match group_order {
2769 AnyFetchGroupOrder::Ascending => {
2770 crate::draft20::data_stream::GroupOrder::Ascending
2771 }
2772 AnyFetchGroupOrder::Descending => {
2773 crate::draft20::data_stream::GroupOrder::Descending
2774 }
2775 }),
2776 ),
2777 #[cfg(feature = "draft21")]
2778 AnyFetchHeader::Draft21(_) => FetchReaderState::Draft21(
2779 crate::draft21::data_stream::FetchObjectReader::new(match group_order {
2780 AnyFetchGroupOrder::Ascending => {
2781 crate::draft21::data_stream::GroupOrder::Ascending
2782 }
2783 AnyFetchGroupOrder::Descending => {
2784 crate::draft21::data_stream::GroupOrder::Descending
2785 }
2786 }),
2787 ),
2788 #[cfg(feature = "draft22")]
2789 AnyFetchHeader::Draft22(_) => FetchReaderState::Draft22(
2790 crate::draft22::data_stream::FetchObjectReader::new(match group_order {
2791 AnyFetchGroupOrder::Ascending => {
2792 crate::draft22::data_stream::GroupOrder::Ascending
2793 }
2794 AnyFetchGroupOrder::Descending => {
2795 crate::draft22::data_stream::GroupOrder::Descending
2796 }
2797 }),
2798 ),
2799 #[cfg(feature = "draft23")]
2800 AnyFetchHeader::Draft23(_) => FetchReaderState::Draft23(
2801 crate::draft23::data_stream::FetchObjectReader::new(match group_order {
2802 AnyFetchGroupOrder::Ascending => {
2803 crate::draft23::data_stream::GroupOrder::Ascending
2804 }
2805 AnyFetchGroupOrder::Descending => {
2806 crate::draft23::data_stream::GroupOrder::Descending
2807 }
2808 }),
2809 ),
2810 #[allow(unreachable_patterns)]
2811 _ => {
2812 return Err(CodecError::UnsupportedDraft(format!(
2813 "draft {:?} not enabled via feature flag",
2814 header.draft()
2815 )));
2816 }
2817 };
2818 let gaps = FetchGapsState::for_draft(header.draft());
2819 Ok(Self { state, gaps })
2820 }
2821
2822 #[allow(unreachable_code)]
2824 pub fn draft(&self) -> DraftVersion {
2825 match &self.state {
2826 #[cfg(feature = "draft07")]
2827 FetchReaderState::Draft07 => DraftVersion::Draft07,
2828 #[cfg(feature = "draft08")]
2829 FetchReaderState::Draft08 => DraftVersion::Draft08,
2830 #[cfg(feature = "draft09")]
2831 FetchReaderState::Draft09 => DraftVersion::Draft09,
2832 #[cfg(feature = "draft10")]
2833 FetchReaderState::Draft10 => DraftVersion::Draft10,
2834 #[cfg(feature = "draft11")]
2835 FetchReaderState::Draft11 => DraftVersion::Draft11,
2836 #[cfg(feature = "draft12")]
2837 FetchReaderState::Draft12 => DraftVersion::Draft12,
2838 #[cfg(feature = "draft13")]
2839 FetchReaderState::Draft13 => DraftVersion::Draft13,
2840 #[cfg(feature = "draft14")]
2841 FetchReaderState::Draft14 => DraftVersion::Draft14,
2842 #[cfg(feature = "draft15")]
2843 FetchReaderState::Draft15(_) => DraftVersion::Draft15,
2844 #[cfg(feature = "draft16")]
2845 FetchReaderState::Draft16(_) => DraftVersion::Draft16,
2846 #[cfg(feature = "draft17")]
2847 FetchReaderState::Draft17(_) => DraftVersion::Draft17,
2848 #[cfg(feature = "draft18")]
2849 FetchReaderState::Draft18(_) => DraftVersion::Draft18,
2850 #[cfg(feature = "draft19")]
2851 FetchReaderState::Draft19(_) => DraftVersion::Draft19,
2852 #[cfg(feature = "draft20")]
2853 FetchReaderState::Draft20(_) => DraftVersion::Draft20,
2854 #[cfg(feature = "draft21")]
2855 FetchReaderState::Draft21(_) => DraftVersion::Draft21,
2856 #[cfg(feature = "draft22")]
2857 FetchReaderState::Draft22(_) => DraftVersion::Draft22,
2858 #[cfg(feature = "draft23")]
2859 FetchReaderState::Draft23(_) => DraftVersion::Draft23,
2860 #[allow(unreachable_patterns)]
2861 _ => unreachable!("AnyFetchObjectReader has no enabled variants"),
2862 }
2863 }
2864
2865 #[allow(unused_variables, unreachable_code)]
2888 pub fn read_object(&mut self, buf: &mut impl Buf) -> Result<AnyFetchObject, CodecError> {
2889 let read: Result<AnyFetchObject, CodecError> = match &mut self.state {
2890 #[cfg(feature = "draft07")]
2891 FetchReaderState::Draft07 => fo07::read_object(buf),
2892 #[cfg(feature = "draft08")]
2893 FetchReaderState::Draft08 => fo08::read_object(buf),
2894 #[cfg(feature = "draft09")]
2895 FetchReaderState::Draft09 => fo09::read_object(buf),
2896 #[cfg(feature = "draft10")]
2897 FetchReaderState::Draft10 => fo10::read_object(buf),
2898 #[cfg(feature = "draft11")]
2899 FetchReaderState::Draft11 => fo11::read_object(buf),
2900 #[cfg(feature = "draft12")]
2901 FetchReaderState::Draft12 => fo12::read_object(buf),
2902 #[cfg(feature = "draft13")]
2903 FetchReaderState::Draft13 => fo13::read_object(buf),
2904 #[cfg(feature = "draft14")]
2905 FetchReaderState::Draft14 => fo14::read_object(buf),
2906 #[cfg(feature = "draft15")]
2907 FetchReaderState::Draft15(inner) => fo15::read_object(inner, buf),
2908 #[cfg(feature = "draft16")]
2909 FetchReaderState::Draft16(inner) => fo16::read_object(inner, buf),
2910 #[cfg(feature = "draft17")]
2911 FetchReaderState::Draft17(inner) => fo17::read_object(inner, buf),
2912 #[cfg(feature = "draft18")]
2913 FetchReaderState::Draft18(inner) => fo18::read_object(inner, buf),
2914 #[cfg(feature = "draft19")]
2915 FetchReaderState::Draft19(inner) => fo19::read_object(inner, buf),
2916 #[cfg(feature = "draft20")]
2917 FetchReaderState::Draft20(inner) => fo20::read_object(inner, buf),
2918 #[cfg(feature = "draft21")]
2919 FetchReaderState::Draft21(inner) => fo21::read_object(inner, buf),
2920 #[cfg(feature = "draft22")]
2921 FetchReaderState::Draft22(inner) => fo22::read_object(inner, buf),
2922 #[cfg(feature = "draft23")]
2923 FetchReaderState::Draft23(inner) => fo23::read_object(inner, buf),
2924 #[allow(unreachable_patterns)]
2925 _ => unreachable!("AnyFetchObjectReader has no enabled variants"),
2926 };
2927 let object = read?;
2928 crate::object_properties::parse_object_properties(self.draft(), &object.extension_headers)?;
2929 if object.end_of_range.is_none() {
2930 self.gaps.check(object.group_id, object.object_id, &object.extension_headers)?;
2931 }
2932 Ok(object)
2933 }
2934
2935 #[allow(unused_variables, unreachable_code)]
2948 pub fn read_object_frame(&mut self, buf: &mut impl Buf) -> Result<AnyFetchFrame, CodecError> {
2949 match &mut self.state {
2950 #[cfg(feature = "draft07")]
2951 FetchReaderState::Draft07 => fo07::read_object_meta(buf)
2952 .map(|meta| AnyFetchFrame::absolute(DraftVersion::Draft07, meta)),
2953 #[cfg(feature = "draft08")]
2954 FetchReaderState::Draft08 => fo08::read_object_meta(buf)
2955 .map(|meta| AnyFetchFrame::absolute(DraftVersion::Draft08, meta)),
2956 #[cfg(feature = "draft09")]
2957 FetchReaderState::Draft09 => fo09::read_object_meta(buf)
2958 .map(|meta| AnyFetchFrame::absolute(DraftVersion::Draft09, meta)),
2959 #[cfg(feature = "draft10")]
2960 FetchReaderState::Draft10 => fo10::read_object_meta(buf)
2961 .map(|meta| AnyFetchFrame::absolute(DraftVersion::Draft10, meta)),
2962 #[cfg(feature = "draft11")]
2963 FetchReaderState::Draft11 => fo11::read_object_meta(buf)
2964 .map(|meta| AnyFetchFrame::absolute(DraftVersion::Draft11, meta)),
2965 #[cfg(feature = "draft12")]
2966 FetchReaderState::Draft12 => fo12::read_object_meta(buf)
2967 .map(|meta| AnyFetchFrame::absolute(DraftVersion::Draft12, meta)),
2968 #[cfg(feature = "draft13")]
2969 FetchReaderState::Draft13 => fo13::read_object_meta(buf)
2970 .map(|meta| AnyFetchFrame::absolute(DraftVersion::Draft13, meta)),
2971 #[cfg(feature = "draft14")]
2972 FetchReaderState::Draft14 => fo14::read_object_meta(buf)
2973 .map(|meta| AnyFetchFrame::absolute(DraftVersion::Draft14, meta)),
2974 #[cfg(feature = "draft15")]
2975 FetchReaderState::Draft15(inner) => fo15::read_object_frame(inner, buf),
2976 #[cfg(feature = "draft16")]
2977 FetchReaderState::Draft16(inner) => fo16::read_object_frame(inner, buf),
2978 #[cfg(feature = "draft17")]
2979 FetchReaderState::Draft17(inner) => fo17::read_object_frame(inner, buf),
2980 #[cfg(feature = "draft18")]
2981 FetchReaderState::Draft18(inner) => fo18::read_object_frame(inner, buf),
2982 #[cfg(feature = "draft19")]
2983 FetchReaderState::Draft19(inner) => fo19::read_object_frame(inner, buf),
2984 #[cfg(feature = "draft20")]
2985 FetchReaderState::Draft20(inner) => fo20::read_object_frame(inner, buf),
2986 #[cfg(feature = "draft21")]
2987 FetchReaderState::Draft21(inner) => fo21::read_object_frame(inner, buf),
2988 #[cfg(feature = "draft22")]
2989 FetchReaderState::Draft22(inner) => fo22::read_object_frame(inner, buf),
2990 #[cfg(feature = "draft23")]
2991 FetchReaderState::Draft23(inner) => fo23::read_object_frame(inner, buf),
2992 #[allow(unreachable_patterns)]
2993 _ => unreachable!("AnyFetchObjectReader has no enabled variants"),
2994 }
2995 }
2996
2997 #[allow(unused_variables, unreachable_code)]
3003 pub fn read_object_meta(
3004 &mut self,
3005 buf: &mut impl Buf,
3006 ) -> Result<AnyFetchObjectMeta, CodecError> {
3007 match &mut self.state {
3008 #[cfg(feature = "draft07")]
3009 FetchReaderState::Draft07 => fo07::read_object_meta(buf),
3010 #[cfg(feature = "draft08")]
3011 FetchReaderState::Draft08 => fo08::read_object_meta(buf),
3012 #[cfg(feature = "draft09")]
3013 FetchReaderState::Draft09 => fo09::read_object_meta(buf),
3014 #[cfg(feature = "draft10")]
3015 FetchReaderState::Draft10 => fo10::read_object_meta(buf),
3016 #[cfg(feature = "draft11")]
3017 FetchReaderState::Draft11 => fo11::read_object_meta(buf),
3018 #[cfg(feature = "draft12")]
3019 FetchReaderState::Draft12 => fo12::read_object_meta(buf),
3020 #[cfg(feature = "draft13")]
3021 FetchReaderState::Draft13 => fo13::read_object_meta(buf),
3022 #[cfg(feature = "draft14")]
3023 FetchReaderState::Draft14 => fo14::read_object_meta(buf),
3024 #[cfg(feature = "draft15")]
3025 FetchReaderState::Draft15(inner) => fo15::read_object_meta(inner, buf),
3026 #[cfg(feature = "draft16")]
3027 FetchReaderState::Draft16(inner) => fo16::read_object_meta(inner, buf),
3028 #[cfg(feature = "draft17")]
3029 FetchReaderState::Draft17(inner) => fo17::read_object_meta(inner, buf),
3030 #[cfg(feature = "draft18")]
3031 FetchReaderState::Draft18(inner) => fo18::read_object_meta(inner, buf),
3032 #[cfg(feature = "draft19")]
3033 FetchReaderState::Draft19(inner) => fo19::read_object_meta(inner, buf),
3034 #[cfg(feature = "draft20")]
3035 FetchReaderState::Draft20(inner) => fo20::read_object_meta(inner, buf),
3036 #[cfg(feature = "draft21")]
3037 FetchReaderState::Draft21(inner) => fo21::read_object_meta(inner, buf),
3038 #[cfg(feature = "draft22")]
3039 FetchReaderState::Draft22(inner) => fo22::read_object_meta(inner, buf),
3040 #[cfg(feature = "draft23")]
3041 FetchReaderState::Draft23(inner) => fo23::read_object_meta(inner, buf),
3042 #[allow(unreachable_patterns)]
3043 _ => unreachable!("AnyFetchObjectReader has no enabled variants"),
3044 }
3045 }
3046}
3047
3048#[derive(Debug, Clone)]
3058enum FetchFrameShape {
3059 #[cfg(any(
3061 feature = "draft07",
3062 feature = "draft08",
3063 feature = "draft09",
3064 feature = "draft10",
3065 feature = "draft11",
3066 feature = "draft12",
3067 feature = "draft13",
3068 feature = "draft14"
3069 ))]
3070 Absolute,
3071 #[cfg(feature = "draft15")]
3072 Draft15(crate::draft15::data_stream::FetchObjectHeader),
3073 #[cfg(feature = "draft16")]
3074 Draft16(
3075 crate::draft16::data_stream::FetchObjectHeader,
3076 crate::draft16::data_stream::FetchObjectLocation,
3077 ),
3078 #[cfg(feature = "draft17")]
3079 Draft17(crate::draft17::data_stream::FetchObject),
3080 #[cfg(feature = "draft18")]
3081 Draft18(crate::draft18::data_stream::FetchObject),
3082 #[cfg(feature = "draft19")]
3083 Draft19(crate::draft19::data_stream::FetchObject),
3084 #[cfg(feature = "draft20")]
3085 Draft20(crate::draft20::data_stream::FetchObject),
3086 #[cfg(feature = "draft21")]
3087 Draft21(crate::draft21::data_stream::FetchObject),
3088 #[cfg(feature = "draft22")]
3089 Draft22(crate::draft22::data_stream::FetchObject),
3090 #[cfg(feature = "draft23")]
3091 Draft23(crate::draft23::data_stream::FetchObject),
3092}
3093
3094#[derive(Debug, Clone)]
3103pub struct AnyFetchFrame {
3104 pub meta: AnyFetchObjectMeta,
3107 draft: DraftVersion,
3108 shape: FetchFrameShape,
3109}
3110
3111impl AnyFetchFrame {
3112 #[must_use]
3118 pub fn draft(&self) -> DraftVersion {
3119 self.draft
3120 }
3121
3122 #[cfg(any(
3124 feature = "draft07",
3125 feature = "draft08",
3126 feature = "draft09",
3127 feature = "draft10",
3128 feature = "draft11",
3129 feature = "draft12",
3130 feature = "draft13",
3131 feature = "draft14"
3132 ))]
3133 fn absolute(draft: DraftVersion, meta: AnyFetchObjectMeta) -> Self {
3134 Self { meta, draft, shape: FetchFrameShape::Absolute }
3135 }
3136}
3137
3138#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3148pub enum FetchReemit {
3149 Unchanged,
3156 Reframed {
3160 framing_bytes_before: usize,
3162 framing_bytes_after: usize,
3164 },
3165}
3166
3167#[derive(Debug, Clone)]
3169enum FetchWriterState {
3170 #[cfg(any(
3174 feature = "draft07",
3175 feature = "draft08",
3176 feature = "draft09",
3177 feature = "draft10",
3178 feature = "draft11",
3179 feature = "draft12",
3180 feature = "draft13",
3181 feature = "draft14"
3182 ))]
3183 Absolute(DraftVersion),
3184 #[cfg(feature = "draft15")]
3185 Draft15(crate::draft15::data_stream::FetchObjectWriter),
3186 #[cfg(feature = "draft16")]
3187 Draft16(crate::draft16::data_stream::FetchObjectWriter),
3188 #[cfg(feature = "draft17")]
3189 Draft17(crate::draft17::data_stream::FetchObjectWriter),
3190 #[cfg(feature = "draft18")]
3191 Draft18(crate::draft18::data_stream::FetchObjectWriter),
3192 #[cfg(feature = "draft19")]
3193 Draft19(crate::draft19::data_stream::FetchObjectWriter),
3194 #[cfg(feature = "draft20")]
3195 Draft20(crate::draft20::data_stream::FetchObjectWriter),
3196 #[cfg(feature = "draft21")]
3197 Draft21(crate::draft21::data_stream::FetchObjectWriter),
3198 #[cfg(feature = "draft22")]
3199 Draft22(crate::draft22::data_stream::FetchObjectWriter),
3200 #[cfg(feature = "draft23")]
3201 Draft23(crate::draft23::data_stream::FetchObjectWriter),
3202}
3203
3204#[derive(Debug, Clone)]
3238pub struct AnyFetchObjectWriter {
3239 state: FetchWriterState,
3240}
3241
3242impl AnyFetchObjectWriter {
3243 #[allow(unused_variables, unreachable_code)]
3251 pub fn new(
3252 header: &AnyFetchHeader,
3253 group_order: AnyFetchGroupOrder,
3254 ) -> Result<Self, CodecError> {
3255 let state = match header {
3256 #[cfg(feature = "draft07")]
3257 AnyFetchHeader::Draft07(_) => FetchWriterState::Absolute(DraftVersion::Draft07),
3258 #[cfg(feature = "draft08")]
3259 AnyFetchHeader::Draft08(_) => FetchWriterState::Absolute(DraftVersion::Draft08),
3260 #[cfg(feature = "draft09")]
3261 AnyFetchHeader::Draft09(_) => FetchWriterState::Absolute(DraftVersion::Draft09),
3262 #[cfg(feature = "draft10")]
3263 AnyFetchHeader::Draft10(_) => FetchWriterState::Absolute(DraftVersion::Draft10),
3264 #[cfg(feature = "draft11")]
3265 AnyFetchHeader::Draft11(_) => FetchWriterState::Absolute(DraftVersion::Draft11),
3266 #[cfg(feature = "draft12")]
3267 AnyFetchHeader::Draft12(_) => FetchWriterState::Absolute(DraftVersion::Draft12),
3268 #[cfg(feature = "draft13")]
3269 AnyFetchHeader::Draft13(_) => FetchWriterState::Absolute(DraftVersion::Draft13),
3270 #[cfg(feature = "draft14")]
3271 AnyFetchHeader::Draft14(_) => FetchWriterState::Absolute(DraftVersion::Draft14),
3272 #[cfg(feature = "draft15")]
3273 AnyFetchHeader::Draft15(_) => {
3274 FetchWriterState::Draft15(crate::draft15::data_stream::FetchObjectWriter::new())
3275 }
3276 #[cfg(feature = "draft16")]
3277 AnyFetchHeader::Draft16(_) => {
3278 FetchWriterState::Draft16(crate::draft16::data_stream::FetchObjectWriter::new())
3279 }
3280 #[cfg(feature = "draft17")]
3281 AnyFetchHeader::Draft17(_) => {
3282 FetchWriterState::Draft17(crate::draft17::data_stream::FetchObjectWriter::new())
3283 }
3284 #[cfg(feature = "draft18")]
3285 AnyFetchHeader::Draft18(_) => FetchWriterState::Draft18(
3286 crate::draft18::data_stream::FetchObjectWriter::new(match group_order {
3287 AnyFetchGroupOrder::Ascending => {
3288 crate::draft18::data_stream::GroupOrder::Ascending
3289 }
3290 AnyFetchGroupOrder::Descending => {
3291 crate::draft18::data_stream::GroupOrder::Descending
3292 }
3293 }),
3294 ),
3295 #[cfg(feature = "draft19")]
3296 AnyFetchHeader::Draft19(_) => FetchWriterState::Draft19(
3297 crate::draft19::data_stream::FetchObjectWriter::new(match group_order {
3298 AnyFetchGroupOrder::Ascending => {
3299 crate::draft19::data_stream::GroupOrder::Ascending
3300 }
3301 AnyFetchGroupOrder::Descending => {
3302 crate::draft19::data_stream::GroupOrder::Descending
3303 }
3304 }),
3305 ),
3306 #[cfg(feature = "draft20")]
3307 AnyFetchHeader::Draft20(_) => FetchWriterState::Draft20(
3308 crate::draft20::data_stream::FetchObjectWriter::new(match group_order {
3309 AnyFetchGroupOrder::Ascending => {
3310 crate::draft20::data_stream::GroupOrder::Ascending
3311 }
3312 AnyFetchGroupOrder::Descending => {
3313 crate::draft20::data_stream::GroupOrder::Descending
3314 }
3315 }),
3316 ),
3317 #[cfg(feature = "draft21")]
3318 AnyFetchHeader::Draft21(_) => FetchWriterState::Draft21(
3319 crate::draft21::data_stream::FetchObjectWriter::new(match group_order {
3320 AnyFetchGroupOrder::Ascending => {
3321 crate::draft21::data_stream::GroupOrder::Ascending
3322 }
3323 AnyFetchGroupOrder::Descending => {
3324 crate::draft21::data_stream::GroupOrder::Descending
3325 }
3326 }),
3327 ),
3328 #[cfg(feature = "draft22")]
3329 AnyFetchHeader::Draft22(_) => FetchWriterState::Draft22(
3330 crate::draft22::data_stream::FetchObjectWriter::new(match group_order {
3331 AnyFetchGroupOrder::Ascending => {
3332 crate::draft22::data_stream::GroupOrder::Ascending
3333 }
3334 AnyFetchGroupOrder::Descending => {
3335 crate::draft22::data_stream::GroupOrder::Descending
3336 }
3337 }),
3338 ),
3339 #[cfg(feature = "draft23")]
3340 AnyFetchHeader::Draft23(_) => FetchWriterState::Draft23(
3341 crate::draft23::data_stream::FetchObjectWriter::new(match group_order {
3342 AnyFetchGroupOrder::Ascending => {
3343 crate::draft23::data_stream::GroupOrder::Ascending
3344 }
3345 AnyFetchGroupOrder::Descending => {
3346 crate::draft23::data_stream::GroupOrder::Descending
3347 }
3348 }),
3349 ),
3350 #[allow(unreachable_patterns)]
3351 _ => {
3352 return Err(CodecError::UnsupportedDraft(format!(
3353 "draft {:?} not enabled via feature flag",
3354 header.draft()
3355 )));
3356 }
3357 };
3358 Ok(Self { state })
3359 }
3360
3361 #[must_use]
3363 #[allow(unreachable_code)]
3364 pub fn draft(&self) -> DraftVersion {
3365 match &self.state {
3366 #[cfg(any(
3367 feature = "draft07",
3368 feature = "draft08",
3369 feature = "draft09",
3370 feature = "draft10",
3371 feature = "draft11",
3372 feature = "draft12",
3373 feature = "draft13",
3374 feature = "draft14"
3375 ))]
3376 FetchWriterState::Absolute(draft) => *draft,
3377 #[cfg(feature = "draft15")]
3378 FetchWriterState::Draft15(_) => DraftVersion::Draft15,
3379 #[cfg(feature = "draft16")]
3380 FetchWriterState::Draft16(_) => DraftVersion::Draft16,
3381 #[cfg(feature = "draft17")]
3382 FetchWriterState::Draft17(_) => DraftVersion::Draft17,
3383 #[cfg(feature = "draft18")]
3384 FetchWriterState::Draft18(_) => DraftVersion::Draft18,
3385 #[cfg(feature = "draft19")]
3386 FetchWriterState::Draft19(_) => DraftVersion::Draft19,
3387 #[cfg(feature = "draft20")]
3388 FetchWriterState::Draft20(_) => DraftVersion::Draft20,
3389 #[cfg(feature = "draft21")]
3390 FetchWriterState::Draft21(_) => DraftVersion::Draft21,
3391 #[cfg(feature = "draft22")]
3392 FetchWriterState::Draft22(_) => DraftVersion::Draft22,
3393 #[cfg(feature = "draft23")]
3394 FetchWriterState::Draft23(_) => DraftVersion::Draft23,
3395 #[allow(unreachable_patterns)]
3396 _ => unreachable!("AnyFetchObjectWriter has no enabled variants"),
3397 }
3398 }
3399
3400 #[allow(unused_variables)]
3432 pub fn reemit_object(
3433 &mut self,
3434 frame: &AnyFetchFrame,
3435 raw: &[u8],
3436 out: &mut impl BufMut,
3437 ) -> Result<FetchReemit, CodecError> {
3438 if frame.draft != self.draft() {
3439 return Err(CodecError::UnsupportedDraft(format!(
3440 "a draft {:?} fetch frame cannot be written onto a draft {:?} stream",
3441 frame.draft,
3442 self.draft()
3443 )));
3444 }
3445
3446 let framing_len = frame.meta.wire_len.saturating_sub(frame.meta.payload_length);
3447 let framing_len = usize::try_from(framing_len).map_err(|_| CodecError::InvalidField)?;
3448 if framing_len > raw.len() {
3449 return Err(CodecError::InvalidField);
3450 }
3451 let (framing, rest) = raw.split_at(framing_len);
3452
3453 match (&mut self.state, &frame.shape) {
3454 #[cfg(any(
3457 feature = "draft07",
3458 feature = "draft08",
3459 feature = "draft09",
3460 feature = "draft10",
3461 feature = "draft11",
3462 feature = "draft12",
3463 feature = "draft13",
3464 feature = "draft14"
3465 ))]
3466 (FetchWriterState::Absolute(_), FetchFrameShape::Absolute) => {
3467 Ok(FetchReemit::Unchanged)
3468 }
3469 #[cfg(feature = "draft15")]
3470 (FetchWriterState::Draft15(writer), FetchFrameShape::Draft15(original)) => {
3471 let reframed = writer.header_for(original)?;
3472 if reframed == *original {
3473 writer.advance(original);
3474 return Ok(FetchReemit::Unchanged);
3475 }
3476 let mut encoded = Vec::with_capacity(framing.len() + 16);
3477 reframed.encode(&mut encoded)?;
3478 writer.advance(&reframed);
3479 Ok(put_reframed(&encoded, framing.len(), rest, out))
3480 }
3481 #[cfg(feature = "draft16")]
3482 (FetchWriterState::Draft16(writer), FetchFrameShape::Draft16(original, location)) => {
3483 let reframed = writer.header_for(original, location)?;
3484 if reframed == *original {
3485 writer.advance(original, location);
3486 return Ok(FetchReemit::Unchanged);
3487 }
3488 let mut encoded = Vec::with_capacity(framing.len() + 16);
3489 reframed.encode(&mut encoded)?;
3490 writer.advance(&reframed, location);
3491 Ok(put_reframed(&encoded, framing.len(), rest, out))
3492 }
3493 #[cfg(feature = "draft17")]
3494 (FetchWriterState::Draft17(writer), FetchFrameShape::Draft17(original)) => {
3495 let reframed = writer.header_for(original)?;
3496 if reframed == original.header {
3497 writer.advance(original);
3498 return Ok(FetchReemit::Unchanged);
3499 }
3500 let mut encoded = Vec::with_capacity(framing.len() + 16);
3501 reframed.encode(&mut encoded)?;
3502 writer.advance(original);
3503 Ok(put_reframed(&encoded, framing.len(), rest, out))
3504 }
3505 #[cfg(feature = "draft18")]
3506 (FetchWriterState::Draft18(writer), FetchFrameShape::Draft18(original)) => {
3507 let reframed = writer.header_for(original)?;
3508 if reframed == original.header {
3509 writer.advance(original);
3510 return Ok(FetchReemit::Unchanged);
3511 }
3512 let mut encoded = Vec::with_capacity(framing.len() + 16);
3513 reframed.encode(&mut encoded)?;
3514 writer.advance(original);
3515 Ok(put_reframed(&encoded, framing.len(), rest, out))
3516 }
3517 #[cfg(feature = "draft19")]
3518 (FetchWriterState::Draft19(writer), FetchFrameShape::Draft19(original)) => {
3519 let reframed = writer.header_for(original)?;
3520 if reframed == original.header {
3521 writer.advance(original);
3522 return Ok(FetchReemit::Unchanged);
3523 }
3524 let mut encoded = Vec::with_capacity(framing.len() + 16);
3525 reframed.encode(&mut encoded)?;
3526 writer.advance(original);
3527 Ok(put_reframed(&encoded, framing.len(), rest, out))
3528 }
3529 #[cfg(feature = "draft20")]
3530 (FetchWriterState::Draft20(writer), FetchFrameShape::Draft20(original)) => {
3531 let reframed = writer.header_for(original)?;
3532 if reframed == original.header {
3533 writer.advance(original);
3534 return Ok(FetchReemit::Unchanged);
3535 }
3536 let mut encoded = Vec::with_capacity(framing.len() + 16);
3537 reframed.encode(&mut encoded)?;
3538 writer.advance(original);
3539 Ok(put_reframed(&encoded, framing.len(), rest, out))
3540 }
3541 #[cfg(feature = "draft21")]
3542 (FetchWriterState::Draft21(writer), FetchFrameShape::Draft21(original)) => {
3543 let reframed = writer.header_for(original)?;
3544 if reframed == original.header {
3545 writer.advance(original);
3546 return Ok(FetchReemit::Unchanged);
3547 }
3548 let mut encoded = Vec::with_capacity(framing.len() + 16);
3549 reframed.encode(&mut encoded)?;
3550 writer.advance(original);
3551 Ok(put_reframed(&encoded, framing.len(), rest, out))
3552 }
3553 #[cfg(feature = "draft22")]
3554 (FetchWriterState::Draft22(writer), FetchFrameShape::Draft22(original)) => {
3555 let reframed = writer.header_for(original)?;
3556 if reframed == original.header {
3557 writer.advance(original);
3558 return Ok(FetchReemit::Unchanged);
3559 }
3560 let mut encoded = Vec::with_capacity(framing.len() + 16);
3561 reframed.encode(&mut encoded)?;
3562 writer.advance(original);
3563 Ok(put_reframed(&encoded, framing.len(), rest, out))
3564 }
3565 #[cfg(feature = "draft23")]
3566 (FetchWriterState::Draft23(writer), FetchFrameShape::Draft23(original)) => {
3567 let reframed = writer.header_for(original)?;
3568 if reframed == original.header {
3569 writer.advance(original);
3570 return Ok(FetchReemit::Unchanged);
3571 }
3572 let mut encoded = Vec::with_capacity(framing.len() + 16);
3573 reframed.encode(&mut encoded)?;
3574 writer.advance(original);
3575 Ok(put_reframed(&encoded, framing.len(), rest, out))
3576 }
3577 #[allow(unreachable_patterns)]
3580 _ => Err(CodecError::UnsupportedDraft(format!(
3581 "no fetch writer for draft {:?}",
3582 frame.draft
3583 ))),
3584 }
3585 }
3586}
3587
3588#[cfg(any(
3591 feature = "draft15",
3592 feature = "draft16",
3593 feature = "draft17",
3594 feature = "draft18",
3595 feature = "draft19",
3596 feature = "draft20",
3597 feature = "draft21",
3598 feature = "draft22",
3599 feature = "draft23"
3600))]
3601fn put_reframed(
3602 encoded: &[u8],
3603 framing_bytes_before: usize,
3604 rest: &[u8],
3605 out: &mut impl BufMut,
3606) -> FetchReemit {
3607 out.put_slice(encoded);
3608 out.put_slice(rest);
3609 FetchReemit::Reframed { framing_bytes_before, framing_bytes_after: encoded.len() }
3610}
3611
3612macro_rules! subgroup_extensions_fn {
3618 ($name:ident, $feat:literal, $draft:ident) => {
3619 #[cfg(feature = $feat)]
3620 fn $name(header: &crate::$draft::data_stream::SubgroupHeader) -> Result<bool, CodecError> {
3621 if !header.stream_type.is_subgroup() {
3622 return Err(CodecError::InvalidField);
3623 }
3624 Ok(header.stream_type.has_extensions())
3625 }
3626 };
3627}
3628
3629subgroup_extensions_fn!(subgroup_extensions_11, "draft11", draft11);
3630subgroup_extensions_fn!(subgroup_extensions_12, "draft12", draft12);
3631subgroup_extensions_fn!(subgroup_extensions_13, "draft13", draft13);