Merge pull request #94 from sgodin/renames-and-comments-only

Author: englishm

moq-clock-ietf/src/clock.rs

@@ -7,64 +7,72 @@ use moq_transport::serve::{
 use chrono::prelude::*;
 use tokio::task;
 
+/// Publishes the current time every second in the format "YYYY-MM-DD HH:MM:SS"
 pub struct Publisher {
-    track: SubgroupsWriter,
+    track_subgroups_writer: SubgroupsWriter,
 }
 
 impl Publisher {
-    pub fn new(track: SubgroupsWriter) -> Self {
-        Self { track }
+    pub fn new(track_subgroups_writer: SubgroupsWriter) -> Self {
+        Self {
+            track_subgroups_writer,
+        }
     }
 
+    /// Runs the publisher, sending the current time every second.  Creates a new group for each minute.
     pub async fn run(mut self) -> anyhow::Result<()> {
         let start = Utc::now();
         let mut now = start;
 
         // Just for fun, don't start at zero.
-        let mut sequence = start.minute();
+        let mut next_group_id = start.minute();
 
+        // Create a new group for each minute.
         loop {
-            let segment = self
-                .track
+            let subgroup_writer = self
+                .track_subgroups_writer
                 .create(Subgroup {
-                    group_id: sequence as u64,
+                    group_id: next_group_id as u64,
                     subgroup_id: 0,
                     priority: 0,
                 })
                 .context("failed to create minute segment")?;
 
-            sequence += 1;
+            next_group_id += 1;
 
+            // Spawn a new task to handle sending the object every second
             tokio::spawn(async move {
-                if let Err(err) = Self::send_segment(segment, now).await {
+                if let Err(err) = Self::send_subgroup_objects(subgroup_writer, now).await {
                     log::warn!("failed to send minute: {:?}", err);
                 }
             });
 
             let next = now + chrono::Duration::try_minutes(1).unwrap();
             let next = next.with_second(0).unwrap().with_nanosecond(0).unwrap();
 
+            // Sleep until the start of the next minute
             let delay = (next - now).to_std().unwrap();
             tokio::time::sleep(delay).await;
 
             now = next; // just assume we didn't undersleep
         }
     }
 
-    async fn send_segment(
-        mut segment: SubgroupWriter,
+    /// Sends the current time every second within a minute group.
+    async fn send_subgroup_objects(
+        mut subgroup_writer: SubgroupWriter,
         mut now: DateTime<Utc>,
     ) -> anyhow::Result<()> {
         // Everything but the second.
         let base = now.format("%Y-%m-%d %H:%M:").to_string();
 
-        segment
+        subgroup_writer
             .write(base.clone().into())
             .context("failed to write base")?;
 
         loop {
             let delta = now.format("%S").to_string();
-            segment
+            subgroup_writer
                 .write(delta.clone().into())
                 .context("failed to write delta")?;
 
@@ -73,6 +81,7 @@ impl Publisher {
             let next = now + chrono::Duration::try_seconds(1).unwrap();
             let next = next.with_nanosecond(0).unwrap();
 
+            // Sleep until the next second
             let delay = (next - now).to_std().unwrap();
             tokio::time::sleep(delay).await;
 
@@ -86,26 +95,35 @@ impl Publisher {
         }
     }
 }
+
+/// Subscribes to the clock and prints received time updates to stdout.
 pub struct Subscriber {
-    track: TrackReader,
+    track_reader: TrackReader,
 }
 
 impl Subscriber {
-    pub fn new(track: TrackReader) -> Self {
-        Self { track }
+    pub fn new(track_reader: TrackReader) -> Self {
+        Self { track_reader }
     }
 
+    /// Runs the subscriber, receiving time updates and printing them to stdout.
     pub async fn run(self) -> anyhow::Result<()> {
-        match self.track.mode().await.context("failed to get mode")? {
+        match self
+            .track_reader
+            .mode()
+            .await
+            .context("failed to get mode")?
+        {
             TrackReaderMode::Stream(stream) => Self::recv_stream(stream).await,
             TrackReaderMode::Subgroups(subgroups) => Self::recv_subgroups(subgroups).await,
             TrackReaderMode::Datagrams(datagrams) => Self::recv_datagrams(datagrams).await,
         }
     }
 
-    async fn recv_stream(mut track: StreamReader) -> anyhow::Result<()> {
-        while let Some(mut subgroup) = track.next().await? {
-            while let Some(object) = subgroup.read_next().await? {
+    /// Receives time updates from a stream and prints them to stdout.
+    async fn recv_stream(mut stream_reader: StreamReader) -> anyhow::Result<()> {
+        while let Some(mut stream_group_reader) = stream_reader.next().await? {
+            while let Some(object) = stream_group_reader.read_next().await? {
                 let str = String::from_utf8_lossy(&object);
                 println!("{str}");
             }
@@ -114,20 +132,22 @@ impl Subscriber {
         Ok(())
     }
 
-    async fn recv_subgroups(mut subgroups: SubgroupsReader) -> anyhow::Result<()> {
-        while let Some(mut subgroup) = subgroups.next().await? {
-            // Spawn a new task to handle the subgroup concurrently
+    /// Receives time updates from subgroups and prints them to stdout.
+    async fn recv_subgroups(mut subgroups_reader: SubgroupsReader) -> anyhow::Result<()> {
+        while let Some(mut subgroup_reader) = subgroups_reader.next().await? {
+            // Spawn a new task to handle the subgroup concurrently, so we
+            // don't rely on the publisher ending the previous stream before starting a new one.
             task::spawn(async move {
                 if let Err(e) = async {
-                    let base = subgroup
+                    let base = subgroup_reader
                         .read_next()
                         .await
                         .context("failed to get first object")?
                         .context("empty subgroup")?;
 
                     let base = String::from_utf8_lossy(&base);
 
-                    while let Some(object) = subgroup.read_next().await? {
+                    while let Some(object) = subgroup_reader.read_next().await? {
                         let str = String::from_utf8_lossy(&object);
                         println!("{base}{str}");
                     }
@@ -144,8 +164,9 @@ impl Subscriber {
         Ok(())
     }
 
-    async fn recv_datagrams(mut datagrams: DatagramsReader) -> anyhow::Result<()> {
-        while let Some(datagram) = datagrams.read().await? {
+    /// Receives time updates from datagrams and prints them to stdout.
+    async fn recv_datagrams(mut datagrams_reader: DatagramsReader) -> anyhow::Result<()> {
+        while let Some(datagram) = datagrams_reader.read().await? {
             let str = String::from_utf8_lossy(&datagram.payload);
             println!("{str}");
         }

moq-clock-ietf/src/main.rs

@@ -40,6 +40,7 @@ pub struct Cli {
     pub track: String,
 }
 
+/// The main entry point for the MoQ Clock IETF example.
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     env_logger::init();
@@ -53,6 +54,7 @@ async fn main() -> anyhow::Result<()> {
     let config = Cli::parse();
     let tls = config.tls.load()?;
 
+    // Create the QUIC endpoint
     let quic = quic::Endpoint::new(quic::Config {
         bind: config.bind,
         qlog_dir: None,
@@ -61,48 +63,52 @@ async fn main() -> anyhow::Result<()> {
 
     log::info!("connecting to server: url={}", config.url);
 
+    // Connect to the server
     let (session, connection_id) = quic.client.connect(&config.url).await?;
 
     log::info!(
         "connected with CID: {} (use this to look up qlog/mlog on server)",
         connection_id
     );
 
+    // Depending on whether we are publishing or subscribing, create the appropriate session
     if config.publish {
+        // Create the publisher session
         let (session, mut publisher) = Publisher::connect(session)
             .await
             .context("failed to create MoQ Transport session")?;
 
-        let (mut writer, _, reader) = serve::Tracks {
+        let (mut tracks_writer, _, tracks_reader) = serve::Tracks {
             namespace: TrackNamespace::from_utf8_path(&config.namespace),
         }
         .produce();
 
-        let track = writer.create(&config.track).unwrap();
-        let clock = clock::Publisher::new(track.groups()?);
+        let track_writer = tracks_writer.create(&config.track).unwrap();
+        let clock_publisher = clock::Publisher::new(track_writer.subgroups()?);
 
         tokio::select! {
             res = session.run() => res.context("session error")?,
-            res = clock.run() => res.context("clock error")?,
-            res = publisher.announce(reader) => res.context("failed to serve tracks")?,
+            res = clock_publisher.run() => res.context("clock error")?,
+            res = publisher.announce(tracks_reader) => res.context("failed to serve tracks")?,
         }
     } else {
+        // Create the subscriber session
         let (session, mut subscriber) = Subscriber::connect(session)
             .await
             .context("failed to create MoQ Transport session")?;
 
-        let (prod, sub) = serve::Track::new(
+        let (track_writer, track_reader) = serve::Track::new(
             TrackNamespace::from_utf8_path(&config.namespace),
             config.track,
         )
         .produce();
 
-        let clock = clock::Subscriber::new(sub);
+        let clock_subscriber = clock::Subscriber::new(track_reader);
 
         tokio::select! {
             res = session.run() => res.context("session error")?,
-            res = clock.run() => res.context("clock error")?,
-            res = subscriber.subscribe(prod) => res.context("failed to subscribe to track")?,
+            res = clock_subscriber.run() => res.context("clock error")?,
+            res = subscriber.subscribe(track_writer) => res.context("failed to subscribe to track")?,
         }
     }
 

moq-pub/src/media.rs

@@ -31,11 +31,11 @@ impl Media {
         let catalog = broadcast
             .create(".catalog")
             .context("broadcast closed")?
-            .groups()?;
+            .subgroups()?;
         let init = broadcast
             .create("0.mp4")
             .context("broadcast closed")?
-            .groups()?;
+            .subgroups()?;
 
         Ok(Media {
             tracks: Default::default(),
@@ -333,7 +333,7 @@ struct Track {
 impl Track {
     fn new(track: TrackWriter, handler: TrackType, timescale: u64) -> Self {
         Self {
-            track: track.groups().unwrap(),
+            track: track.subgroups().unwrap(),
             current: None,
             timescale,
             handler,

moq-relay-ietf/src/api.rs

@@ -1,5 +1,6 @@
 use url::Url;
 
+/// API client for moq-api.
 #[derive(Clone)]
 pub struct Api {
     client: moq_api::Client,
@@ -14,12 +15,14 @@ impl Api {
         Self { client, origin }
     }
 
+    /// Set the origin for a given namespace, returning a refresher.
     pub async fn set_origin(&self, namespace: String) -> Result<Refresh, moq_api::ApiError> {
         let refresh = Refresh::new(self.client.clone(), self.origin.clone(), namespace);
         refresh.update().await?;
         Ok(refresh)
     }
 
+    /// Get the origin for a given namespace.
     pub async fn get_origin(
         &self,
         namespace: &str,
@@ -28,6 +31,7 @@ impl Api {
     }
 }
 
+/// Periodically refreshes the origin registration in moq-api.
 pub struct Refresh {
     client: moq_api::Client,
     origin: moq_api::Origin,
@@ -37,6 +41,7 @@ pub struct Refresh {
 
 impl Refresh {
     fn new(client: moq_api::Client, origin: moq_api::Origin, namespace: String) -> Self {
+        // Refresh every 5 minutes
         let duration = tokio::time::Duration::from_secs(300);
         let mut refresh = tokio::time::interval(tokio::time::Duration::from_secs(300));
         refresh.reset_after(duration); // skip the first tick
@@ -49,18 +54,20 @@ impl Refresh {
         }
     }
 
+    /// Update the origin registration in moq-api.
     async fn update(&self) -> Result<(), moq_api::ApiError> {
-        // Register the origin in moq-api.
         log::debug!(
             "registering origin: namespace={} url={}",
             self.namespace,
             self.origin.url
         );
+        // Register the origin in moq-api.
         self.client
             .set_origin(&self.namespace, self.origin.clone())
             .await
     }
 
+    /// Run the refresher loop.
     pub async fn run(&mut self) -> anyhow::Result<()> {
         loop {
             self.refresh.tick().await;
@@ -69,6 +76,7 @@ impl Refresh {
     }
 }
 
+/// Unregister the origin on drop.
 impl Drop for Refresh {
     fn drop(&mut self) {
         // TODO this is really lazy