Merge pull request #32 from englishm/me/draft-07

moq-transport draft-07 initial wire format updates

Author: englishm

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,63 @@
+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

@@ -36,6 +36,10 @@ mod announce;
 mod announce_cancel;
 mod announce_error;
 mod announce_ok;
+mod fetch;
+mod fetch_cancel;
+mod fetch_error;
+mod fetch_ok;
 mod filter_type;
 mod go_away;
 mod group_order;
@@ -60,6 +64,10 @@ pub use announce::*;
 pub use announce_cancel::*;
 pub use announce_error::*;
 pub use announce_ok::*;
+pub use fetch::*;
+pub use fetch_cancel::*;
+pub use fetch_error::*;
+pub use fetch_ok::*;
 pub use filter_type::*;
 pub use go_away::*;
 pub use group_order::*;
@@ -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

@@ -52,4 +52,6 @@ publisher_msgs! {
     SubscribeDone,
     MaxSubscribeId,
     TrackStatus,
+    FetchOk,
+    FetchError,
 }

moq-transport/src/message/subscriber.rs

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

moq-transport/src/session/mod.rs

@@ -81,7 +81,7 @@ impl Session {
         let mut sender = Writer::new(control.0);
         let mut recver = Reader::new(control.1);
 
-        let versions: setup::Versions = [setup::Version::DRAFT_06].into();
+        let versions: setup::Versions = [setup::Version::DRAFT_07].into();
 
         let client = setup::Client {
             role,
@@ -134,10 +134,10 @@ impl Session {
         let client: setup::Client = recver.decode().await?;
         log::debug!("received client SETUP: {:?}", client);
 
-        if !client.versions.contains(&setup::Version::DRAFT_06) {
+        if !client.versions.contains(&setup::Version::DRAFT_07) {
             return Err(SessionError::Version(
                 client.versions,
-                [setup::Version::DRAFT_06].into(),
+                [setup::Version::DRAFT_07].into(),
             ));
         }
 
@@ -162,7 +162,7 @@ impl Session {
 
         let server = setup::Server {
             role,
-            version: setup::Version::DRAFT_06,
+            version: setup::Version::DRAFT_07,
             params: Default::default(),
         };
 

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

@@ -87,6 +87,9 @@ impl Subscriber {
             message::Publisher::SubscribeDone(msg) => self.recv_subscribe_done(msg),
             message::Publisher::MaxSubscribeId(msg) => self.recv_max_subscribe_id(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 {