use bytes::{Buf, BufMut};
use super::types::ObjectStatus;
use crate::error::CodecError;
use crate::varint::{Moqt17 as Wire, VarInt};
fn skip(buf: &mut impl Buf, len: u64) -> Result<(), CodecError> {
let len = usize::try_from(len).map_err(|_| CodecError::UnexpectedEnd)?;
if buf.remaining() < len {
return Err(CodecError::UnexpectedEnd);
}
buf.advance(len);
Ok(())
}
fn decoded_status(code: u64) -> Result<ObjectStatus, CodecError> {
ObjectStatus::from_u64(code).ok_or(CodecError::InvalidField)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PayloadPermission {
Permitted,
Forbidden,
}
impl PayloadPermission {
pub fn permits(self) -> bool {
matches!(self, PayloadPermission::Permitted)
}
}
const SUBGROUP_PROPERTIES_BIT: u8 = 0x01;
const SUBGROUP_ID_MODE_MASK: u8 = 0x06;
const SUBGROUP_END_OF_GROUP_BIT: u8 = 0x08;
const SUBGROUP_BASE_BIT: u8 = 0x10;
const SUBGROUP_DEFAULT_PRIORITY_BIT: u8 = 0x20;
const SUBGROUP_FORM_MASK: u8 = 0xD0;
const SUBGROUP_ID_MODE_RESERVED: u8 = 0x06;
fn subgroup_type_is_valid(raw: u64) -> bool {
raw <= 0xFF && {
let t = raw as u8;
t & SUBGROUP_FORM_MASK == SUBGROUP_BASE_BIT
&& t & SUBGROUP_ID_MODE_MASK != SUBGROUP_ID_MODE_RESERVED
}
}
fn subgroup_type_is_reserved_mode(raw: u64) -> bool {
raw <= 0xFF && {
let t = raw as u8;
t & SUBGROUP_FORM_MASK == SUBGROUP_BASE_BIT
&& t & SUBGROUP_ID_MODE_MASK == SUBGROUP_ID_MODE_RESERVED
}
}
const SETUP_STREAM_TYPE: u64 = 0x2F00;
fn wide_type_refusal(
buf: &mut impl Buf,
refusal: fn(u64) -> CodecError,
) -> Result<Option<CodecError>, CodecError> {
if !buf.has_remaining() {
return Err(CodecError::UnexpectedEnd);
}
if buf.chunk()[0] < 0x80 {
return Ok(None);
}
let raw = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
Ok(Some(refusal(raw)))
}
fn stream_type_error(raw: u64) -> CodecError {
if raw == FETCH_STREAM_TYPE || raw == SETUP_STREAM_TYPE || subgroup_type_is_valid(raw) {
CodecError::InvalidField
} else if subgroup_type_is_reserved_mode(raw) {
CodecError::InvalidTypeValue {
raw,
detail: "its SUBGROUP_ID_MODE is 0b11, which this draft reserves",
}
} else {
CodecError::UnknownStreamType(raw)
}
}
#[derive(Debug, Clone)]
pub struct SubgroupHeader {
pub header_type: u8,
pub track_alias: VarInt,
pub group_id: VarInt,
pub subgroup_id: VarInt,
pub publisher_priority: Option<u8>,
}
impl SubgroupHeader {
pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
if let Some(err) = wide_type_refusal(buf, stream_type_error)? {
return Err(err);
}
let raw = buf.get_u8() as u64;
if !subgroup_type_is_valid(raw) {
return Err(stream_type_error(raw));
}
let header_type = raw as u8;
let track_alias = VarInt::decode_moqt::<Wire>(buf)?;
let group_id = VarInt::decode_moqt::<Wire>(buf)?;
let subgroup_id_mode = (header_type & SUBGROUP_ID_MODE_MASK) >> 1;
let subgroup_id = match subgroup_id_mode {
0 => VarInt::from_u64_moqt(0),
2 => VarInt::decode_moqt::<Wire>(buf)?,
_ => VarInt::from_u64_moqt(0),
};
let publisher_priority = if header_type & SUBGROUP_DEFAULT_PRIORITY_BIT == 0 {
if buf.remaining() < 1 {
return Err(CodecError::UnexpectedEnd);
}
Some(buf.get_u8())
} else {
None
};
Ok(SubgroupHeader { header_type, track_alias, group_id, subgroup_id, publisher_priority })
}
pub fn encode(&self, buf: &mut impl BufMut) {
buf.put_u8(self.header_type);
self.track_alias.encode_moqt::<Wire>(buf);
self.group_id.encode_moqt::<Wire>(buf);
let subgroup_id_mode = (self.header_type & SUBGROUP_ID_MODE_MASK) >> 1;
if subgroup_id_mode == 2 {
self.subgroup_id.encode_moqt::<Wire>(buf);
}
if self.header_type & SUBGROUP_DEFAULT_PRIORITY_BIT == 0 {
buf.put_u8(self.publisher_priority.unwrap_or(128));
}
}
pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
if !subgroup_type_is_valid(self.header_type as u64) {
return Err(stream_type_error(self.header_type as u64));
}
self.encode(buf);
Ok(())
}
pub fn has_properties(&self) -> bool {
self.header_type & SUBGROUP_PROPERTIES_BIT != 0
}
pub fn subgroup_id_mode(&self) -> u8 {
(self.header_type & SUBGROUP_ID_MODE_MASK) >> 1
}
pub fn is_end_of_group(&self) -> bool {
self.header_type & SUBGROUP_END_OF_GROUP_BIT != 0
}
}
#[derive(Debug, Clone)]
pub struct SubgroupObject {
pub object_id: VarInt,
pub extension_headers: Vec<u8>,
pub payload_length: VarInt,
pub object_status: Option<ObjectStatus>,
pub payload: Vec<u8>,
}
impl SubgroupObject {
pub fn status(&self) -> ObjectStatus {
self.object_status.unwrap_or(ObjectStatus::Normal)
}
pub fn properties_permitted(&self) -> bool {
self.extension_headers.is_empty() || self.status() == ObjectStatus::Normal
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SubgroupObjectMeta {
pub object_id: u64,
pub extension_headers_len: u64,
pub payload_length: u64,
pub status: Option<u64>,
pub wire_len: u64,
}
impl SubgroupObjectMeta {
pub fn payload_permission(&self) -> Option<PayloadPermission> {
self.status.map(|code| {
if code == ObjectStatus::Normal.as_u64() {
PayloadPermission::Permitted
} else {
PayloadPermission::Forbidden
}
})
}
}
#[derive(Debug, Clone)]
pub struct SubgroupObjectReader {
extensions_present: bool,
prev_object_id: Option<u64>,
}
impl SubgroupObjectReader {
pub fn new(header: &SubgroupHeader) -> Self {
Self { extensions_present: header.has_properties(), prev_object_id: None }
}
pub fn read_object(&mut self, buf: &mut impl Buf) -> Result<SubgroupObject, CodecError> {
let delta = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
let object_id_val = match self.prev_object_id {
None => delta,
Some(prev) => prev
.checked_add(1)
.and_then(|v| v.checked_add(delta))
.ok_or(CodecError::ObjectIdOverflow(prev, delta))?,
};
self.prev_object_id = Some(object_id_val);
let object_id = VarInt::from_u64(object_id_val).map_err(|_| CodecError::InvalidField)?;
let extension_headers = if self.extensions_present {
let ext_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
crate::types::read_bytes(buf, ext_len)?
} else {
Vec::new()
};
let payload_length_vi = VarInt::decode_moqt::<Wire>(buf)?;
let payload_length_val = payload_length_vi.into_inner() as usize;
let (object_status, payload) = if payload_length_val == 0 {
let status = VarInt::decode_moqt::<Wire>(buf)?;
(Some(decoded_status(status.into_inner())?), Vec::new())
} else {
let payload = crate::types::read_bytes(buf, payload_length_val)?;
(None, payload)
};
Ok(SubgroupObject {
object_id,
extension_headers,
payload_length: payload_length_vi,
object_status,
payload,
})
}
pub fn read_object_meta(
&mut self,
buf: &mut impl Buf,
) -> Result<SubgroupObjectMeta, CodecError> {
let start = buf.remaining();
let delta = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
let object_id_val = match self.prev_object_id {
None => delta,
Some(prev) => prev
.checked_add(1)
.and_then(|v| v.checked_add(delta))
.ok_or(CodecError::ObjectIdOverflow(prev, delta))?,
};
self.prev_object_id = Some(object_id_val);
let object_id =
VarInt::from_u64(object_id_val).map_err(|_| CodecError::InvalidField)?.into_inner();
let extension_headers_len = if self.extensions_present {
let ext_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
skip(buf, ext_len)?;
ext_len
} else {
0
};
let payload_length = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
let status = if payload_length == 0 {
let code = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
Some(decoded_status(code)?.as_u64())
} else {
skip(buf, payload_length)?;
None
};
Ok(SubgroupObjectMeta {
object_id,
extension_headers_len,
payload_length,
status,
wire_len: (start - buf.remaining()) as u64,
})
}
pub fn write_object(
&mut self,
object: &SubgroupObject,
buf: &mut impl BufMut,
) -> Result<(), CodecError> {
let declared = object.payload_length.into_inner();
if declared != object.payload.len() as u64 {
return Err(CodecError::InvalidField);
}
let oid = object.object_id.into_inner();
let delta = match self.prev_object_id {
None => oid,
Some(prev) => oid
.checked_sub(prev)
.and_then(|v| v.checked_sub(1))
.ok_or(CodecError::InvalidField)?,
};
VarInt::from_u64(delta).map_err(|_| CodecError::InvalidField)?.encode_moqt::<Wire>(buf);
if self.extensions_present {
VarInt::from_u64(object.extension_headers.len() as u64)
.map_err(|_| CodecError::InvalidField)?
.encode_moqt::<Wire>(buf);
buf.put_slice(&object.extension_headers);
}
object.payload_length.encode_moqt::<Wire>(buf);
if object.payload_length.into_inner() == 0 {
let status = object.object_status.unwrap_or(ObjectStatus::Normal);
VarInt::from_u64_moqt(status.as_u64()).encode_moqt::<Wire>(buf);
} else {
buf.put_slice(&object.payload);
}
self.prev_object_id = Some(oid);
Ok(())
}
}
const DATAGRAM_PROPERTIES_BIT: u8 = 0x01;
const DATAGRAM_END_OF_GROUP_BIT: u8 = 0x02;
const DATAGRAM_ZERO_OBJECT_ID_BIT: u8 = 0x04;
const DATAGRAM_DEFAULT_PRIORITY_BIT: u8 = 0x08;
const DATAGRAM_STATUS_BIT: u8 = 0x20;
const DATAGRAM_FORM_MASK: u8 = 0xD0;
fn datagram_type_is_valid(raw: u64) -> bool {
raw <= 0xFF && {
let t = raw as u8;
t & DATAGRAM_FORM_MASK == 0
&& t & (DATAGRAM_STATUS_BIT | DATAGRAM_END_OF_GROUP_BIT)
!= (DATAGRAM_STATUS_BIT | DATAGRAM_END_OF_GROUP_BIT)
}
}
fn datagram_type_error(raw: u64) -> CodecError {
if datagram_type_is_valid(raw) {
CodecError::InvalidField
} else if raw <= 0xFF
&& raw as u8 & DATAGRAM_FORM_MASK == 0
&& raw as u8 & (DATAGRAM_STATUS_BIT | DATAGRAM_END_OF_GROUP_BIT)
== (DATAGRAM_STATUS_BIT | DATAGRAM_END_OF_GROUP_BIT)
{
CodecError::InvalidTypeValue {
raw,
detail: "it sets both the STATUS bit and the END_OF_GROUP bit",
}
} else {
CodecError::UnknownDatagramType(raw)
}
}
#[derive(Debug, Clone)]
pub struct DatagramHeader {
pub datagram_type: u8,
pub track_alias: VarInt,
pub group_id: VarInt,
pub object_id: VarInt,
pub publisher_priority: Option<u8>,
pub properties: Vec<u8>,
pub object_status: Option<ObjectStatus>,
}
impl DatagramHeader {
pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
if let Some(err) = wide_type_refusal(buf, datagram_type_error)? {
return Err(err);
}
let raw = buf.get_u8() as u64;
if !datagram_type_is_valid(raw) {
return Err(datagram_type_error(raw));
}
let datagram_type = raw as u8;
let track_alias = VarInt::decode_moqt::<Wire>(buf)?;
let group_id = VarInt::decode_moqt::<Wire>(buf)?;
let object_id = if datagram_type & DATAGRAM_ZERO_OBJECT_ID_BIT != 0 {
VarInt::from_usize(0)
} else {
VarInt::decode_moqt::<Wire>(buf)?
};
let publisher_priority = if datagram_type & DATAGRAM_DEFAULT_PRIORITY_BIT == 0 {
if buf.remaining() < 1 {
return Err(CodecError::UnexpectedEnd);
}
Some(buf.get_u8())
} else {
None
};
let properties = if datagram_type & DATAGRAM_PROPERTIES_BIT != 0 {
let props_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
crate::types::read_bytes(buf, props_len)?
} else {
Vec::new()
};
let object_status = if datagram_type & DATAGRAM_STATUS_BIT != 0 {
if buf.remaining() < 1 {
return Err(CodecError::UnexpectedEnd);
}
let status = buf.get_u8();
Some(decoded_status(status as u64)?)
} else {
None
};
Ok(DatagramHeader {
datagram_type,
track_alias,
group_id,
object_id,
publisher_priority,
properties,
object_status,
})
}
pub fn decode_object(buf: &mut impl Buf) -> Result<(Self, Vec<u8>), CodecError> {
let header = Self::decode(buf)?;
let payload = crate::types::read_bytes(buf, buf.remaining())?;
if !payload.is_empty() && !header.permits_payload() {
return Err(CodecError::PayloadNotPermitted {
status: header.status().as_u64(),
len: payload.len(),
detail: if header.has_status() {
"its type states a status in place of a payload"
} else {
"its status is registered as forbidding one"
},
});
}
Ok((header, payload))
}
pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
if !datagram_type_is_valid(self.datagram_type as u64) {
return Err(datagram_type_error(self.datagram_type as u64));
}
if !self.has_status() && matches!(self.object_status, Some(s) if s != ObjectStatus::Normal)
{
return Err(CodecError::InvalidField);
}
if !self.properties_block_well_formed() || !self.properties_permitted() {
return Err(CodecError::InvalidField);
}
self.encode(buf);
Ok(())
}
pub fn encode(&self, buf: &mut impl BufMut) {
buf.put_u8(self.datagram_type);
self.track_alias.encode_moqt::<Wire>(buf);
self.group_id.encode_moqt::<Wire>(buf);
if self.datagram_type & DATAGRAM_ZERO_OBJECT_ID_BIT == 0 {
self.object_id.encode_moqt::<Wire>(buf);
}
if self.datagram_type & DATAGRAM_DEFAULT_PRIORITY_BIT == 0 {
buf.put_u8(self.publisher_priority.unwrap_or(128));
}
if self.datagram_type & DATAGRAM_PROPERTIES_BIT != 0 {
VarInt::from_usize(self.properties.len()).encode_moqt::<Wire>(buf);
buf.put_slice(&self.properties);
}
if self.datagram_type & DATAGRAM_STATUS_BIT != 0 {
buf.put_u8(self.object_status.unwrap_or(ObjectStatus::Normal).as_u8());
}
}
pub fn is_end_of_group(&self) -> bool {
self.datagram_type & DATAGRAM_END_OF_GROUP_BIT != 0
}
pub fn has_status(&self) -> bool {
self.datagram_type & DATAGRAM_STATUS_BIT != 0
}
pub fn has_properties(&self) -> bool {
self.datagram_type & DATAGRAM_PROPERTIES_BIT != 0
}
pub fn status(&self) -> ObjectStatus {
self.object_status.unwrap_or(ObjectStatus::Normal)
}
pub fn properties_permitted(&self) -> bool {
self.properties.is_empty() || self.status() == ObjectStatus::Normal
}
pub fn properties_block_well_formed(&self) -> bool {
self.has_properties() != self.properties.is_empty()
}
pub fn permits_payload(&self) -> bool {
if self.has_status() {
return false;
}
match self.object_status {
None => true,
Some(status) => status == ObjectStatus::Normal,
}
}
}
const FETCH_STREAM_TYPE: u64 = 0x05;
#[derive(Debug, Clone)]
pub struct FetchHeader {
pub request_id: VarInt,
}
impl FetchHeader {
pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
let stream_type = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
if stream_type != FETCH_STREAM_TYPE {
return Err(stream_type_error(stream_type));
}
let request_id = VarInt::decode_moqt::<Wire>(buf)?;
Ok(FetchHeader { request_id })
}
pub fn encode(&self, buf: &mut impl BufMut) {
VarInt::from_usize(FETCH_STREAM_TYPE as usize).encode_moqt::<Wire>(buf);
self.request_id.encode_moqt::<Wire>(buf);
}
}
const FETCH_SUBGROUP_MODE_MASK: u64 = 0x03;
const FETCH_SUBGROUP_MODE_PRIOR: u64 = 0x01;
const FETCH_SUBGROUP_MODE_PRIOR_PLUS_ONE: u64 = 0x02;
const FETCH_SUBGROUP_MODE_EXPLICIT: u64 = 0x03;
const FETCH_OBJECT_ID_BIT: u64 = 0x04;
const FETCH_GROUP_ID_BIT: u64 = 0x08;
const FETCH_PRIORITY_BIT: u64 = 0x10;
const FETCH_PROPERTIES_BIT: u64 = 0x20;
const FETCH_DATAGRAM_BIT: u64 = 0x40;
const FETCH_FLAGS_BIT_FORM_MAX: u64 = 0x7f;
const FETCH_END_OF_NON_EXISTENT_RANGE: u64 = 0x8c;
const FETCH_END_OF_UNKNOWN_RANGE: u64 = 0x10c;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EndOfRange {
NonExistent,
Unknown,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FetchObjectHeader {
pub serialization_flags: VarInt,
pub group_id: Option<VarInt>,
pub subgroup_id: Option<VarInt>,
pub object_id: Option<VarInt>,
pub publisher_priority: Option<u8>,
pub properties: Vec<u8>,
pub payload_length: VarInt,
}
impl FetchObjectHeader {
pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
let serialization_flags = VarInt::decode_moqt::<Wire>(buf)?;
let flags = serialization_flags.into_inner();
if !fetch_flags_are_defined(flags) {
return Err(CodecError::InvalidField);
}
let group_id =
if fetch_has_group_id(flags) { Some(VarInt::decode_moqt::<Wire>(buf)?) } else { None };
let subgroup_id = if fetch_has_subgroup_id(flags) {
Some(VarInt::decode_moqt::<Wire>(buf)?)
} else {
None
};
let object_id =
if fetch_has_object_id(flags) { Some(VarInt::decode_moqt::<Wire>(buf)?) } else { None };
let publisher_priority = if fetch_has_priority(flags) {
if buf.remaining() < 1 {
return Err(CodecError::UnexpectedEnd);
}
Some(buf.get_u8())
} else {
None
};
let properties = if fetch_has_properties(flags) {
let props_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
crate::types::read_bytes(buf, props_len)?
} else {
Vec::new()
};
let payload_length = VarInt::decode_moqt::<Wire>(buf)?;
Ok(FetchObjectHeader {
serialization_flags,
group_id,
subgroup_id,
object_id,
publisher_priority,
properties,
payload_length,
})
}
pub fn encode(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
let flags = self.serialization_flags.into_inner();
if !fetch_flags_are_defined(flags) {
return Err(CodecError::InvalidField);
}
if fetch_has_group_id(flags) != self.group_id.is_some()
|| fetch_has_subgroup_id(flags) != self.subgroup_id.is_some()
|| fetch_has_object_id(flags) != self.object_id.is_some()
|| fetch_has_priority(flags) != self.publisher_priority.is_some()
|| (!fetch_has_properties(flags) && !self.properties.is_empty())
{
return Err(CodecError::InvalidField);
}
self.serialization_flags.encode_moqt::<Wire>(buf);
if let Some(group_id) = self.group_id {
group_id.encode_moqt::<Wire>(buf);
}
if let Some(subgroup_id) = self.subgroup_id {
subgroup_id.encode_moqt::<Wire>(buf);
}
if let Some(object_id) = self.object_id {
object_id.encode_moqt::<Wire>(buf);
}
if let Some(priority) = self.publisher_priority {
buf.put_u8(priority);
}
if fetch_has_properties(flags) {
VarInt::from_usize(self.properties.len()).encode_moqt::<Wire>(buf);
buf.put_slice(&self.properties);
}
self.payload_length.encode_moqt::<Wire>(buf);
Ok(())
}
pub fn end_of_range(&self) -> Option<EndOfRange> {
match self.serialization_flags.into_inner() {
FETCH_END_OF_NON_EXISTENT_RANGE => Some(EndOfRange::NonExistent),
FETCH_END_OF_UNKNOWN_RANGE => Some(EndOfRange::Unknown),
_ => None,
}
}
pub fn subgroup_id_mode(&self) -> u64 {
self.serialization_flags.into_inner() & FETCH_SUBGROUP_MODE_MASK
}
pub fn is_datagram(&self) -> bool {
self.serialization_flags.into_inner() & FETCH_DATAGRAM_BIT != 0
}
pub fn has_properties(&self) -> bool {
fetch_has_properties(self.serialization_flags.into_inner())
}
pub fn references_prior_object(&self) -> bool {
if self.end_of_range().is_some() {
return false;
}
let flags = self.serialization_flags.into_inner();
let inherits_subgroup = !self.is_datagram()
&& matches!(
flags & FETCH_SUBGROUP_MODE_MASK,
FETCH_SUBGROUP_MODE_PRIOR | FETCH_SUBGROUP_MODE_PRIOR_PLUS_ONE
);
!fetch_has_group_id(flags)
|| !fetch_has_object_id(flags)
|| !fetch_has_priority(flags)
|| inherits_subgroup
}
}
fn fetch_flags_are_defined(flags: u64) -> bool {
flags <= FETCH_FLAGS_BIT_FORM_MAX
|| flags == FETCH_END_OF_NON_EXISTENT_RANGE
|| flags == FETCH_END_OF_UNKNOWN_RANGE
}
fn fetch_has_group_id(flags: u64) -> bool {
flags & FETCH_GROUP_ID_BIT != 0
}
fn fetch_has_subgroup_id(flags: u64) -> bool {
flags & FETCH_DATAGRAM_BIT == 0
&& flags & FETCH_SUBGROUP_MODE_MASK == FETCH_SUBGROUP_MODE_EXPLICIT
}
fn fetch_has_object_id(flags: u64) -> bool {
flags & FETCH_OBJECT_ID_BIT != 0
}
fn fetch_has_priority(flags: u64) -> bool {
flags & FETCH_PRIORITY_BIT != 0
}
fn fetch_has_properties(flags: u64) -> bool {
flags & FETCH_PROPERTIES_BIT != 0
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FetchObject {
pub header: FetchObjectHeader,
pub group_id: u64,
pub subgroup_id: Option<u64>,
pub object_id: u64,
pub publisher_priority: Option<u8>,
}
#[derive(Debug, Clone, Default)]
pub struct FetchObjectReader {
prior_location: Option<(u64, u64)>,
prior_subgroup_id: Option<u64>,
prior_publisher_priority: Option<u8>,
}
impl FetchObjectReader {
pub fn new() -> Self {
Self::default()
}
pub fn read_object_header(&mut self, buf: &mut impl Buf) -> Result<FetchObject, CodecError> {
let header = FetchObjectHeader::decode(buf)?;
if header.end_of_range().is_some() {
let group_id = header.group_id.ok_or(CodecError::InvalidField)?.into_inner();
let object_id = header.object_id.ok_or(CodecError::InvalidField)?.into_inner();
self.prior_location = Some((group_id, object_id));
let publisher_priority = self.prior_publisher_priority;
return Ok(FetchObject {
header,
group_id,
subgroup_id: None,
object_id,
publisher_priority,
});
}
let group_id = match header.group_id {
Some(v) => v.into_inner(),
None => self.prior_location.ok_or(CodecError::InvalidField)?.0,
};
let object_id = match header.object_id {
Some(v) => v.into_inner(),
None => self
.prior_location
.ok_or(CodecError::InvalidField)?
.1
.checked_add(1)
.ok_or(CodecError::InvalidField)?,
};
let subgroup_id = if header.is_datagram() {
None
} else {
Some(match header.subgroup_id_mode() {
0x00 => 0,
0x01 => self.prior_subgroup_id.ok_or(CodecError::InvalidField)?,
0x02 => self
.prior_subgroup_id
.ok_or(CodecError::InvalidField)?
.checked_add(1)
.ok_or(CodecError::InvalidField)?,
_ => header.subgroup_id.ok_or(CodecError::InvalidField)?.into_inner(),
})
};
let publisher_priority = match header.publisher_priority {
Some(p) => p,
None => self.prior_publisher_priority.ok_or(CodecError::InvalidField)?,
};
self.prior_location = Some((group_id, object_id));
if let Some(subgroup_id) = subgroup_id {
self.prior_subgroup_id = Some(subgroup_id);
}
self.prior_publisher_priority = Some(publisher_priority);
Ok(FetchObject {
header,
group_id,
subgroup_id,
object_id,
publisher_priority: Some(publisher_priority),
})
}
}
#[derive(Debug, Clone, Default)]
pub struct FetchObjectWriter {
prior_location: Option<(u64, u64)>,
prior_subgroup_id: Option<u64>,
prior_publisher_priority: Option<u8>,
}
impl FetchObjectWriter {
pub fn new() -> Self {
Self::default()
}
pub fn header_for(&self, frame: &FetchObject) -> Result<FetchObjectHeader, CodecError> {
let original = &frame.header;
if original.end_of_range().is_some() {
return Ok(FetchObjectHeader {
serialization_flags: original.serialization_flags,
group_id: Some(VarInt::from_u64(frame.group_id)?),
subgroup_id: None,
object_id: Some(VarInt::from_u64(frame.object_id)?),
publisher_priority: None,
properties: Vec::new(),
payload_length: original.payload_length,
});
}
let (group_id, object_id) = self.identity_fields(frame, original)?;
let (subgroup_mode, subgroup_id) = self.subgroup_field(frame, original)?;
let publisher_priority = self.priority_field(frame, original)?;
let flags = original.serialization_flags.into_inner();
let mut new_flags = subgroup_mode;
if flags & FETCH_DATAGRAM_BIT != 0 {
new_flags |= FETCH_DATAGRAM_BIT;
}
if group_id.is_some() {
new_flags |= FETCH_GROUP_ID_BIT;
}
if object_id.is_some() {
new_flags |= FETCH_OBJECT_ID_BIT;
}
if publisher_priority.is_some() {
new_flags |= FETCH_PRIORITY_BIT;
}
if flags & FETCH_PROPERTIES_BIT != 0 {
new_flags |= FETCH_PROPERTIES_BIT;
}
Ok(FetchObjectHeader {
serialization_flags: VarInt::from_u64(new_flags)?,
group_id,
subgroup_id,
object_id,
publisher_priority,
properties: original.properties.clone(),
payload_length: original.payload_length,
})
}
fn identity_fields(
&self,
frame: &FetchObject,
original: &FetchObjectHeader,
) -> Result<(Option<VarInt>, Option<VarInt>), CodecError> {
let group_id = match self.prior_location {
Some((prior_group, _))
if original.group_id.is_none() && prior_group == frame.group_id =>
{
None
}
_ => Some(VarInt::from_u64(frame.group_id)?),
};
let object_id = match self.prior_location {
Some((_, prior_object))
if original.object_id.is_none()
&& prior_object.checked_add(1) == Some(frame.object_id) =>
{
None
}
_ => Some(VarInt::from_u64(frame.object_id)?),
};
Ok((group_id, object_id))
}
fn subgroup_field(
&self,
frame: &FetchObject,
original: &FetchObjectHeader,
) -> Result<(u64, Option<VarInt>), CodecError> {
if original.is_datagram() {
return Ok((
original.serialization_flags.into_inner() & FETCH_SUBGROUP_MODE_MASK,
None,
));
}
let subgroup_id = frame.subgroup_id.ok_or(CodecError::InvalidField)?;
let inherits = self.prior_subgroup_id == Some(subgroup_id);
let successor =
self.prior_subgroup_id.is_some_and(|p| p.checked_add(1) == Some(subgroup_id));
let kept = match original.subgroup_id_mode() {
0x00 if subgroup_id == 0 => Some((0x00, None)),
FETCH_SUBGROUP_MODE_PRIOR if inherits => Some((FETCH_SUBGROUP_MODE_PRIOR, None)),
FETCH_SUBGROUP_MODE_PRIOR_PLUS_ONE if successor => {
Some((FETCH_SUBGROUP_MODE_PRIOR_PLUS_ONE, None))
}
FETCH_SUBGROUP_MODE_EXPLICIT => Some((FETCH_SUBGROUP_MODE_EXPLICIT, Some(subgroup_id))),
_ => None,
};
let (mode, explicit) = match kept {
Some(pair) => pair,
None if subgroup_id == 0 => (0x00, None),
None if inherits => (FETCH_SUBGROUP_MODE_PRIOR, None),
None if successor => (FETCH_SUBGROUP_MODE_PRIOR_PLUS_ONE, None),
None => (FETCH_SUBGROUP_MODE_EXPLICIT, Some(subgroup_id)),
};
Ok((mode, explicit.map(VarInt::from_u64).transpose()?))
}
fn priority_field(
&self,
frame: &FetchObject,
original: &FetchObjectHeader,
) -> Result<Option<u8>, CodecError> {
let priority = frame.publisher_priority.ok_or(CodecError::InvalidField)?;
if original.publisher_priority.is_some() || self.prior_publisher_priority != Some(priority)
{
return Ok(Some(priority));
}
Ok(None)
}
pub fn write_object_header(
&mut self,
frame: &FetchObject,
out: &mut impl BufMut,
) -> Result<FetchObjectHeader, CodecError> {
let header = self.header_for(frame)?;
header.encode(out)?;
self.advance(frame);
Ok(header)
}
pub fn advance(&mut self, frame: &FetchObject) {
self.prior_location = Some((frame.group_id, frame.object_id));
if let Some(subgroup_id) = frame.subgroup_id {
self.prior_subgroup_id = Some(subgroup_id);
}
if frame.header.end_of_range().is_none() {
if let Some(priority) = frame.publisher_priority {
self.prior_publisher_priority = Some(priority);
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
const VECTORS: &[&str] = &[
"100100800004deadbeef",
"100100800004deadbeef0002cafe",
"3001000004deadbeef",
"11010080000004deadbeef",
"180105800004deadbeef",
"120100800504deadbeef",
"1101008000043c02020104deadbeef",
"100100800004deadbeef000003",
"10010080000004",
"11010080000004deadbeef000002cafe",
"1101008000023c0204deadbeef00023c0302cafe",
"1101008000023c010003",
];
fn vi(v: u64) -> VarInt {
VarInt::from_u64_moqt(v)
}
fn hex(s: &str) -> Vec<u8> {
(0..s.len()).step_by(2).map(|i| u8::from_str_radix(&s[i..i + 2], 16).unwrap()).collect()
}
fn decode_all(bytes: &[u8]) -> (SubgroupHeader, Vec<SubgroupObject>) {
let mut cursor = bytes;
let header = SubgroupHeader::decode(&mut cursor)
.unwrap_or_else(|e| panic!("header decode failed: {e:?}"));
let mut reader = SubgroupObjectReader::new(&header);
let mut objects = Vec::new();
while cursor.has_remaining() {
objects.push(
reader
.read_object(&mut cursor)
.unwrap_or_else(|e| panic!("object {} decode failed: {e:?}", objects.len())),
);
}
(header, objects)
}
fn encode_all(header: &SubgroupHeader, objects: &[SubgroupObject]) -> Vec<u8> {
let mut buf = Vec::new();
header.encode(&mut buf);
let mut writer = SubgroupObjectReader::new(header);
for o in objects {
writer.write_object(o, &mut buf).unwrap_or_else(|e| panic!("write failed: {e:?}"));
}
buf
}
fn object(id: u64, extensions: Vec<u8>, payload: Vec<u8>) -> SubgroupObject {
SubgroupObject {
object_id: vi(id),
extension_headers: extensions,
payload_length: vi(payload.len() as u64),
object_status: None,
payload,
}
}
#[test]
fn two_objects_with_properties_have_distinct_ids() {
let bytes = hex("11010080000004deadbeef000002cafe");
let (header, objects) = decode_all(&bytes);
assert!(header.has_properties());
assert_eq!(objects.len(), 2);
assert_eq!(objects[0].object_id.into_inner(), 0);
assert_eq!(objects[1].object_id.into_inner(), 1);
assert_eq!(objects[0].payload, hex("deadbeef"));
assert_eq!(objects[1].payload, hex("cafe"));
assert!(objects.iter().all(|o| o.extension_headers.is_empty()));
}
#[test]
fn deltas_resolve_sparse_ids() {
let header = SubgroupHeader::decode(&mut &hex("100100800004deadbeef")[..]).unwrap();
let objects: Vec<_> =
[3u64, 4, 40].iter().map(|&id| object(id, vec![], vec![0xAA, id as u8])).collect();
let (_, decoded) = decode_all(&encode_all(&header, &objects));
let ids: Vec<u64> = decoded.iter().map(|o| o.object_id.into_inner()).collect();
assert_eq!(ids, vec![3, 4, 40]);
}
#[test]
fn write_rejects_non_increasing_ids() {
let header = SubgroupHeader::decode(&mut &hex("100100800004deadbeef")[..]).unwrap();
let mut writer = SubgroupObjectReader::new(&header);
let mut buf = Vec::new();
writer.write_object(&object(7, vec![], vec![0x01]), &mut buf).unwrap();
for id in [7u64, 6, 0] {
let err = writer.write_object(&object(id, vec![], vec![0x01]), &mut buf).unwrap_err();
assert!(matches!(err, CodecError::InvalidField), "id {id} gave {err:?}");
}
}
#[test]
fn eliding_an_object_renumbers_its_successor() {
let header = SubgroupHeader::decode(&mut &hex("100100800004deadbeef")[..]).unwrap();
let all: Vec<_> = (0..5u64).map(|id| object(id, vec![], vec![id as u8])).collect();
for elided in 0..5u64 {
let kept: Vec<_> =
all.iter().filter(|o| o.object_id.into_inner() != elided).cloned().collect();
let (_, decoded) = decode_all(&encode_all(&header, &kept));
let ids: Vec<u64> = decoded.iter().map(|o| o.object_id.into_inner()).collect();
let expected: Vec<u64> = (0..5u64).filter(|&i| i != elided).collect();
assert_eq!(ids, expected, "eliding object {elided}");
}
}
#[test]
fn properties_blob_excludes_its_length_prefix() {
let bytes = hex("1101008000023c0204deadbeef00023c0302cafe");
let (_, objects) = decode_all(&bytes);
assert_eq!(objects.len(), 2);
assert_eq!(objects[0].object_id.into_inner(), 0);
assert_eq!(objects[1].object_id.into_inner(), 1);
assert_eq!(objects[0].extension_headers, hex("3c02"));
assert_eq!(objects[1].extension_headers, hex("3c03"));
assert_eq!(objects[0].payload, hex("deadbeef"));
assert_eq!(objects[1].payload, hex("cafe"));
}
#[test]
fn status_object_carries_its_properties_block() {
let (_, objects) = decode_all(&hex("1101008000023c010003"));
assert_eq!(objects.len(), 1);
assert_eq!(objects[0].extension_headers, hex("3c01"));
assert_eq!(objects[0].payload_length.into_inner(), 0);
assert_eq!(objects[0].object_status.map(ObjectStatus::as_u64), Some(3));
assert!(objects[0].payload.is_empty());
}
#[test]
fn vectors_re_encode_byte_identically() {
for vector in VECTORS {
let bytes = hex(vector);
let (header, objects) = decode_all(&bytes);
assert_eq!(encode_all(&header, &objects), bytes, "[{vector}] re-encode");
}
}
#[test]
fn meta_matches_read_object() {
for vector in VECTORS {
let bytes = hex(vector);
let mut cursor = &bytes[..];
let header = SubgroupHeader::decode(&mut cursor).unwrap();
let mut full_reader = SubgroupObjectReader::new(&header);
let mut meta_reader = SubgroupObjectReader::new(&header);
let mut full_cursor = cursor;
let mut meta_cursor = cursor;
while meta_cursor.has_remaining() {
let before = meta_cursor.remaining();
let object = full_reader.read_object(&mut full_cursor).unwrap();
let meta = meta_reader.read_object_meta(&mut meta_cursor).unwrap();
assert_eq!(meta.object_id, object.object_id.into_inner(), "[{vector}]");
assert_eq!(
meta.extension_headers_len,
object.extension_headers.len() as u64,
"[{vector}]"
);
assert_eq!(meta.payload_length, object.payload_length.into_inner(), "[{vector}]");
assert_eq!(
meta.status,
object.object_status.map(ObjectStatus::as_u64),
"[{vector}]"
);
assert_eq!(meta.wire_len, (before - meta_cursor.remaining()) as u64, "[{vector}]");
assert_eq!(full_cursor.remaining(), meta_cursor.remaining(), "[{vector}]");
}
}
}
#[test]
fn short_buffers_report_unexpected_end() {
let bytes = hex("1101008000023c0204deadbeef00023c0302cafe");
let mut cursor = &bytes[..];
let header = SubgroupHeader::decode(&mut cursor).unwrap();
let objects_start = bytes.len() - cursor.len();
for cut in objects_start..bytes.len() {
let mut reader = SubgroupObjectReader::new(&header);
let mut meta_reader = SubgroupObjectReader::new(&header);
let mut cursor = &bytes[objects_start..cut];
let mut meta_cursor = cursor;
while cursor.has_remaining() {
if let Err(err) = reader.read_object(&mut cursor) {
assert!(
matches!(err, CodecError::UnexpectedEnd | CodecError::VarInt(_)),
"cut {cut} gave {err:?}"
);
break;
}
}
while meta_cursor.has_remaining() {
if let Err(err) = meta_reader.read_object_meta(&mut meta_cursor) {
assert!(
matches!(err, CodecError::UnexpectedEnd | CodecError::VarInt(_)),
"cut {cut} gave {err:?}"
);
break;
}
}
}
}
fn subgroup_status_stream(status: u64) -> Vec<u8> {
vec![0x10, 0x01, 0x00, 0x80, 0x00, 0x00, status as u8]
}
fn status_datagram(status: u64) -> Vec<u8> {
vec![0x20, 0x01, 0x00, 0x00, 0x80, status as u8]
}
fn status_object(status: Option<ObjectStatus>) -> SubgroupObject {
SubgroupObject {
object_id: vi(0),
extension_headers: Vec::new(),
payload_length: vi(0),
object_status: status,
payload: Vec::new(),
}
}
fn status_datagram_header(status: Option<ObjectStatus>) -> DatagramHeader {
DatagramHeader {
datagram_type: 0x20,
track_alias: vi(1),
group_id: vi(0),
object_id: vi(0),
publisher_priority: Some(128),
properties: Vec::new(),
object_status: status,
}
}
#[test]
fn every_assigned_status_survives_a_round_trip() {
let header = SubgroupHeader::decode(&mut &hex("100100800004deadbeef")[..]).unwrap();
for &status in ObjectStatus::ALL {
let mut buf = Vec::new();
SubgroupObjectReader::new(&header)
.write_object(&status_object(Some(status)), &mut buf)
.unwrap_or_else(|e| panic!("write_object refused {status:?}: {e:?}"));
let mut cursor = &buf[..];
let object =
SubgroupObjectReader::new(&header).read_object(&mut cursor).unwrap_or_else(|e| {
panic!("read_object refused the bytes written for {status:?}: {e:?}")
});
assert_eq!(object.object_status, Some(status), "subgroup object status");
assert!(!cursor.has_remaining(), "{status:?}: bytes left over after read_object");
let meta =
SubgroupObjectReader::new(&header).read_object_meta(&mut &buf[..]).unwrap_or_else(
|e| panic!("read_object_meta refused the bytes written for {status:?}: {e:?}"),
);
assert_eq!(meta.status, Some(status.as_u64()), "subgroup meta status");
let mut datagram = Vec::new();
status_datagram_header(Some(status)).encode(&mut datagram);
let decoded = DatagramHeader::decode(&mut &datagram[..]).unwrap_or_else(|e| {
panic!("datagram decode refused the bytes written for {status:?}: {e:?}")
});
assert_eq!(decoded.object_status, Some(status), "datagram object status");
}
let mut buf = Vec::new();
SubgroupObjectReader::new(&header).write_object(&status_object(None), &mut buf).unwrap();
let object = SubgroupObjectReader::new(&header)
.read_object(&mut &buf[..])
.expect("a zero-length object with no status must still decode");
assert_eq!(object.object_status, Some(ObjectStatus::Normal));
let mut datagram = Vec::new();
status_datagram_header(None).encode(&mut datagram);
let decoded = DatagramHeader::decode(&mut &datagram[..])
.expect("a status datagram with no status must still decode");
assert_eq!(decoded.object_status, Some(ObjectStatus::Normal));
}
#[test]
fn the_encoder_writes_exactly_the_frames_the_decoder_accepts() {
for code in 0x00u64..=0x3f {
let assigned = ObjectStatus::ALL.iter().copied().find(|s| s.as_u64() == code);
let stream = subgroup_status_stream(code);
let mut cursor: &[u8] = &stream;
let header = SubgroupHeader::decode(&mut cursor).unwrap();
let objects = cursor;
let read = SubgroupObjectReader::new(&header).read_object(&mut { objects });
let meta = SubgroupObjectReader::new(&header).read_object_meta(&mut { objects });
let datagram = status_datagram(code);
let decoded = DatagramHeader::decode(&mut &datagram[..]);
match assigned {
Some(status) => {
let object = read.unwrap_or_else(|e| {
panic!(
"read_object refused status {code:#x}, which the draft assigns: {e:?}"
)
});
assert_eq!(object.object_status, Some(status));
assert_eq!(meta.unwrap().status, Some(code));
assert_eq!(decoded.unwrap().object_status, Some(status));
let mut written = Vec::new();
header.encode(&mut written);
SubgroupObjectReader::new(&header)
.write_object(&status_object(Some(status)), &mut written)
.unwrap();
assert_eq!(
written, stream,
"the encoder must produce the frame the decoder accepted for status {code:#x}"
);
let mut written = Vec::new();
status_datagram_header(Some(status)).encode(&mut written);
assert_eq!(
written, datagram,
"the encoder must produce the datagram the decoder accepted for status {code:#x}"
);
}
None => {
for (site, result) in [
("subgroup read_object", read.map(|_| ())),
("subgroup read_object_meta", meta.map(|_| ())),
("status datagram", decoded.map(|_| ())),
] {
match result {
Ok(()) => panic!(
"{site} accepted status {code:#x}, which the draft does not assign"
),
Err(error) => assert!(
matches!(error, CodecError::InvalidField),
"{site} refused status {code:#x} with {error:?}, not InvalidField"
),
}
}
}
}
}
}
}