Clippy

moq-transport/src/message/mod.rs

@@ -36,7 +36,6 @@ mod announce_cancel;
 mod announce_error;
 mod announce_ok;
 mod go_away;
-mod message;
 mod publisher;
 mod subscribe;
 mod subscribe_done;
@@ -51,7 +50,6 @@ pub use announce_cancel::*;
 pub use announce_error::*;
 pub use announce_ok::*;
 pub use go_away::*;
-pub use message::*;
 pub use publisher::*;
 pub use subscribe::*;
 pub use subscribe_done::*;
@@ -60,3 +58,102 @@ pub use subscribe_ok::*;
 pub use subscriber::*;
 pub use unannounce::*;
 pub use unsubscribe::*;
+
+use crate::coding::{AsyncRead, AsyncWrite, Decode, DecodeError, Encode, EncodeError};
+use std::fmt;
+
+// Use a macro to generate the message types rather than copy-paste.
+// This implements a decode/encode method that uses the specified type.
+macro_rules! message_types {
+    {$($name:ident = $val:expr,)*} => {
+		/// All supported message types.
+		#[derive(Clone)]
+		pub enum Message {
+			$($name($name)),*
+		}
+
+		impl Message {
+			pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
+				let t = u64::decode(r).await?;
+
+				match t {
+					$($val => {
+						let msg = $name::decode(r).await?;
+						Ok(Self::$name(msg))
+					})*
+					_ => Err(DecodeError::InvalidMessage(t)),
+				}
+			}
+
+			pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
+				match self {
+					$(Self::$name(ref m) => {
+						self.id().encode(w).await?;
+						m.encode(w).await
+					},)*
+				}
+			}
+
+			pub fn id(&self) -> u64 {
+				match self {
+					$(Self::$name(_) => {
+						$val
+					},)*
+				}
+			}
+
+			pub fn name(&self) -> &'static str {
+				match self {
+					$(Self::$name(_) => {
+						stringify!($name)
+					},)*
+				}
+			}
+		}
+
+		$(impl From<$name> for Message {
+			fn from(m: $name) -> Self {
+				Message::$name(m)
+			}
+		})*
+
+		impl fmt::Debug for Message {
+			// Delegate to the message formatter
+			fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+				match self {
+					$(Self::$name(ref m) => m.fmt(f),)*
+				}
+			}
+		}
+    }
+}
+
+// Each message is prefixed with the given VarInt type.
+message_types! {
+	// NOTE: Object and Setup are in other modules.
+	// Object = 0x0
+	// ObjectUnbounded = 0x2
+	// SetupClient = 0x40
+	// SetupServer = 0x41
+
+	// SUBSCRIBE family, sent by subscriber
+	Subscribe = 0x3,
+	Unsubscribe = 0xa,
+
+	// SUBSCRIBE family, sent by publisher
+	SubscribeOk = 0x4,
+	SubscribeError = 0x5,
+	SubscribeDone = 0xb,
+
+	// ANNOUNCE family, sent by publisher
+	Announce = 0x6,
+	Unannounce = 0x9,
+
+	// ANNOUNCE family, sent by subscriber
+	AnnounceOk = 0x7,
+	AnnounceError = 0x8,
+	AnnounceCancel = 0xc,
+
+	// Misc
+	GoAway = 0x10,
+}

moq-transport/src/message/publisher.rs

@@ -7,9 +7,9 @@ macro_rules! publisher_msgs {
 			$($name(message::$name)),*
 		}
 
-		$(impl Into<Publisher> for message::$name {
-			fn into(self) -> Publisher {
-				Publisher::$name(self)
+		$(impl From<message::$name> for Publisher {
+			fn from(msg: message::$name) -> Self {
+				Publisher::$name(msg)
 			}
 		})*
 

moq-transport/src/message/subscriber.rs

@@ -7,9 +7,9 @@ macro_rules! subscriber_msgs {
 			$($name(message::$name)),*
 		}
 
-		$(impl Into<Subscriber> for message::$name {
-			fn into(self) -> Subscriber {
-				Subscriber::$name(self)
+		$(impl From<message::$name> for Subscriber {
+			fn from(msg: message::$name) -> Self {
+				Subscriber::$name(msg)
 			}
 		})*
 

moq-transport/src/serve/group.rs

@@ -93,7 +93,7 @@ impl GroupPublisher {
 	pub fn create_object(&mut self, size: usize) -> Result<ObjectPublisher, ServeError> {
 		let (publisher, subscriber) = ObjectHeader {
 			group_id: self.id,
-			object_id: self.next.try_into().unwrap(),
+			object_id: self.next,
 			send_order: self.send_order,
 			size,
 		}

moq-transport/src/serve/object.rs

@@ -133,7 +133,7 @@ impl ObjectPublisher {
 	fn new(state: Watch<State>, info: Arc<ObjectHeader>) -> Self {
 		Self {
 			state,
-			remain: info.size as usize,
+			remain: info.size,
 			info,
 		}
 	}

moq-transport/src/serve/track.rs

@@ -18,7 +18,7 @@ use super::{
 	Datagram, Group, GroupPublisher, GroupSubscriber, Object, ObjectHeader, ObjectPublisher, ObjectSubscriber,
 	ServeError, Stream, StreamPublisher, StreamSubscriber,
 };
-use std::{fmt, ops::Deref, sync::Arc};
+use std::{cmp, fmt, ops::Deref, sync::Arc};
 
 /// Static information about a track.
 #[derive(Debug)]
@@ -70,14 +70,11 @@ impl State {
 
 		match &self.mode {
 			Mode::Init => {}
-			Mode::Group(old) => {
-				if old.id == group.id {
-					return Err(ServeError::Duplicate);
-				} else if old.id > group.id {
-					log::warn!("dropping old group");
-					return Ok(());
-				}
-			}
+			Mode::Group(old) => match group.id.cmp(&old.id) {
+				cmp::Ordering::Less => return Ok(()),
+				cmp::Ordering::Equal => return Err(ServeError::Duplicate),
+				cmp::Ordering::Greater => {}
+			},
 			_ => return Err(ServeError::Mode),
 		};
 
@@ -97,11 +94,11 @@ impl State {
 			Mode::Object(objects) => {
 				let first = objects.first().unwrap();
 
-				if first.group_id > object.group_id {
+				match object.group_id.cmp(&first.group_id) {
 					// Drop this old group
-					return Ok(());
-				} else if first.group_id < object.group_id {
-					objects.clear()
+					cmp::Ordering::Less => return Ok(()),
+					cmp::Ordering::Greater => objects.clear(),
+					cmp::Ordering::Equal => {}
 				}
 
 				objects.push(object);