moq-net 0.2.16

/*
9.13. PUBLISH
The publisher sends the PUBLISH control message to initiate a subscription to a track. The receiver verifies the publisher is authorized to publish this track.

PUBLISH Message {
  Type (i) = 0x1D,
  Length (i),
  Request ID (i),
  Track Namespace (tuple),
  Track Name Length (i),
  Track Name (..),
  Track Alias (i),
  Group Order (8),
  Content Exists (8),
  [Largest Location (Location),]
  Forward (8),
  Number of Parameters (i),
  Parameters (..) ...,
}
Figure 15: MOQT PUBLISH Message
Request ID: See Section 9.1.

Track Namespace: Identifies a track's namespace as defined in (Section 2.4.1)

Track Name: Identifies the track name as defined in (Section 2.4.1).

Track Alias: The identifer used for this track in Subgroups or Datagrams (see Section 10.1). The same Track Alias MUST NOT be used to refer to two different Tracks simultaneously. If a subscriber receives a PUBLISH that uses the same Track Alias as a different track with an active subscription, it MUST close the session with error DUPLICATE_TRACK_ALIAS.

Group Order: Indicates the subscription will be delivered in Ascending (0x1) or Descending (0x2) order by group. See Section 7. Values of 0x0 and those larger than 0x2 are a protocol error.

Content Exists: 1 if an object has been published on this track, 0 if not. If 0, then the Largest Group ID and Largest Object ID fields will not be present. Any other value is a protocol error and MUST terminate the session with a PROTOCOL_VIOLATION (Section 3.4).

Largest Location: The location of the largest object available for this track.

Forward: The forward mode for this subscription. Any value other than 0 or 1 is a PROTOCOL_VIOLATION. 0 indicates the publisher will not transmit any objects until the subscriber sets the Forward State to 1. 1 indicates the publisher will start transmitting objects immediately, even before PUBLISH_OK.

Parameters: The parameters are defined in Section 9.2.1.

A subscriber receiving a PUBLISH for a Track it does not wish to receive SHOULD send PUBLISH_ERROR with error code UNINTERESTED, and abandon reading any publisher initiated streams associated with that subscription using a STOP_SENDING frame.

9.14. PUBLISH_OK
The subscriber sends a PUBLISH_OK control message to acknowledge the successful authorization and acceptance of a PUBLISH message, and establish a subscription.

PUBLISH_OK Message {
  Type (i) = 0x1E,
  Length (i),
  Request ID (i),
  Forward (8),
  Subscriber Priority (8),
  Group Order (8),
  Filter Type (i),
  [Start Location (Location)],
  [End Group (i)],
  Number of Parameters (i),
  Parameters (..) ...,
}
Figure 16: MOQT PUBLISH_OK Message
Request ID: The Request ID of the PUBLISH this message is replying to Section 9.13.

Forward: The Forward State for this subscription, either 0 (don't forward) or 1 (forward).

Subscriber Priority: The Subscriber Priority for this subscription.

Group Order: Indicates the subscription will be delivered in Ascending (0x1) or Descending (0x2) order by group. See Section 7. Values of 0x0 and those larger than 0x2 are a protocol error. This overwrites the GroupOrder specified PUBLISH.

Filter Type, Start Location, End Group: See Section 9.7.

Parameters: Parameters associated with this message.

9.15. PUBLISH_ERROR
The subscriber sends a PUBLISH_ERROR control message to reject a subscription initiated by PUBLISH.

PUBLISH_ERROR Message {
  Type (i) = 0x1F,
  Length (i),
  Request ID (i),
  Error Code (i),
  Error Reason (Reason Phrase),
}
Figure 17: MOQT PUBLISH_ERROR Message
Request ID: The Request ID of the PUBLISH this message is replying to Section 9.13.

Error Code: Identifies an integer error code for failure.

Error Reason: Provides the reason for subscription error. See Section 1.4.3.

The application SHOULD use a relevant error code in PUBLISH_ERROR, as defined below:

INTERNAL_ERROR (0x0):
An implementation specific or generic error occurred.

UNAUTHORIZED (0x1):
The publisher is not authorized to publish the given namespace or track.

TIMEOUT (0x2):
The subscription could not be established before an implementation specific timeout.

NOT_SUPPORTED (0x3):
The endpoint does not support the PUBLISH method.

UNINTERESTED (0x4):
The namespace or track is not of interest to the endpoint.


*/

