moq-transport: add fetch messages

Author: mcintyrehh

moq-transport/src/message/fetch.rs

@@ -0,0 +1,83 @@
+use crate::coding::{Decode, DecodeError, Encode, EncodeError, Params, Tuple};
+use crate::message::GroupOrder;
+
+/// Sent by the subscriber to request to request a range
+/// of already published objects within a track.
+#[derive(Clone, Debug)]
+pub struct Fetch {
+    /// The subscription ID
+    pub id: u64,
+
+    /// Track properties
+    pub track_namespace: Tuple,
+    pub track_name: String,
+
+    /// Subscriber Priority
+    pub subscriber_priority: u8,
+
+    pub group_order: GroupOrder,
+
+    /// The start/end group/object.
+    pub start_group: u64,
+    pub start_object: u64,
+    pub end_group: u64,
+    pub end_object: u64,
+
+    /// Optional parameters
+    pub params: Params,
+}
+
+impl Decode for Fetch {
+    fn decode<R: bytes::Buf>(r: &mut R) -> Result<Self, DecodeError> {
+        let id = u64::decode(r)?;
+
+        let track_namespace = Tuple::decode(r)?;
+        let track_name = String::decode(r)?;
+
+        let subscriber_priority = u8::decode(r)?;
+
+        let group_order = GroupOrder::decode(r)?;
+
+        let start_group = u64::decode(r)?;
+        let start_object = u64::decode(r)?;
+        let end_group = u64::decode(r)?;
+        let end_object = u64::decode(r)?;
+
+        let params = Params::decode(r)?;
+
+        Ok(Self {
+            id,
+            track_namespace,
+            track_name,
+            subscriber_priority,
+            group_order,
+            start_group,
+            start_object,
+            end_group,
+            end_object,
+            params,
+        })
+    }
+}
+
+impl Encode for Fetch {
+    fn encode<W: bytes::BufMut>(&self, w: &mut W) -> Result<(), EncodeError> {
+        self.id.encode(w)?;
+
+        self.track_namespace.encode(w)?;
+        self.track_name.encode(w)?;
+
+        self.subscriber_priority.encode(w)?;
+
+        self.group_order.encode(w)?;
+
+        self.start_group.encode(w)?;
+        self.start_object.encode(w)?;
+        self.end_group.encode(w)?;
+        self.end_object.encode(w)?;
+
+        self.params.encode(w)?;
+
+        Ok(())
+    }
+}

moq-transport/src/message/fetch_cancel.rs

@@ -0,0 +1,24 @@
+use crate::coding::{Decode, DecodeError, Encode, EncodeError};
+
+/// A subscriber issues a FETCH_CANCEL message to a publisher indicating it is
+/// no longer interested in receiving Objects for the fetch specified by 'Subscribe ID'.
+#[derive(Clone, Debug)]
+pub struct FetchCancel {
+    /// The subscription ID
+    pub id: u64,
+}
+
+impl Decode for FetchCancel {
+    fn decode<R: bytes::Buf>(r: &mut R) -> Result<Self, DecodeError> {
+        let id = u64::decode(r)?;
+        Ok(Self { id })
+    }
+}
+
+impl Encode for FetchCancel {
+    fn encode<W: bytes::BufMut>(&self, w: &mut W) -> Result<(), EncodeError> {
+        self.id.encode(w)?;
+
+        Ok(())
+    }
+}

moq-transport/src/message/fetch_error.rs

@@ -0,0 +1,36 @@
+use crate::coding::{Decode, DecodeError, Encode, EncodeError};
+
+/// Sent by the server to indicate that the client should connect to a different server.
+#[derive(Clone, Debug)]
+pub struct FetchError {
+    /// The ID for this subscription.
+    pub id: u64,
+
+    /// An error code.
+    pub code: u64,
+
+    /// An optional, human-readable reason.
+    pub reason: String,
+}
+
+impl Decode for FetchError {
+    fn decode<R: bytes::Buf>(r: &mut R) -> Result<Self, DecodeError> {
+        let id = u64::decode(r)?;
+
+        let code = u64::decode(r)?;
+        let reason = String::decode(r)?;
+
+        Ok(Self { id, code, reason })
+    }
+}
+
+impl Encode for FetchError {
+    fn encode<W: bytes::BufMut>(&self, w: &mut W) -> Result<(), EncodeError> {
+        self.id.encode(w)?;
+
+        self.code.encode(w)?;
+        self.reason.encode(w)?;
+
+        Ok(())
+    }
+}

moq-transport/src/message/fetch_ok.rs

