moq_transport/message/
subscribe_update.rs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
use crate::coding::{Decode, DecodeError, Encode, EncodeError, Params};
use crate::message::subscribe::{SubscribeLocation, SubscribePair};
use crate::message::FilterType;
use crate::message::GroupOrder;

/// Sent by the subscriber to request all future objects for the given track.
///
/// Objects will use the provided ID instead of the full track name, to save bytes.
#[derive(Clone, Debug)]
pub struct SubscribeUpdate {
	/// The subscription ID
	pub id: u64,

	/// Track properties
	pub track_alias: u64, // This alias is useless but part of the spec
	pub track_namespace: String,
	pub track_name: String,

	// Subscriber Priority
	pub subscriber_priority: u8,
	pub group_order: GroupOrder,

	/// Filter type
	pub filter_type: FilterType,

	/// The start/end group/object. (TODO: Make optional)
	pub start: Option<SubscribePair>, // TODO: Make optional
	pub end: Option<SubscribePair>, // TODO: Make optional

	/// Optional parameters
	pub params: Params,
}

impl Decode for SubscribeUpdate {
	fn decode<R: bytes::Buf>(r: &mut R) -> Result<Self, DecodeError> {
		let id = u64::decode(r)?;
		let track_alias = u64::decode(r)?;
		let track_namespace = String::decode(r)?;
		let track_name = String::decode(r)?;

		let subscriber_priority = u8::decode(r)?;
		let group_order = GroupOrder::decode(r)?;

		let filter_type = FilterType::decode(r)?;

		let start: Option<SubscribePair>;
		let end: Option<SubscribePair>;
		match filter_type {
			FilterType::AbsoluteStart => {
				if r.remaining() < 2 {
					return Err(DecodeError::MissingField);
				}
				start = Some(SubscribePair::decode(r)?);
				end = None;
			}
			FilterType::AbsoluteRange => {
				if r.remaining() < 4 {
					return Err(DecodeError::MissingField);
				}
				start = Some(SubscribePair::decode(r)?);
				end = Some(SubscribePair::decode(r)?);
			}
			_ => {
				start = None;
				end = None;
			}
		}

		if let Some(s) = &start {
			// You can't have a start object without a start group.
			if s.group == SubscribeLocation::None && s.object != SubscribeLocation::None {
				return Err(DecodeError::InvalidSubscribeLocation);
			}
		}
		if let Some(e) = &end {
			// You can't have an end object without an end group.
			if e.group == SubscribeLocation::None && e.object != SubscribeLocation::None {
				return Err(DecodeError::InvalidSubscribeLocation);
			}
		}

		// NOTE: There's some more location restrictions in the draft, but they're enforced at a higher level.

		let params = Params::decode(r)?;

		Ok(Self {
			id,
			track_alias,
			track_namespace,
			track_name,
			subscriber_priority,
			group_order,
			filter_type,
			start,
			end,
			params,
		})
	}
}

impl Encode for SubscribeUpdate {
	fn encode<W: bytes::BufMut>(&self, w: &mut W) -> Result<(), EncodeError> {
		self.id.encode(w)?;
		self.track_alias.encode(w)?;
		self.track_namespace.encode(w)?;
		self.track_name.encode(w)?;

		self.subscriber_priority.encode(w)?;
		self.group_order.encode(w)?;

		self.filter_type.encode(w)?;

		if self.filter_type == FilterType::AbsoluteStart || self.filter_type == FilterType::AbsoluteRange {
			if self.start.is_none() || self.end.is_none() {
				return Err(EncodeError::MissingField);
			}
			if let Some(start) = &self.start {
				start.encode(w)?;
			}
			if let Some(end) = &self.end {
				end.encode(w)?;
			}
		}

		self.params.encode(w)?;

		Ok(())
	}
}