use std::borrow::Cow;

use crate::{
	Path,
	coding::{Decode, DecodeError, Encode, EncodeError},
	ietf::{
		Filter, GroupOrder, Location, Parameters, Properties, RequestId,
		namespace::{decode_namespace, encode_namespace},
	},
};

use super::Message;

use super::Version;

/// PUBLISH_DONE statuses emitted by this implementation.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum PublishDoneStatus {
	/// An implementation-specific failure ended the subscription.
	InternalError,
	/// The track is no longer being published.
	TrackEnded,
}

impl PublishDoneStatus {
	/// Return the registered wire code for every supported draft.
	pub(crate) const fn code(self, version: Version) -> u64 {
		match version {
			Version::Draft14
			| Version::Draft15
			| Version::Draft16
			| Version::Draft17
			| Version::Draft18
			| Version::Draft19
			// Draft-20 removed SUBSCRIPTION_ENDED (0x3), which this implementation never
			// emitted, and left the rest of the registry alone.
			| Version::Draft20 => match self {
				Self::InternalError => 0x0,
				Self::TrackEnded => 0x2,
			},
		}
	}
}

/// Used to be called SubscribeDone
#[derive(Clone, Debug)]
pub struct PublishDone<'a> {
	pub request_id: Option<RequestId>,
	pub status_code: u64,
	pub stream_count: u64,
	pub reason_phrase: Cow<'a, str>,
}

impl Message for PublishDone<'_> {
	const ID: u64 = 0x0b;

	fn encode_msg<W: bytes::BufMut>(&self, w: &mut W, version: Version) -> Result<(), EncodeError> {
		if matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16) {
			self.request_id
				.expect("request_id required for draft14-16")
				.encode(w, version)?;
		} else {
			assert!(self.request_id.is_none(), "request_id must be None for draft17+");
		}
		self.status_code.encode(w, version)?;
		self.stream_count.encode(w, version)?;
		self.reason_phrase.encode(w, version)?;
		Ok(())
	}

	fn decode_msg<R: bytes::Buf>(r: &mut R, version: Version) -> Result<Self, DecodeError> {
		let request_id = if matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16) {
			Some(RequestId::decode(r, version)?)
		} else {
			None
		};
		let status_code = u64::decode(r, version)?;
		let stream_count = u64::decode(r, version)?;
		let reason_phrase = Cow::<str>::decode(r, version)?;

		Ok(Self {
			request_id,
			status_code,
			stream_count,
			reason_phrase,
		})
	}
}

#[derive(Debug)]
pub struct Publish<'a> {
	pub request_id: RequestId,
	pub track_namespace: Path<'a>,
	pub track_name: Cow<'a, str>,
	pub track_alias: u64,
	pub largest_location: Option<Location>,
	pub forward: bool,

	/// Metadata about the track, sent as Track Properties (draft-17+).
	pub properties: Properties,
	// pub parameters: Parameters,
}