@@ -0,0 +1,56 @@
+use crate::coding::{Decode, DecodeError, Encode, EncodeError, Params};
+
+/// A publisher sends a FETCH_OK control message in response to successful fetches.
+#[derive(Clone, Debug)]
+pub struct FetchOk {
+    /// The subscription ID
+    pub id: u64,
+
+    /// Order groups will be delivered in
+    pub group_order: u8,
+
+    /// True if all objects have been published on this track
+    pub end_of_track: u8,
+
+    /// The largest Group ID available for this track (last if end_of_track)
+    pub largest_group_id: u64,
+    /// The largest Object ID available within the largest Group ID for this track (last if end_of_track)
+    pub largest_object_id: u64,
+
+    /// Optional parameters
+    pub params: Params,
+}
+
+impl Decode for FetchOk {
+    fn decode<R: bytes::Buf>(r: &mut R) -> Result<Self, DecodeError> {
+        let id = u64::decode(r)?;
+
+        let group_order = u8::decode(r)?;
+
+        let end_of_track = u8::decode(r)?;
+
+        let largest_group_id = u64::decode(r)?;
+        let largest_object_id = u64::decode(r)?;
+
+        let params = Params::decode(r)?;
+
+        Ok(Self { id, group_order, end_of_track, largest_group_id, largest_object_id, params })
+    }
+}
+
+impl Encode for FetchOk {
+    fn encode<W: bytes::BufMut>(&self, w: &mut W) -> Result<(), EncodeError> {
+        self.id.encode(w)?;
+
+        self.group_order.encode(w)?;
+
+        self.end_of_track.encode(w)?;
+
+        self.largest_group_id.encode(w)?;
+        self.largest_object_id.encode(w)?;
+
+        self.params.encode(w)?;
+
+        Ok(())
+    }
+}

moq-transport/src/message/mod.rs

@@ -55,6 +55,10 @@ mod track_status_request;
 mod unannounce;
 mod unsubscribe;
 mod unsubscribe_namespace;
+mod fetch;
+mod fetch_cancel;
+mod fetch_ok;
+mod fetch_error;
 
 pub use announce::*;
 pub use announce_cancel::*;
@@ -79,6 +83,10 @@ pub use track_status_request::*;
 pub use unannounce::*;
 pub use unsubscribe::*;
 pub use unsubscribe_namespace::*;
+pub use fetch::*;
+pub use fetch_cancel::*;
+pub use fetch_ok::*;
+pub use fetch_error::*;
 
 use crate::coding::{Decode, DecodeError, Encode, EncodeError};
 use std::fmt;
@@ -209,6 +217,12 @@ message_types! {
     SubscribeNamespaceOk = 0x12,
     SubscribeNamespaceError = 0x13,
     UnsubscribeNamespace = 0x14,
+
+    // FETCH family, sent by subscriber
+    Fetch = 0x16,
+    FetchCancel = 0x17,
+    FetchOk = 0x18,
+    FetchError = 0x19,
 }
 
 /// Track Status Codes

moq-transport/src/message/publisher.rs

@@ -51,4 +51,6 @@ publisher_msgs! {
     SubscribeError,
     SubscribeDone,
     TrackStatus,
+    FetchOk,
+    FetchError,
 }

moq-transport/src/message/subscriber.rs

@@ -56,4 +56,6 @@ subscriber_msgs! {
     SubscribeNamespaceOk,
     SubscribeNamespaceError,
     UnsubscribeNamespace,
+    Fetch,
+    FetchCancel,
 }

moq-transport/src/session/publisher.rs

@@ -183,6 +183,9 @@ impl Publisher {
             message::Subscriber::SubscribeNamespaceOk(_msg) => unimplemented!(),
             message::Subscriber::SubscribeNamespaceError(_msg) => unimplemented!(),
             message::Subscriber::UnsubscribeNamespace(_msg) => unimplemented!(),
+            // TODO: Implement fetch messages
+            message::Subscriber::Fetch(_msg) => todo!(),
+            message::Subscriber::FetchCancel(_msg) => todo!(),
         };
 
         if let Err(err) = res {

moq-transport/src/session/subscriber.rs

@@ -86,6 +86,9 @@ impl Subscriber {
             message::Publisher::SubscribeError(msg) => self.recv_subscribe_error(msg),
             message::Publisher::SubscribeDone(msg) => self.recv_subscribe_done(msg),
             message::Publisher::TrackStatus(msg) => self.recv_track_status(msg),
+            // TODO: Implement fetch messages
+            message::Publisher::FetchOk(_msg) => todo!(),
+            message::Publisher::FetchError(_msg) => todo!(),
         };
 
         if let Err(SessionError::Serve(err)) = res {