use crate::auth_token::{AuthorizationToken, AUTH_TOKEN_PARAMETER};
use crate::error::MAX_FULL_TRACK_NAME_LENGTH;
pub use crate::error::{
CodecError, MAX_GOAWAY_URI_LENGTH, MAX_MESSAGE_LENGTH, MAX_NAMESPACE_TUPLE_SIZE,
MAX_REASON_PHRASE_LENGTH,
};
use crate::kvp::{KeyValuePair, KvpError, KvpValue, MAX_KVP_VALUE_LEN};
use crate::subscription_filter::{SubscriptionFilter, SUBSCRIPTION_FILTER_PARAMETER};
use crate::types::check_location_range;
use crate::types::*;
use crate::varint::{Moqt18 as Wire, VarInt};
use bytes::{Buf, BufMut};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ParamEncoding {
Varint,
Uint8,
Location,
LengthPrefixed,
TrackNamespaceValue,
}
fn param_encoding(key: u64) -> Option<ParamEncoding> {
match key {
0x02 | 0x04 | 0x06 | 0x08 | 0x0A | 0x32 => Some(ParamEncoding::Varint),
0x10 | 0x20 | 0x22 => Some(ParamEncoding::Uint8),
0x09 => Some(ParamEncoding::Location),
0x34 => Some(ParamEncoding::TrackNamespaceValue),
0x03 | 0x21 | 0x25 | 0x26 | 0x27 | 0x28 | 0x29 => Some(ParamEncoding::LengthPrefixed),
_ => None,
}
}
fn uint8_value_in_range(key: u64, value: u8) -> bool {
match key {
0x10 => value <= 1,
0x22 => value == 1 || value == 2,
_ => true,
}
}
const AUTHORIZATION_TOKEN: u64 = 0x03;
fn add_delta(prev_key: u64, delta: u64) -> Result<u64, CodecError> {
prev_key.checked_add(delta).ok_or(CodecError::KeyDeltaOverflow(prev_key, delta))
}
fn check_full_track_name(namespace: &TrackNamespace, track_name: &[u8]) -> Result<(), CodecError> {
let total = namespace.field_bytes_len().saturating_add(track_name.len());
if total > MAX_FULL_TRACK_NAME_LENGTH {
return Err(CodecError::TrackNameTooLong);
}
Ok(())
}
fn check_authorization_tokens(parameters: &[KeyValuePair]) -> Result<(), CodecError> {
for parameter in parameters {
let key = parameter.key.into_inner();
if key != AUTH_TOKEN_PARAMETER {
continue;
}
match ¶meter.value {
KvpValue::Bytes(value) => {
AuthorizationToken::decode_moqt::<Wire>(key, value)?;
}
KvpValue::Varint(_) => {
return Err(CodecError::KeyValueFormatting {
key,
detail: "its value is a bare varint where the type defines a Token structure",
});
}
}
}
Ok(())
}
fn check_subscription_filters(parameters: &[KeyValuePair]) -> Result<(), CodecError> {
for parameter in parameters {
if parameter.key.into_inner() != SUBSCRIPTION_FILTER_PARAMETER {
continue;
}
match ¶meter.value {
KvpValue::Bytes(value) => {
SubscriptionFilter::decode_moqt::<Wire>(value)?.last_group()?;
}
KvpValue::Varint(_) => {
return Err(CodecError::SubscriptionFilterMalformed {
detail: "its value is a bare varint where the type defines a filter",
});
}
}
}
Ok(())
}
fn decode_parameters(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
let count = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
let mut params = crate::types::reserve_bounded(count, buf);
let mut prev_key: u64 = 0;
for i in 0..count {
let delta = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
let abs_key = add_delta(prev_key, delta)?;
if i > 0 && delta == 0 && abs_key != AUTHORIZATION_TOKEN {
return Err(CodecError::DuplicateParameter(abs_key));
}
prev_key = abs_key;
let encoding =
param_encoding(abs_key).ok_or(CodecError::UnknownMessageParameter(abs_key))?;
let value = match encoding {
ParamEncoding::Varint => {
let v = VarInt::decode_moqt::<Wire>(buf)?;
KvpValue::Varint(v)
}
ParamEncoding::Uint8 => {
if buf.remaining() < 1 {
return Err(CodecError::UnexpectedEnd);
}
let byte = buf.get_u8();
if !uint8_value_in_range(abs_key, byte) {
return Err(CodecError::ParameterValueOutOfRange {
key: abs_key,
value: byte as u64,
});
}
KvpValue::Varint(VarInt::from_u64_moqt(byte as u64))
}
ParamEncoding::Location => {
let group = VarInt::decode_moqt::<Wire>(buf)?;
let object = VarInt::decode_moqt::<Wire>(buf)?;
let mut encoded = Vec::new();
group.encode_moqt::<Wire>(&mut encoded);
object.encode_moqt::<Wire>(&mut encoded);
KvpValue::Bytes(encoded)
}
ParamEncoding::LengthPrefixed => {
let len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
let data = read_bytes(buf, len)?;
KvpValue::Bytes(data)
}
ParamEncoding::TrackNamespaceValue => {
let ns = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
let mut encoded = Vec::new();
ns.encode_moqt::<Wire>(&mut encoded);
KvpValue::Bytes(encoded)
}
};
params.push(KeyValuePair { key: VarInt::from_u64_moqt(abs_key), value });
}
check_authorization_tokens(¶ms)?;
check_subscription_filters(¶ms)?;
Ok(params)
}
fn is_location_value(bytes: &[u8]) -> bool {
let mut buf = bytes;
VarInt::decode_moqt::<Wire>(&mut buf).is_ok()
&& VarInt::decode_moqt::<Wire>(&mut buf).is_ok()
&& !buf.has_remaining()
}
fn is_track_namespace_value(bytes: &[u8]) -> bool {
let mut buf = bytes;
TrackNamespace::decode_allow_empty_moqt::<Wire>(&mut buf).is_ok() && !buf.has_remaining()
}
fn encode_parameters(params: &[KeyValuePair], buf: &mut impl BufMut) -> Result<(), CodecError> {
check_authorization_tokens(params)?;
check_subscription_filters(params)?;
VarInt::from_usize(params.len()).encode_moqt::<Wire>(buf);
let mut prev_key: u64 = 0;
for (i, p) in params.iter().enumerate() {
let abs_key = p.key.into_inner();
let delta = abs_key
.checked_sub(prev_key)
.ok_or(CodecError::ParametersOutOfOrder(prev_key, abs_key))?;
if i > 0 && delta == 0 && abs_key != AUTHORIZATION_TOKEN {
return Err(CodecError::DuplicateParameter(abs_key));
}
prev_key = abs_key;
VarInt::from_u64_moqt(delta).encode_moqt::<Wire>(buf);
if let KvpValue::Bytes(b) = &p.value {
if b.len() > MAX_KVP_VALUE_LEN {
return Err(KvpError::ValueTooLong(b.len()).into());
}
}
let encoding = param_encoding(abs_key);
match (&p.value, encoding) {
(KvpValue::Varint(v), Some(ParamEncoding::Varint)) => {
v.encode_moqt::<Wire>(buf);
}
(KvpValue::Varint(v), Some(ParamEncoding::Uint8)) => {
let raw = v.into_inner();
let byte = u8::try_from(raw).map_err(|_| CodecError::InvalidField)?;
if !uint8_value_in_range(abs_key, byte) {
return Err(CodecError::ParameterValueOutOfRange {
key: abs_key,
value: byte as u64,
});
}
buf.put_u8(byte);
}
(KvpValue::Bytes(b), Some(ParamEncoding::Location)) => {
if !is_location_value(b) {
return Err(CodecError::InvalidField);
}
buf.put_slice(b);
}
(KvpValue::Bytes(b), Some(ParamEncoding::TrackNamespaceValue)) => {
if !is_track_namespace_value(b) {
return Err(CodecError::InvalidField);
}
buf.put_slice(b);
}
(KvpValue::Bytes(b), Some(ParamEncoding::LengthPrefixed)) => {
VarInt::from_usize(b.len()).encode_moqt::<Wire>(buf);
buf.put_slice(b);
}
_ => {
match &p.value {
KvpValue::Varint(v) => v.encode_moqt::<Wire>(buf),
KvpValue::Bytes(b) => {
VarInt::from_usize(b.len()).encode_moqt::<Wire>(buf);
buf.put_slice(b);
}
}
}
}
}
Ok(())
}
fn decode_kvp_delta(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
let mut pairs = Vec::new();
let mut prev_key: u64 = 0;
while buf.has_remaining() {
let delta = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
let abs_key = add_delta(prev_key, delta)?;
prev_key = abs_key;
let value = if abs_key.is_multiple_of(2) {
let v = VarInt::decode_moqt::<Wire>(buf)?;
KvpValue::Varint(v)
} else {
let len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
if len > MAX_KVP_VALUE_LEN {
return Err(KvpError::ValueTooLong(len).into());
}
let data = read_bytes(buf, len)?;
KvpValue::Bytes(data)
};
pairs.push(KeyValuePair { key: VarInt::from_u64_moqt(abs_key), value });
}
Ok(pairs)
}
fn encode_kvp_delta(pairs: &[KeyValuePair], buf: &mut impl BufMut) -> Result<(), CodecError> {
let mut prev_key: u64 = 0;
for p in pairs {
let abs_key = p.key.into_inner();
let delta = abs_key
.checked_sub(prev_key)
.ok_or(CodecError::ParametersOutOfOrder(prev_key, abs_key))?;
prev_key = abs_key;
VarInt::from_u64_moqt(delta).encode_moqt::<Wire>(buf);
match &p.value {
KvpValue::Varint(v) => v.encode_moqt::<Wire>(buf),
KvpValue::Bytes(b) => {
if b.len() > MAX_KVP_VALUE_LEN {
return Err(KvpError::ValueTooLong(b.len()).into());
}
VarInt::from_usize(b.len()).encode_moqt::<Wire>(buf);
buf.put_slice(b);
}
}
}
Ok(())
}
const IMMUTABLE_PROPERTIES: u64 = 0x0B;
fn track_property_value_in_range(key: u64, value: u64) -> bool {
match key {
0x22 => value == 1 || value == 2,
0x30 => value <= 1,
_ => true,
}
}
fn check_track_property_values(properties: &[KeyValuePair]) -> Result<(), CodecError> {
for property in properties {
let key = property.key.into_inner();
match &property.value {
KvpValue::Varint(value) => {
let value = value.into_inner();
if !track_property_value_in_range(key, value) {
return Err(CodecError::TrackPropertyValueOutOfRange { key, value });
}
}
KvpValue::Bytes(bytes) if key == IMMUTABLE_PROPERTIES => {
let mut inner = &bytes[..];
match decode_kvp_delta(&mut inner) {
Ok(nested) => check_track_property_values(&nested)?,
Err(_) => return Ok(()),
}
}
KvpValue::Bytes(_) => {}
}
}
Ok(())
}
fn decode_track_properties(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
let properties = decode_kvp_delta(buf)?;
check_track_property_values(&properties)?;
Ok(properties)
}
fn encode_track_properties(
properties: &[KeyValuePair],
buf: &mut impl BufMut,
) -> Result<(), CodecError> {
check_track_property_values(properties)?;
encode_kvp_delta(properties, buf)
}
const KNOWN_SETUP_OPTIONS: &[u64] = &[0x01, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08];
const REPEATABLE_SETUP_OPTION: u64 = 0x03;
fn decode_setup_options(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
let options = decode_kvp_delta(buf)?;
for (i, option) in options.iter().enumerate() {
let key = option.key.into_inner();
if key == REPEATABLE_SETUP_OPTION || !KNOWN_SETUP_OPTIONS.contains(&key) {
continue;
}
if options[..i].iter().any(|earlier| earlier.key == option.key) {
return Err(CodecError::DuplicateParameter(key));
}
}
check_authorization_tokens(&options)?;
Ok(options)
}
fn encode_setup_options(options: &[KeyValuePair], buf: &mut impl BufMut) -> Result<(), CodecError> {
check_authorization_tokens(options)?;
for (i, option) in options.iter().enumerate() {
if option.key.into_inner() == REPEATABLE_SETUP_OPTION {
continue;
}
if options[..i].iter().any(|earlier| earlier.key == option.key) {
return Err(CodecError::DuplicateParameter(option.key.into_inner()));
}
}
encode_kvp_delta(options, buf)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u64)]
pub enum MessageType {
RequestUpdate = 0x02,
Subscribe = 0x03,
SubscribeOk = 0x04,
RequestError = 0x05,
PublishNamespace = 0x06,
RequestOk = 0x07,
Namespace = 0x08,
PublishDone = 0x0B,
TrackStatus = 0x0D,
NamespaceDone = 0x0E,
PublishSkipped = 0x0F,
GoAway = 0x10,
Fetch = 0x16,
FetchOk = 0x18,
Publish = 0x1D,
SubscribeNamespace = 0x50,
SubscribeTracks = 0x51,
Setup = 0x2F00,
}
impl MessageType {
pub fn from_id(id: u64) -> Option<Self> {
match id {
0x02 => Some(MessageType::RequestUpdate),
0x03 => Some(MessageType::Subscribe),
0x04 => Some(MessageType::SubscribeOk),
0x05 => Some(MessageType::RequestError),
0x06 => Some(MessageType::PublishNamespace),
0x07 => Some(MessageType::RequestOk),
0x08 => Some(MessageType::Namespace),
0x0B => Some(MessageType::PublishDone),
0x0D => Some(MessageType::TrackStatus),
0x0E => Some(MessageType::NamespaceDone),
0x0F => Some(MessageType::PublishSkipped),
0x10 => Some(MessageType::GoAway),
0x16 => Some(MessageType::Fetch),
0x18 => Some(MessageType::FetchOk),
0x1D => Some(MessageType::Publish),
0x50 => Some(MessageType::SubscribeNamespace),
0x51 => Some(MessageType::SubscribeTracks),
0x2F00 => Some(MessageType::Setup),
_ => None,
}
}
pub fn id(&self) -> u64 {
*self as u64
}
pub fn name(&self) -> &'static str {
match self {
MessageType::RequestUpdate => "request_update",
MessageType::Subscribe => "subscribe",
MessageType::SubscribeOk => "subscribe_ok",
MessageType::RequestError => "request_error",
MessageType::PublishNamespace => "publish_namespace",
MessageType::RequestOk => "request_ok",
MessageType::Namespace => "namespace",
MessageType::PublishDone => "publish_done",
MessageType::TrackStatus => "track_status",
MessageType::NamespaceDone => "namespace_done",
MessageType::PublishSkipped => "publish_skipped",
MessageType::GoAway => "goaway",
MessageType::Fetch => "fetch",
MessageType::FetchOk => "fetch_ok",
MessageType::Publish => "publish",
MessageType::SubscribeNamespace => "subscribe_namespace",
MessageType::SubscribeTracks => "subscribe_tracks",
MessageType::Setup => "setup",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Setup {
pub options: Vec<KeyValuePair>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GoAway {
pub new_session_uri: Vec<u8>,
pub timeout: VarInt,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RequestOk {
pub parameters: Vec<KeyValuePair>,
pub track_properties: Vec<KeyValuePair>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Redirect {
pub connect_uri: Vec<u8>,
pub track_namespace: TrackNamespace,
pub track_name: Vec<u8>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RequestError {
pub error_code: VarInt,
pub retry_interval: VarInt,
pub reason_phrase: Vec<u8>,
pub redirect: Option<Redirect>,
}
pub mod request_error_codes {
pub const UNSUPPORTED_EXTENSION: u64 = 0x33;
pub const REDIRECT: u64 = 0x34;
pub const CONFLICTING_FILTERS: u64 = 0x35;
pub const INVALID_FILTER: u64 = 0x36;
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Subscribe {
pub request_id: VarInt,
pub track_namespace: TrackNamespace,
pub track_name: Vec<u8>,
pub parameters: Vec<KeyValuePair>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SubscribeOk {
pub track_alias: VarInt,
pub parameters: Vec<KeyValuePair>,
pub track_properties: Vec<KeyValuePair>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RequestUpdate {
pub request_id: VarInt,
pub parameters: Vec<KeyValuePair>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Publish {
pub request_id: VarInt,
pub track_namespace: TrackNamespace,
pub track_name: Vec<u8>,
pub track_alias: VarInt,
pub parameters: Vec<KeyValuePair>,
pub track_properties: Vec<KeyValuePair>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PublishDone {
pub status_code: VarInt,
pub stream_count: VarInt,
pub reason_phrase: Vec<u8>,
}
pub mod publish_done_codes {
pub const TOO_FAR_BEHIND: u64 = 0x05;
pub const EXPIRED: u64 = 0x06;
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PublishNamespace {
pub request_id: VarInt,
pub track_namespace: TrackNamespace,
pub parameters: Vec<KeyValuePair>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Namespace {
pub namespace_suffix: TrackNamespace,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NamespaceDone {
pub namespace_suffix: TrackNamespace,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SubscribeNamespace {
pub request_id: VarInt,
pub namespace_prefix: TrackNamespace,
pub parameters: Vec<KeyValuePair>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SubscribeTracks {
pub request_id: VarInt,
pub namespace_prefix: TrackNamespace,
pub parameters: Vec<KeyValuePair>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TrackStatus {
pub request_id: VarInt,
pub track_namespace: TrackNamespace,
pub track_name: Vec<u8>,
pub parameters: Vec<KeyValuePair>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u64)]
pub enum FetchType {
Standalone = 1,
RelativeJoining = 2,
AbsoluteJoining = 3,
}
impl FetchType {
pub fn from_u64(v: u64) -> Option<Self> {
match v {
1 => Some(FetchType::Standalone),
2 => Some(FetchType::RelativeJoining),
3 => Some(FetchType::AbsoluteJoining),
_ => None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Fetch {
pub request_id: VarInt,
pub fetch_type: FetchType,
pub fetch_payload: FetchPayload,
pub parameters: Vec<KeyValuePair>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FetchPayload {
Standalone {
track_namespace: TrackNamespace,
track_name: Vec<u8>,
start_group: VarInt,
start_object: VarInt,
end_group: VarInt,
end_object: VarInt,
},
Joining {
joining_request_id: VarInt,
joining_start: VarInt,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FetchOk {
pub end_of_track: u8,
pub end_group: VarInt,
pub end_object: VarInt,
pub parameters: Vec<KeyValuePair>,
pub track_properties: Vec<KeyValuePair>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PublishSkipped {
pub namespace_suffix: TrackNamespace,
pub track_name: Vec<u8>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ControlMessage {
Setup(Setup),
GoAway(GoAway),
RequestOk(RequestOk),
RequestError(RequestError),
Subscribe(Subscribe),
SubscribeOk(SubscribeOk),
RequestUpdate(RequestUpdate),
Publish(Publish),
PublishDone(PublishDone),
PublishNamespace(PublishNamespace),
Namespace(Namespace),
NamespaceDone(NamespaceDone),
SubscribeNamespace(SubscribeNamespace),
SubscribeTracks(SubscribeTracks),
TrackStatus(TrackStatus),
Fetch(Fetch),
FetchOk(FetchOk),
PublishSkipped(PublishSkipped),
}
fn check_ranges(message: &ControlMessage) -> Result<(), CodecError> {
match message {
ControlMessage::Fetch(m) => match &m.fetch_payload {
FetchPayload::Standalone {
start_group, start_object, end_group, end_object, ..
} => check_location_range(
start_group.into_inner(),
start_object.into_inner(),
end_group.into_inner(),
end_object.into_inner(),
),
FetchPayload::Joining { .. } => Ok(()),
},
_ => Ok(()),
}
}
fn check_discriminators(message: &ControlMessage) -> Result<(), CodecError> {
match message {
ControlMessage::Fetch(m) => {
let body_is_standalone = matches!(m.fetch_payload, FetchPayload::Standalone { .. });
if body_is_standalone != (m.fetch_type == FetchType::Standalone) {
return Err(CodecError::InvalidField);
}
}
ControlMessage::RequestError(m) => {
let code_is_redirect = m.error_code.into_inner() == request_error_codes::REDIRECT;
if code_is_redirect != m.redirect.is_some() {
return Err(CodecError::InvalidField);
}
}
_ => {}
}
Ok(())
}
fn parameter_in_scope(key: u64, message: MessageType) -> bool {
use MessageType as M;
match key {
0x02 => {
matches!(message, M::RequestOk | M::Subscribe | M::RequestUpdate | M::SubscribeTracks)
}
0x03 => matches!(
message,
M::Publish
| M::Subscribe
| M::RequestUpdate
| M::SubscribeNamespace
| M::SubscribeTracks
| M::PublishNamespace
| M::TrackStatus
| M::Fetch
),
0x04 => matches!(message, M::Subscribe | M::SubscribeTracks),
0x06 => {
matches!(message, M::RequestOk | M::Subscribe | M::RequestUpdate | M::SubscribeTracks)
}
0x08 => matches!(message, M::SubscribeOk | M::Publish | M::RequestOk),
0x09 => matches!(message, M::SubscribeOk | M::Publish | M::RequestOk),
0x0A => matches!(message, M::Fetch),
0x10 => matches!(
message,
M::Subscribe | M::RequestUpdate | M::Publish | M::RequestOk | M::SubscribeTracks
),
0x20 => matches!(
message,
M::Subscribe | M::Fetch | M::RequestUpdate | M::RequestOk | M::SubscribeTracks
),
0x21 => {
matches!(message, M::Subscribe | M::RequestOk | M::RequestUpdate | M::SubscribeTracks)
}
0x22 => matches!(message, M::Subscribe | M::SubscribeTracks | M::Fetch),
0x25..=0x28 => matches!(
message,
M::Fetch | M::Subscribe | M::SubscribeTracks | M::RequestOk | M::RequestUpdate
),
0x29 => matches!(message, M::SubscribeTracks | M::RequestUpdate),
0x32 => {
matches!(message, M::RequestOk | M::Subscribe | M::RequestUpdate | M::SubscribeTracks)
}
0x34 => matches!(message, M::RequestUpdate),
_ => true,
}
}
fn check_parameter_scope(message: &ControlMessage) -> Result<(), CodecError> {
let parameters = match message {
ControlMessage::RequestOk(m) => &m.parameters,
ControlMessage::Subscribe(m) => &m.parameters,
ControlMessage::SubscribeOk(m) => &m.parameters,
ControlMessage::RequestUpdate(m) => &m.parameters,
ControlMessage::Publish(m) => &m.parameters,
ControlMessage::PublishNamespace(m) => &m.parameters,
ControlMessage::SubscribeNamespace(m) => &m.parameters,
ControlMessage::SubscribeTracks(m) => &m.parameters,
ControlMessage::TrackStatus(m) => &m.parameters,
ControlMessage::Fetch(m) => &m.parameters,
ControlMessage::FetchOk(m) => &m.parameters,
ControlMessage::Setup(_)
| ControlMessage::GoAway(_)
| ControlMessage::RequestError(_)
| ControlMessage::PublishDone(_)
| ControlMessage::Namespace(_)
| ControlMessage::NamespaceDone(_)
| ControlMessage::PublishSkipped(_) => return Ok(()),
};
let message_type = message.message_type();
for parameter in parameters {
let key = parameter.key.into_inner();
if !parameter_in_scope(key, message_type) {
return Err(CodecError::ParameterOutOfScope { key, message_type: message_type.id() });
}
}
Ok(())
}
impl ControlMessage {
pub fn encode(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
check_discriminators(self)?;
check_ranges(self)?;
check_parameter_scope(self)?;
let mut body = Vec::with_capacity(256);
self.encode_body(&mut body)?;
if body.len() > MAX_MESSAGE_LENGTH {
return Err(CodecError::MessageTooLong(body.len()));
}
let msg_type = self.message_type();
VarInt::from_usize(msg_type.id() as usize).encode_moqt::<Wire>(buf);
buf.put_u16(body.len() as u16);
buf.put_slice(&body);
Ok(())
}
pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
let type_id = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
let msg_type =
MessageType::from_id(type_id).ok_or(CodecError::UnknownMessageType(type_id))?;
if buf.remaining() < 2 {
return Err(CodecError::UnexpectedEnd);
}
let body_len = buf.get_u16() as usize;
if buf.remaining() < body_len {
return Err(CodecError::UnexpectedEnd);
}
let body_bytes = buf.copy_to_bytes(body_len);
let mut body = &body_bytes[..];
let msg = match Self::decode_body(msg_type, &mut body) {
Ok(msg) => msg,
Err(
CodecError::UnexpectedEnd
| CodecError::Kvp(crate::kvp::KvpError::UnexpectedEnd)
| CodecError::Kvp(crate::kvp::KvpError::VarInt(
crate::varint::VarIntError::UnexpectedEnd,
))
| CodecError::VarInt(crate::varint::VarIntError::UnexpectedEnd),
) => {
return Err(CodecError::ControlMessageLengthMismatch {
declared: body_len,
detail: "its fields ran past the end",
});
}
Err(e) => return Err(e),
};
check_ranges(&msg)?;
check_parameter_scope(&msg)?;
if body.has_remaining() {
return Err(CodecError::ControlMessageLengthMismatch {
declared: body_len,
detail: "its fields left bytes unread",
});
}
Ok(msg)
}
fn encode_body(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
match self {
ControlMessage::Setup(m) => {
encode_setup_options(&m.options, buf)?;
}
ControlMessage::GoAway(m) => {
if m.new_session_uri.len() > MAX_GOAWAY_URI_LENGTH {
return Err(CodecError::GoAwayUriTooLong);
}
VarInt::from_usize(m.new_session_uri.len()).encode_moqt::<Wire>(buf);
buf.put_slice(&m.new_session_uri);
m.timeout.encode_moqt::<Wire>(buf);
}
ControlMessage::RequestOk(m) => {
encode_parameters(&m.parameters, buf)?;
encode_track_properties(&m.track_properties, buf)?;
}
ControlMessage::RequestError(m) => {
if m.reason_phrase.len() > MAX_REASON_PHRASE_LENGTH {
return Err(CodecError::ReasonPhraseTooLong);
}
m.error_code.encode_moqt::<Wire>(buf);
m.retry_interval.encode_moqt::<Wire>(buf);
VarInt::from_usize(m.reason_phrase.len()).encode_moqt::<Wire>(buf);
buf.put_slice(&m.reason_phrase);
if let Some(r) = &m.redirect {
r.track_namespace.validate_moqt()?;
check_full_track_name(&r.track_namespace, &r.track_name)?;
VarInt::from_usize(r.connect_uri.len()).encode_moqt::<Wire>(buf);
buf.put_slice(&r.connect_uri);
r.track_namespace.encode_moqt::<Wire>(buf);
VarInt::from_usize(r.track_name.len()).encode_moqt::<Wire>(buf);
buf.put_slice(&r.track_name);
}
}
ControlMessage::Subscribe(m) => {
m.track_namespace.validate_moqt()?;
check_full_track_name(&m.track_namespace, &m.track_name)?;
m.request_id.encode_moqt::<Wire>(buf);
m.track_namespace.encode_moqt::<Wire>(buf);
VarInt::from_usize(m.track_name.len()).encode_moqt::<Wire>(buf);
buf.put_slice(&m.track_name);
encode_parameters(&m.parameters, buf)?;
}
ControlMessage::SubscribeOk(m) => {
m.track_alias.encode_moqt::<Wire>(buf);
encode_parameters(&m.parameters, buf)?;
encode_track_properties(&m.track_properties, buf)?;
}
ControlMessage::RequestUpdate(m) => {
m.request_id.encode_moqt::<Wire>(buf);
encode_parameters(&m.parameters, buf)?;
}
ControlMessage::Publish(m) => {
m.track_namespace.validate_moqt()?;
check_full_track_name(&m.track_namespace, &m.track_name)?;
m.request_id.encode_moqt::<Wire>(buf);
m.track_namespace.encode_moqt::<Wire>(buf);
VarInt::from_usize(m.track_name.len()).encode_moqt::<Wire>(buf);
buf.put_slice(&m.track_name);
m.track_alias.encode_moqt::<Wire>(buf);
encode_parameters(&m.parameters, buf)?;
encode_track_properties(&m.track_properties, buf)?;
}
ControlMessage::PublishDone(m) => {
if m.reason_phrase.len() > MAX_REASON_PHRASE_LENGTH {
return Err(CodecError::ReasonPhraseTooLong);
}
m.status_code.encode_moqt::<Wire>(buf);
m.stream_count.encode_moqt::<Wire>(buf);
VarInt::from_usize(m.reason_phrase.len()).encode_moqt::<Wire>(buf);
buf.put_slice(&m.reason_phrase);
}
ControlMessage::PublishNamespace(m) => {
m.track_namespace.validate_moqt()?;
m.request_id.encode_moqt::<Wire>(buf);
m.track_namespace.encode_moqt::<Wire>(buf);
encode_parameters(&m.parameters, buf)?;
}
ControlMessage::Namespace(m) => {
m.namespace_suffix.validate_moqt()?;
m.namespace_suffix.encode_moqt::<Wire>(buf);
}
ControlMessage::NamespaceDone(m) => {
m.namespace_suffix.validate_moqt()?;
m.namespace_suffix.encode_moqt::<Wire>(buf);
}
ControlMessage::SubscribeNamespace(m) => {
m.namespace_prefix.validate_moqt()?;
m.request_id.encode_moqt::<Wire>(buf);
m.namespace_prefix.encode_moqt::<Wire>(buf);
encode_parameters(&m.parameters, buf)?;
}
ControlMessage::SubscribeTracks(m) => {
m.namespace_prefix.validate_moqt()?;
m.request_id.encode_moqt::<Wire>(buf);
m.namespace_prefix.encode_moqt::<Wire>(buf);
encode_parameters(&m.parameters, buf)?;
}
ControlMessage::TrackStatus(m) => {
m.track_namespace.validate_moqt()?;
check_full_track_name(&m.track_namespace, &m.track_name)?;
m.request_id.encode_moqt::<Wire>(buf);
m.track_namespace.encode_moqt::<Wire>(buf);
VarInt::from_usize(m.track_name.len()).encode_moqt::<Wire>(buf);
buf.put_slice(&m.track_name);
encode_parameters(&m.parameters, buf)?;
}
ControlMessage::Fetch(m) => {
m.request_id.encode_moqt::<Wire>(buf);
VarInt::from_usize(m.fetch_type as usize).encode_moqt::<Wire>(buf);
match &m.fetch_payload {
FetchPayload::Standalone {
track_namespace,
track_name,
start_group,
start_object,
end_group,
end_object,
} => {
track_namespace.validate_moqt()?;
check_full_track_name(track_namespace, track_name)?;
track_namespace.encode_moqt::<Wire>(buf);
VarInt::from_usize(track_name.len()).encode_moqt::<Wire>(buf);
buf.put_slice(track_name);
start_group.encode_moqt::<Wire>(buf);
start_object.encode_moqt::<Wire>(buf);
end_group.encode_moqt::<Wire>(buf);
end_object.encode_moqt::<Wire>(buf);
}
FetchPayload::Joining { joining_request_id, joining_start } => {
joining_request_id.encode_moqt::<Wire>(buf);
joining_start.encode_moqt::<Wire>(buf);
}
}
encode_parameters(&m.parameters, buf)?;
}
ControlMessage::FetchOk(m) => {
buf.put_u8(m.end_of_track);
m.end_group.encode_moqt::<Wire>(buf);
m.end_object.encode_moqt::<Wire>(buf);
encode_parameters(&m.parameters, buf)?;
encode_track_properties(&m.track_properties, buf)?;
}
ControlMessage::PublishSkipped(m) => {
m.namespace_suffix.validate_moqt()?;
check_full_track_name(&m.namespace_suffix, &m.track_name)?;
m.namespace_suffix.encode_moqt::<Wire>(buf);
VarInt::from_usize(m.track_name.len()).encode_moqt::<Wire>(buf);
buf.put_slice(&m.track_name);
}
}
Ok(())
}
fn decode_body(msg_type: MessageType, buf: &mut impl Buf) -> Result<Self, CodecError> {
match msg_type {
MessageType::Setup => {
let options = decode_setup_options(buf)?;
Ok(ControlMessage::Setup(Setup { options }))
}
MessageType::GoAway => {
let uri_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
if uri_len > MAX_GOAWAY_URI_LENGTH {
return Err(CodecError::GoAwayUriTooLong);
}
let uri = read_bytes(buf, uri_len)?;
let timeout = VarInt::decode_moqt::<Wire>(buf)?;
Ok(ControlMessage::GoAway(GoAway { new_session_uri: uri, timeout }))
}
MessageType::RequestOk => {
let parameters = decode_parameters(buf)?;
let track_properties = decode_track_properties(buf)?;
Ok(ControlMessage::RequestOk(RequestOk { parameters, track_properties }))
}
MessageType::RequestError => {
let error_code = VarInt::decode_moqt::<Wire>(buf)?;
let retry_interval = VarInt::decode_moqt::<Wire>(buf)?;
let reason_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
if reason_len > MAX_REASON_PHRASE_LENGTH {
return Err(CodecError::ReasonPhraseTooLong);
}
let reason_phrase = read_bytes(buf, reason_len)?;
let redirect = if error_code.into_inner() == request_error_codes::REDIRECT {
let uri_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
let connect_uri = read_bytes(buf, uri_len)?;
let track_namespace = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
let name_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
let track_name = read_bytes(buf, name_len)?;
check_full_track_name(&track_namespace, &track_name)?;
Some(Redirect { connect_uri, track_namespace, track_name })
} else {
None
};
Ok(ControlMessage::RequestError(RequestError {
error_code,
retry_interval,
reason_phrase,
redirect,
}))
}
MessageType::Subscribe => {
let request_id = VarInt::decode_moqt::<Wire>(buf)?;
let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
let track_name = read_bytes(buf, tn_len)?;
check_full_track_name(&track_namespace, &track_name)?;
let parameters = decode_parameters(buf)?;
Ok(ControlMessage::Subscribe(Subscribe {
request_id,
track_namespace,
track_name,
parameters,
}))
}
MessageType::SubscribeOk => {
let track_alias = VarInt::decode_moqt::<Wire>(buf)?;
let parameters = decode_parameters(buf)?;
let track_properties = decode_track_properties(buf)?;
Ok(ControlMessage::SubscribeOk(SubscribeOk {
track_alias,
parameters,
track_properties,
}))
}
MessageType::RequestUpdate => {
let request_id = VarInt::decode_moqt::<Wire>(buf)?;
let parameters = decode_parameters(buf)?;
Ok(ControlMessage::RequestUpdate(RequestUpdate { request_id, parameters }))
}
MessageType::Publish => {
let request_id = VarInt::decode_moqt::<Wire>(buf)?;
let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
let track_name = read_bytes(buf, tn_len)?;
let track_alias = VarInt::decode_moqt::<Wire>(buf)?;
check_full_track_name(&track_namespace, &track_name)?;
let parameters = decode_parameters(buf)?;
let track_properties = decode_track_properties(buf)?;
Ok(ControlMessage::Publish(Publish {
request_id,
track_namespace,
track_name,
track_alias,
parameters,
track_properties,
}))
}
MessageType::PublishDone => {
let status_code = VarInt::decode_moqt::<Wire>(buf)?;
let stream_count = VarInt::decode_moqt::<Wire>(buf)?;
let reason_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
if reason_len > MAX_REASON_PHRASE_LENGTH {
return Err(CodecError::ReasonPhraseTooLong);
}
let reason_phrase = read_bytes(buf, reason_len)?;
Ok(ControlMessage::PublishDone(PublishDone {
status_code,
stream_count,
reason_phrase,
}))
}
MessageType::PublishNamespace => {
let request_id = VarInt::decode_moqt::<Wire>(buf)?;
let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
let parameters = decode_parameters(buf)?;
Ok(ControlMessage::PublishNamespace(PublishNamespace {
request_id,
track_namespace,
parameters,
}))
}
MessageType::Namespace => {
let namespace_suffix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
Ok(ControlMessage::Namespace(Namespace { namespace_suffix }))
}
MessageType::NamespaceDone => {
let namespace_suffix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
Ok(ControlMessage::NamespaceDone(NamespaceDone { namespace_suffix }))
}
MessageType::SubscribeNamespace => {
let request_id = VarInt::decode_moqt::<Wire>(buf)?;
let namespace_prefix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
let parameters = decode_parameters(buf)?;
Ok(ControlMessage::SubscribeNamespace(SubscribeNamespace {
request_id,
namespace_prefix,
parameters,
}))
}
MessageType::SubscribeTracks => {
let request_id = VarInt::decode_moqt::<Wire>(buf)?;
let namespace_prefix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
let parameters = decode_parameters(buf)?;
Ok(ControlMessage::SubscribeTracks(SubscribeTracks {
request_id,
namespace_prefix,
parameters,
}))
}
MessageType::TrackStatus => {
let request_id = VarInt::decode_moqt::<Wire>(buf)?;
let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
let track_name = read_bytes(buf, tn_len)?;
check_full_track_name(&track_namespace, &track_name)?;
let parameters = decode_parameters(buf)?;
Ok(ControlMessage::TrackStatus(TrackStatus {
request_id,
track_namespace,
track_name,
parameters,
}))
}
MessageType::Fetch => {
let request_id = VarInt::decode_moqt::<Wire>(buf)?;
let fetch_type_val = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
let fetch_type = FetchType::from_u64(fetch_type_val)
.ok_or(CodecError::InvalidFetchType(fetch_type_val))?;
let fetch_payload = match fetch_type {
FetchType::Standalone => {
let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
let track_name = read_bytes(buf, tn_len)?;
let start_group = VarInt::decode_moqt::<Wire>(buf)?;
let start_object = VarInt::decode_moqt::<Wire>(buf)?;
let end_group = VarInt::decode_moqt::<Wire>(buf)?;
let end_object = VarInt::decode_moqt::<Wire>(buf)?;
check_full_track_name(&track_namespace, &track_name)?;
FetchPayload::Standalone {
track_namespace,
track_name,
start_group,
start_object,
end_group,
end_object,
}
}
FetchType::RelativeJoining | FetchType::AbsoluteJoining => {
let joining_request_id = VarInt::decode_moqt::<Wire>(buf)?;
let joining_start = VarInt::decode_moqt::<Wire>(buf)?;
FetchPayload::Joining { joining_request_id, joining_start }
}
};
let parameters = decode_parameters(buf)?;
Ok(ControlMessage::Fetch(Fetch {
request_id,
fetch_type,
fetch_payload,
parameters,
}))
}
MessageType::FetchOk => {
if buf.remaining() < 1 {
return Err(CodecError::UnexpectedEnd);
}
let end_of_track = buf.get_u8();
let end_group = VarInt::decode_moqt::<Wire>(buf)?;
let end_object = VarInt::decode_moqt::<Wire>(buf)?;
let parameters = decode_parameters(buf)?;
let track_properties = decode_track_properties(buf)?;
Ok(ControlMessage::FetchOk(FetchOk {
end_of_track,
end_group,
end_object,
parameters,
track_properties,
}))
}
MessageType::PublishSkipped => {
let namespace_suffix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
let track_name = read_bytes(buf, tn_len)?;
check_full_track_name(&namespace_suffix, &track_name)?;
Ok(ControlMessage::PublishSkipped(PublishSkipped { namespace_suffix, track_name }))
}
}
}
pub fn message_type(&self) -> MessageType {
match self {
ControlMessage::Setup(_) => MessageType::Setup,
ControlMessage::GoAway(_) => MessageType::GoAway,
ControlMessage::RequestOk(_) => MessageType::RequestOk,
ControlMessage::RequestError(_) => MessageType::RequestError,
ControlMessage::Subscribe(_) => MessageType::Subscribe,
ControlMessage::SubscribeOk(_) => MessageType::SubscribeOk,
ControlMessage::RequestUpdate(_) => MessageType::RequestUpdate,
ControlMessage::Publish(_) => MessageType::Publish,
ControlMessage::PublishDone(_) => MessageType::PublishDone,
ControlMessage::PublishNamespace(_) => MessageType::PublishNamespace,
ControlMessage::Namespace(_) => MessageType::Namespace,
ControlMessage::NamespaceDone(_) => MessageType::NamespaceDone,
ControlMessage::SubscribeNamespace(_) => MessageType::SubscribeNamespace,
ControlMessage::SubscribeTracks(_) => MessageType::SubscribeTracks,
ControlMessage::TrackStatus(_) => MessageType::TrackStatus,
ControlMessage::Fetch(_) => MessageType::Fetch,
ControlMessage::FetchOk(_) => MessageType::FetchOk,
ControlMessage::PublishSkipped(_) => MessageType::PublishSkipped,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn frame_with_declared_len(type_id: u64, declared_len: u16, body: &[u8]) -> Vec<u8> {
let mut out = Vec::new();
VarInt::from_u64_moqt(type_id).encode_moqt::<Wire>(&mut out);
out.put_u16(declared_len);
out.put_slice(body);
out
}
fn frame(type_id: u64, body: &[u8]) -> Vec<u8> {
frame_with_declared_len(type_id, body.len() as u16, body)
}
fn subscribe_body(params: &[u8]) -> Vec<u8> {
let mut body = vec![0x01, 0x01, 0x01, b'a', 0x01, b'b'];
body.extend_from_slice(params);
body
}
#[test]
fn a_message_body_shorter_than_the_declared_length_is_refused() {
let body = subscribe_body(&[0x00]);
let mut junked = body.clone();
junked.extend_from_slice(&[0xff, 0xff]);
let bytes = frame_with_declared_len(0x03, (body.len() + 2) as u16, &junked);
let mut buf = &bytes[..];
assert_eq!(
ControlMessage::decode(&mut buf),
Err(CodecError::ControlMessageLengthMismatch {
declared: (body.len() + 2),
detail: "its fields left bytes unread",
})
);
let honest = frame(0x03, &body);
let mut buf = &honest[..];
assert!(ControlMessage::decode(&mut buf).is_ok());
}
#[test]
fn an_over_long_reason_phrase_is_refused_on_decode() {
for (type_id, prefix) in [(0x05u64, vec![0x01, 0x00]), (0x0B, vec![0x01, 0x00])] {
let mut body = prefix;
let over = MAX_REASON_PHRASE_LENGTH + 976;
VarInt::from_usize(over).encode_moqt::<Wire>(&mut body);
body.extend(std::iter::repeat_n(b'x', over));
let bytes = frame(type_id, &body);
let mut buf = &bytes[..];
assert_eq!(
ControlMessage::decode(&mut buf),
Err(CodecError::ReasonPhraseTooLong),
"message type 0x{type_id:x}"
);
}
}
#[test]
fn an_over_long_goaway_uri_is_refused_on_decode() {
let over = MAX_GOAWAY_URI_LENGTH + 808;
let mut body = Vec::new();
VarInt::from_usize(over).encode_moqt::<Wire>(&mut body);
body.extend(std::iter::repeat_n(b'u', over));
body.push(0x00); let bytes = frame(0x10, &body);
let mut buf = &bytes[..];
assert_eq!(ControlMessage::decode(&mut buf), Err(CodecError::GoAwayUriTooLong));
}
#[test]
fn a_uint8_parameter_outside_its_range_is_refused() {
let cases = [(0x22u8, 7u8, 2u8), (0x10, 9, 1)];
for (key, bad, good) in cases {
let bytes = frame(0x03, &subscribe_body(&[0x01, key, bad]));
let mut buf = &bytes[..];
assert_eq!(
ControlMessage::decode(&mut buf),
Err(CodecError::ParameterValueOutOfRange { key: key as u64, value: bad as u64 }),
"parameter 0x{key:x} value {bad}"
);
let bytes = frame(0x03, &subscribe_body(&[0x01, key, good]));
let mut buf = &bytes[..];
assert!(
ControlMessage::decode(&mut buf).is_ok(),
"parameter 0x{key:x} value {good} should still decode"
);
}
}
#[test]
fn subscriber_priority_still_accepts_the_whole_byte_range() {
for value in [0u8, 1, 2, 128, 255] {
let bytes = frame(0x03, &subscribe_body(&[0x01, 0x20, value]));
let mut buf = &bytes[..];
assert!(ControlMessage::decode(&mut buf).is_ok(), "priority {value}");
}
}
fn param(key: u64, value: &[u8]) -> KeyValuePair {
KeyValuePair { key: VarInt::from_u64_moqt(key), value: KvpValue::Bytes(value.to_vec()) }
}
#[test]
fn largest_object_is_two_bare_varints() {
let body = [0x01, 0x09, 0x0a, 0x03];
let bytes = frame(0x07, &body);
let msg = ControlMessage::decode(&mut &bytes[..]).expect("spec-correct frame must decode");
let ControlMessage::RequestOk(ok) = &msg else {
panic!("expected REQUEST_OK, got {msg:?}")
};
assert_eq!(ok.parameters, vec![param(0x09, &[0x0a, 0x03])]);
assert!(ok.track_properties.is_empty(), "the four body bytes are all parameter");
let mut out = Vec::new();
msg.encode(&mut out).expect("re-encode");
assert_eq!(out, bytes, "the value must go back out as the two bare varints it came in as");
}
#[test]
fn a_location_value_that_is_not_two_varints_is_refused_on_encode() {
for (label, value) in
[("one varint", vec![0x0a]), ("three varints", vec![0x0a, 0x03, 0x05])]
{
let msg = ControlMessage::RequestOk(RequestOk {
parameters: vec![param(0x09, &value)],
track_properties: Vec::new(),
});
let mut out = Vec::new();
assert!(
msg.encode(&mut out).is_err(),
"LARGEST_OBJECT of {label} must not encode: the decoder cannot read it back"
);
}
let msg = ControlMessage::RequestOk(RequestOk {
parameters: vec![param(0x09, &[0x0a, 0x03])],
track_properties: Vec::new(),
});
let mut out = Vec::new();
msg.encode(&mut out).expect("a Location of exactly two varints must still encode");
ControlMessage::decode(&mut &out[..]).expect("and must decode back");
}
#[test]
fn a_location_does_not_eat_the_block_that_follows_it() {
let body = [0x05, 0x01, 0x09, 0x0a, 0x03, 0x02, 0x93, 0x88];
let bytes = frame(0x04, &body);
let msg = ControlMessage::decode(&mut &bytes[..]).expect("spec-correct frame must decode");
let ControlMessage::SubscribeOk(ok) = &msg else {
panic!("expected SUBSCRIBE_OK, got {msg:?}")
};
assert_eq!(ok.parameters, vec![param(0x09, &[0x0a, 0x03])]);
assert_eq!(
ok.track_properties,
vec![KeyValuePair {
key: VarInt::from_u64_moqt(0x02),
value: KvpValue::Varint(VarInt::from_u64_moqt(5000)),
}]
);
let mut out = Vec::new();
msg.encode(&mut out).expect("re-encode");
assert_eq!(out, bytes);
}
#[test]
fn track_namespace_prefix_is_a_bare_track_namespace() {
let namespace: Vec<u8> = [&[0x02, 0x04][..], b"live", &[0x06][..], b"sports"].concat();
assert_eq!(namespace.len(), 13);
let body: Vec<u8> = [&[0x07, 0x01, 0x34][..], &namespace].concat();
assert_eq!(body.len(), 16);
let bytes = frame(0x02, &body);
let msg = ControlMessage::decode(&mut &bytes[..]).expect("spec-correct frame must decode");
let ControlMessage::RequestUpdate(update) = &msg else {
panic!("expected REQUEST_UPDATE, got {msg:?}")
};
assert_eq!(update.request_id.into_inner(), 7);
assert_eq!(update.parameters, vec![param(0x34, &namespace)]);
let mut out = Vec::new();
msg.encode(&mut out).expect("re-encode");
assert_eq!(out, bytes, "no outer length may appear in front of the Track Namespace");
}
#[test]
fn an_empty_track_namespace_prefix_is_one_zero_byte() {
let body = [0x07, 0x01, 0x34, 0x00];
let bytes = frame(0x02, &body);
let msg = ControlMessage::decode(&mut &bytes[..]).expect("empty prefix must decode");
let ControlMessage::RequestUpdate(update) = &msg else {
panic!("expected REQUEST_UPDATE, got {msg:?}")
};
assert_eq!(update.parameters, vec![param(0x34, &[0x00])]);
let mut out = Vec::new();
msg.encode(&mut out).expect("re-encode");
assert_eq!(out, bytes);
}
}