impl Message for Publish<'_> {
	const ID: u64 = 0x1D;

	fn encode_msg<W: bytes::BufMut>(&self, w: &mut W, version: Version) -> Result<(), EncodeError> {
		self.request_id.encode(w, version)?;
		if version == Version::Draft17 {
			0u64.encode(w, version)?; // required_request_id_delta = 0
		}
		encode_namespace(w, &self.track_namespace, version)?;
		self.track_name.encode(w, version)?;
		self.track_alias.encode(w, version)?;

		match version {
			Version::Draft14 => {
				self.properties
					.group_order
					.unwrap_or(GroupOrder::Ascending)
					.encode(w, version)?;
				if let Some(location) = &self.largest_location {
					true.encode(w, version)?;
					location.encode(w, version)?;
				} else {
					false.encode(w, version)?;
				}

				self.forward.encode(w, version)?;
				// parameters
				0u8.encode(w, version)?;
			}
			_ => {
				// GROUP_ORDER is a legal PUBLISH parameter only through draft-15; a later peer
				// closes the session with PROTOCOL_VIOLATION when it sees one. The publisher's
				// preference is a DEFAULT_PUBLISHER_GROUP_ORDER track property instead, which we
				// write from draft-17 on. Draft-16 gets neither form; see Properties::encode.
				let group_order = match version {
					Version::Draft15 => self.properties.group_order,
					_ => None,
				};

				encode_params!(w, version,
					0x09 => self.largest_location,
					0x10 => self.forward,
					0x22 => group_order,
				);

				// Track Properties are the final field, so nothing may follow.
				self.properties.encode(w, version)?;
			}
		}

		Ok(())
	}

	fn decode_msg<R: bytes::Buf>(r: &mut R, version: Version) -> Result<Self, DecodeError> {
		let request_id = RequestId::decode(r, version)?;
		if version == Version::Draft17 {
			let _required_request_id_delta = u64::decode(r, version)?;
		}
		let track_namespace = decode_namespace(r, version)?;
		let track_name = Cow::<str>::decode(r, version)?;
		let track_alias = u64::decode(r, version)?;

		match version {
			Version::Draft14 => {
				let group_order = GroupOrder::decode(r, version)?.any_to_descending();
				let content_exists = bool::decode(r, version)?;
				let largest_location = match content_exists {
					true => Some(Location::decode(r, version)?),
					false => None,
				};
				let forward = bool::decode(r, version)?;
				// parameters
				let _params = Parameters::decode(r, version)?;

				Ok(Self {
					request_id,
					track_namespace,
					track_name,
					track_alias,
					largest_location,
					forward,
					properties: Properties {
						group_order: Some(group_order),
						..Default::default()
					},
				})
			}
			_ => {
				// GROUP_ORDER is only legal here through draft-15, but keep accepting it so a
				// peer that still sends it doesn't have its session torn down over a hint.
				//
				// Draft-20 moved the subscription parameters into PUBLISH, so they have to
				// parse here even though reverse publishing is unsupported: an unlisted
				// parameter fails the whole message, which would kill the session instead of
				// letting the request reach its NOT_SUPPORTED response.
				decode_params!(r, version,
					0x02 => object_delivery_timeout: Option<u64>,
					0x06 => subgroup_delivery_timeout: Option<u64>,
					0x08 => _expires: Option<u64>,
					0x09 => largest_location: Option<Location>,
					0x10 => forward: Option<bool>,
					0x20 => subscriber_priority: Option<u8>,
					0x21 => filter: Option<Filter>,
					0x22 => group_order: Option<GroupOrder>,
				);

				// The values are dropped: we refuse the PUBLISH itself, so the subscription
				// settings it proposes never take effect. They still have to be consumed.
				let subscription_params = [
					object_delivery_timeout.is_some(),
					subgroup_delivery_timeout.is_some(),
					subscriber_priority.is_some(),
					filter.is_some(),
				];
				if subscription_params.contains(&true) && !Filter::is_draft20(version) {
					return Err(DecodeError::InvalidValue);
				}
				let mut properties = Properties::decode(r, version)?;
				properties.group_order = properties.group_order.or(group_order);

				let forward = forward.unwrap_or(true);

				Ok(Self {
					request_id,
					track_namespace,
					track_name,
					track_alias,
					largest_location,
					forward,
					properties,
				})
			}
		}
	}
}

#[derive(Debug)]
pub struct PublishOk {
	pub request_id: Option<RequestId>,
	pub forward: bool,
	pub subscriber_priority: u8,
	pub group_order: GroupOrder,
	pub filter: Filter,
	// pub parameters: Parameters,
}

impl Message for PublishOk {
	const ID: u64 = 0x1E;

	fn encode_msg<W: bytes::BufMut>(&self, w: &mut W, version: Version) -> Result<(), EncodeError> {
		if matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16) {
			self.request_id
				.expect("request_id required for draft14-16")
				.encode(w, version)?;
		} else {
			assert!(self.request_id.is_none(), "request_id must be None for draft17+");
		}

		match version {
			Version::Draft14 => {
				self.forward.encode(w, version)?;
				self.subscriber_priority.encode(w, version)?;
				self.group_order.encode(w, version)?;
				// Same as SUBSCRIBE: the Location an absolute filter carries is dropped on
				// decode, so encoding one would truncate the message.
				self.filter.encode(w, version)?;
				// no parameters
				0u8.encode(w, version)?;
			}
			// Draft-20 moved the subscription parameters out of PUBLISH_OK; they belong to
			// PUBLISH and REQUEST_UPDATE now, so a PUBLISH_OK carries none of them.
			_ if Filter::is_draft20(version) => encode_params!(w, version,),
			_ => {
				encode_params!(w, version,
					0x10 => self.forward,
					0x20 => self.subscriber_priority,
					0x21 => self.filter,
					0x22 => self.group_order,
				);
			}
		}

		Ok(())
	}

	fn decode_msg<R: bytes::Buf>(r: &mut R, version: Version) -> Result<Self, DecodeError> {
		let request_id = if matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16) {
			Some(RequestId::decode(r, version)?)
		} else {
			None
		};

		match version {
			Version::Draft14 => {
				let forward = bool::decode(r, version)?;
				let subscriber_priority = u8::decode(r, version)?;
				let group_order = GroupOrder::decode(r, version)?;
				let filter = Filter::decode(r, version)?;

				// no parameters
				let _params = Parameters::decode(r, version)?;

				Ok(Self {
					request_id,
					forward,
					subscriber_priority,
					group_order,
					filter,
				})
			}
			_ => {
				decode_params!(r, version,
					0x10 => forward: Option<bool>,
					0x20 => subscriber_priority: Option<u8>,
					0x21 => filter: Option<Filter>,
					0x22 => group_order: Option<GroupOrder>,
				);

				let forward = forward.unwrap_or(true);
				let subscriber_priority = subscriber_priority.unwrap_or(128);
				let group_order = group_order.unwrap_or(GroupOrder::Descending);
				let filter = filter.unwrap_or(Filter::Unfiltered);

				Ok(Self {
					request_id,
					forward,
					subscriber_priority,
					group_order,
					filter,
				})
			}
		}
	}
}

#[derive(Debug)]
pub struct PublishError<'a> {
	pub request_id: RequestId,
	pub error_code: u64,
	pub reason_phrase: Cow<'a, str>,
}
impl Message for PublishError<'_> {
	const ID: u64 = 0x1F;

	fn encode_msg<W: bytes::BufMut>(&self, w: &mut W, version: Version) -> Result<(), EncodeError> {
		self.request_id.encode(w, version)?;
		self.error_code.encode(w, version)?;
		self.reason_phrase.encode(w, version)?;
		Ok(())
	}

	fn decode_msg<R: bytes::Buf>(r: &mut R, version: Version) -> Result<Self, DecodeError> {
		let request_id = RequestId::decode(r, version)?;
		let error_code = u64::decode(r, version)?;
		let reason_phrase = Cow::<str>::decode(r, version)?;
		Ok(Self {
			request_id,
			error_code,
			reason_phrase,
		})
	}
}

#[cfg(test)]
mod tests {
	use super::*;

	/// Draft-20 moved the subscription parameters into PUBLISH. We refuse the request
	/// itself, but the message still has to parse: an unlisted parameter fails the whole
	/// decode, which kills the session instead of letting the peer get its refusal.
	#[test]
	fn publish_accepts_the_relocated_subscription_parameters() -> Result<(), EncodeError> {
		let mut body = Vec::new();
		RequestId(1).encode(&mut body, Version::Draft20).unwrap();
		super::super::namespace::encode_namespace(&mut body, &crate::Path::new("broadcast"), Version::Draft20).unwrap();
		"video".encode(&mut body, Version::Draft20).unwrap();
		1u64.encode(&mut body, Version::Draft20).unwrap(); // track alias

		// SUBSCRIBER_PRIORITY then LOCATION_FILTER, delta encoded from 0.
		encode_params!(&mut body, Version::Draft20,
			0x20 => 128u8,
			0x21 => Filter::NextObject,
		);
		Properties::default().encode(&mut body, Version::Draft20).unwrap();

		let mut buf = bytes::Bytes::from(body);
		Publish::decode_msg(&mut buf, Version::Draft20).expect("draft-20 PUBLISH parameters must parse");
		Ok(())
	}

	/// They arrived in draft-20, so an earlier peer sending one is still a violation.
	#[test]
	fn older_drafts_reject_the_relocated_parameters() -> Result<(), EncodeError> {
		let mut body = Vec::new();
		RequestId(1).encode(&mut body, Version::Draft19).unwrap();
		super::super::namespace::encode_namespace(&mut body, &crate::Path::new("broadcast"), Version::Draft19).unwrap();
		"video".encode(&mut body, Version::Draft19).unwrap();
		1u64.encode(&mut body, Version::Draft19).unwrap();
		encode_params!(&mut body, Version::Draft19, 0x20 => 128u8);
		Properties::default().encode(&mut body, Version::Draft19).unwrap();

		let mut buf = bytes::Bytes::from(body);
		assert!(Publish::decode_msg(&mut buf, Version::Draft19).is_err());
		Ok(())
	}
	use bytes::BytesMut;

	fn encode_message<M: Message>(msg: &M, version: Version) -> Vec<u8> {
		let mut buf = BytesMut::new();
		msg.encode_msg(&mut buf, version).unwrap();
		buf.to_vec()
	}

	fn decode_message<M: Message>(bytes: &[u8], version: Version) -> Result<M, DecodeError> {
		let mut buf = bytes::Bytes::from(bytes.to_vec());
		M::decode_msg(&mut buf, version)
	}

	#[test]
	fn test_publish_v14_round_trip() {
		let msg = Publish {
			request_id: RequestId(1),
			track_namespace: Path::new("test/ns"),
			track_name: "video".into(),
			track_alias: 42,
			largest_location: Some(Location { group: 10, object: 5 }),
			forward: true,
			properties: Properties {
				group_order: Some(GroupOrder::Descending),
				..Default::default()
			},
		};

		let encoded = encode_message(&msg, Version::Draft14);
		let decoded: Publish = decode_message(&encoded, Version::Draft14).unwrap();

		assert_eq!(decoded.request_id, RequestId(1));
		assert_eq!(decoded.track_namespace.as_str(), "test/ns");
		assert_eq!(decoded.track_name, "video");
		assert_eq!(decoded.track_alias, 42);
		assert_eq!(decoded.largest_location, Some(Location { group: 10, object: 5 }));
		assert!(decoded.forward);
	}

	#[test]
	fn test_publish_v15_round_trip() {
		let msg = Publish {
			request_id: RequestId(1),
			track_namespace: Path::new("test/ns"),
			track_name: "video".into(),
			track_alias: 42,
			largest_location: Some(Location { group: 10, object: 5 }),
			forward: true,
			properties: Properties {
				group_order: Some(GroupOrder::Descending),
				..Default::default()
			},
		};

		let encoded = encode_message(&msg, Version::Draft15);
		let decoded: Publish = decode_message(&encoded, Version::Draft15).unwrap();

		assert_eq!(decoded.request_id, RequestId(1));
		assert_eq!(decoded.track_namespace.as_str(), "test/ns");
		assert_eq!(decoded.track_name, "video");
		assert_eq!(decoded.track_alias, 42);
		assert_eq!(decoded.largest_location, Some(Location { group: 10, object: 5 }));
		assert!(decoded.forward);
	}

	#[test]
	fn test_publish_ok_v14_round_trip() {
		let msg = PublishOk {
			request_id: Some(RequestId(7)),
			forward: true,
			subscriber_priority: 128,
			group_order: GroupOrder::Descending,
			filter: Filter::NextObject,
		};

		let encoded = encode_message(&msg, Version::Draft14);
		let decoded: PublishOk = decode_message(&encoded, Version::Draft14).unwrap();

		assert_eq!(decoded.request_id, Some(RequestId(7)));
		assert!(decoded.forward);
		assert_eq!(decoded.subscriber_priority, 128);
	}

	#[test]
	fn test_publish_ok_v15_round_trip() {
		let msg = PublishOk {
			request_id: Some(RequestId(7)),
			forward: true,
			subscriber_priority: 128,
			group_order: GroupOrder::Descending,
			filter: Filter::NextObject,
		};

		let encoded = encode_message(&msg, Version::Draft15);
		let decoded: PublishOk = decode_message(&encoded, Version::Draft15).unwrap();

		assert_eq!(decoded.request_id, Some(RequestId(7)));
		assert!(decoded.forward);
		assert_eq!(decoded.subscriber_priority, 128);
	}

	#[test]
	fn test_publish_v17_round_trip() {
		let msg = Publish {
			request_id: RequestId(1),
			track_namespace: Path::new("test/ns"),
			track_name: "video".into(),
			track_alias: 42,
			largest_location: Some(Location { group: 10, object: 5 }),
			forward: true,
			properties: Properties {
				group_order: Some(GroupOrder::Descending),
				..Default::default()
			},
		};

		let encoded = encode_message(&msg, Version::Draft17);
		let decoded: Publish = decode_message(&encoded, Version::Draft17).unwrap();

		assert_eq!(decoded.request_id, RequestId(1));
		assert_eq!(decoded.track_namespace.as_str(), "test/ns");
		assert_eq!(decoded.track_name, "video");
		assert_eq!(decoded.track_alias, 42);
		assert_eq!(decoded.largest_location, Some(Location { group: 10, object: 5 }));
		assert!(decoded.forward);
	}

	#[test]
	fn test_publish_ok_v17_round_trip() {
		let msg = PublishOk {
			request_id: None,
			forward: true,
			subscriber_priority: 128,
			group_order: GroupOrder::Descending,
			filter: Filter::NextObject,
		};

		let encoded = encode_message(&msg, Version::Draft17);
		let decoded: PublishOk = decode_message(&encoded, Version::Draft17).unwrap();

		assert_eq!(decoded.request_id, None);
		assert!(decoded.forward);
		assert_eq!(decoded.subscriber_priority, 128);
	}

	#[test]
	fn test_publish_done_registered_statuses_all_versions() {
		for version in [
			Version::Draft14,
			Version::Draft15,
			Version::Draft16,
			Version::Draft17,
			Version::Draft18,
			Version::Draft19,
		] {
			for (status, expected) in [
				(PublishDoneStatus::InternalError, 0x0),
				(PublishDoneStatus::TrackEnded, 0x2),
			] {
				assert_eq!(status.code(version), expected);

				let msg = PublishDone {
					request_id: matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16)
						.then_some(RequestId(7)),
					status_code: status.code(version),
					stream_count: 5,
					reason_phrase: "done".into(),
				};
				let encoded = encode_message(&msg, version);
				let decoded: PublishDone = decode_message(&encoded, version).unwrap();

				assert_eq!(decoded.status_code, expected);
				assert_eq!(decoded.stream_count, 5);
				assert_eq!(decoded.reason_phrase, "done");
			}
		}
	}

	#[test]
	fn test_publish_v18_round_trip() {
		let msg = Publish {
			request_id: RequestId(1),
			track_namespace: Path::new("test/ns"),
			track_name: "video".into(),
			track_alias: 42,
			largest_location: Some(Location { group: 10, object: 5 }),
			forward: true,
			properties: Properties {
				group_order: Some(GroupOrder::Descending),
				..Default::default()
			},
		};

		let encoded = encode_message(&msg, Version::Draft18);
		let decoded: Publish = decode_message(&encoded, Version::Draft18).unwrap();

		assert_eq!(decoded.request_id, RequestId(1));
		assert_eq!(decoded.track_namespace.as_str(), "test/ns");
		assert_eq!(decoded.track_name, "video");
		assert_eq!(decoded.track_alias, 42);
		assert_eq!(decoded.largest_location, Some(Location { group: 10, object: 5 }));
		assert!(decoded.forward);
	}

	/// GROUP_ORDER (0x22) is only a legal PUBLISH *message parameter* through draft-15; a
	/// draft-16+ peer closes the session with PROTOCOL_VIOLATION when it sees one. The
	/// publisher's preference belongs in the DEFAULT_PUBLISHER_GROUP_ORDER track property,
	/// which shares the number 0x22 in the separate property registry.
	#[test]
	fn test_publish_v18_group_order_is_a_property() {
		let msg = Publish {
			request_id: RequestId(1),
			track_namespace: Path::new("ns"),
			track_name: "video".into(),
			track_alias: 42,
			largest_location: None,
			forward: true,
			properties: Properties {
				timescale: None,
				group_order: Some(GroupOrder::Descending),
			},
		};

		#[rustfmt::skip]
		let expected = vec![
			1,                            // request id
			1,                            // namespace tuple count
			2, b'n', b's',                // "ns"
			5, b'v', b'i', b'd', b'e', b'o', // "video"
			42,                           // track alias
			1,                            // one message parameter
			0x10, 0x01,                   // FORWARD = true
			0x22, 0x02,                   // DEFAULT_PUBLISHER_GROUP_ORDER = descending
		];
		assert_eq!(encode_message(&msg, Version::Draft18), expected);

		let decoded: Publish = decode_message(&expected, Version::Draft18).unwrap();
		assert_eq!(decoded.properties.group_order, Some(GroupOrder::Descending));
	}

	/// Draft-15 is the last version that takes it as a message parameter, and has no track
	/// properties to put it in.
	#[test]
	fn test_publish_v15_group_order_is_a_parameter() {
		let msg = Publish {
			request_id: RequestId(1),
			track_namespace: Path::new("ns"),
			track_name: "video".into(),
			track_alias: 42,
			largest_location: None,
			forward: true,
			properties: Properties {
				timescale: None,
				group_order: Some(GroupOrder::Descending),
			},
		};

		#[rustfmt::skip]
		let expected = vec![
			1,                            // request id
			1,                            // namespace tuple count
			2, b'n', b's',                // "ns"
			5, b'v', b'i', b'd', b'e', b'o', // "video"
			42,                           // track alias
			2,                            // two message parameters
			0x10, 0x01,                   // FORWARD = true
			0x22, 0x02,                   // GROUP_ORDER = descending
		];
		assert_eq!(encode_message(&msg, Version::Draft15), expected);

		let decoded: Publish = decode_message(&expected, Version::Draft15).unwrap();
		assert_eq!(decoded.properties.group_order, Some(GroupOrder::Descending));
	}

	/// Draft-18 drops the required_request_id_delta varint (#1615).
	/// For Publish (#1D) the field is a single zero varint = 1 byte.
	#[test]
	fn test_publish_v18_is_one_byte_shorter_than_v17() {
		let msg = Publish {
			request_id: RequestId(1),
			track_namespace: Path::new("test/ns"),
			track_name: "video".into(),
			track_alias: 42,
			largest_location: None,
			forward: true,
			properties: Properties {
				group_order: Some(GroupOrder::Descending),
				..Default::default()
			},
		};

		let v17 = encode_message(&msg, Version::Draft17);
		let v18 = encode_message(&msg, Version::Draft18);
		assert_eq!(v17.len(), v18.len() + 1);
	}

	#[test]
	fn test_publish_ok_v18_round_trip() {
		let msg = PublishOk {
			request_id: None,
			forward: true,
			subscriber_priority: 128,
			group_order: GroupOrder::Descending,
			filter: Filter::NextObject,
		};

		let encoded = encode_message(&msg, Version::Draft18);
		let decoded: PublishOk = decode_message(&encoded, Version::Draft18).unwrap();

		assert_eq!(decoded.request_id, None);
		assert!(decoded.forward);
		assert_eq!(decoded.subscriber_priority, 128);
	}

	#[test]
	fn test_publish_done_preserves_unknown_status() {
		const UNKNOWN: u64 = 0xface;
		let msg = PublishDone {
			request_id: None,
			status_code: UNKNOWN,
			stream_count: 5,
			reason_phrase: "extension".into(),
		};

		let encoded = encode_message(&msg, Version::Draft19);
		let decoded: PublishDone = decode_message(&encoded, Version::Draft19).unwrap();

		assert_eq!(decoded.status_code, UNKNOWN);
	}
}