Merge pull request #106 from cloudflare/me/nested-namespace-support

Add support for nested namespaces

Author: englishm-cloudflare

moq-relay-ietf/src/local.rs

@@ -46,9 +46,32 @@ impl Locals {
         Ok(registration)
     }
 
-    /// Lookup local tracks by namespace.
+    /// Lookup local tracks by namespace using hierarchical prefix matching.
+    /// Returns the TracksReader for the longest matching namespace prefix.
     pub fn route(&self, namespace: &TrackNamespace) -> Option<TracksReader> {
-        self.lookup.lock().unwrap().get(namespace).cloned()
+        let lookup = self.lookup.lock().unwrap();
+
+        // Find the longest matching prefix
+        let mut best_match: Option<TracksReader> = None;
+        let mut best_len = 0;
+
+        for (registered_ns, tracks) in lookup.iter() {
+            // Check if registered_ns is a prefix of namespace
+            if namespace.fields.len() >= registered_ns.fields.len() {
+                let is_prefix = registered_ns
+                    .fields
+                    .iter()
+                    .zip(namespace.fields.iter())
+                    .all(|(a, b)| a == b);
+
+                if is_prefix && registered_ns.fields.len() > best_len {
+                    best_match = Some(tracks.clone());
+                    best_len = registered_ns.fields.len();
+                }
+            }
+        }
+
+        best_match
     }
 }
 

moq-relay-ietf/src/producer.rs

@@ -79,7 +79,10 @@ impl Producer {
     async fn serve_subscribe(self, subscribed: Subscribed) -> Result<(), anyhow::Error> {
         // Check local tracks first, and serve from local if possible
         if let Some(mut local) = self.locals.route(&subscribed.track_namespace) {
-            if let Some(track) = local.subscribe(&subscribed.track_name) {
+            // Pass the full requested namespace, not the announced prefix
+            if let Some(track) =
+                local.subscribe(subscribed.track_namespace.clone(), &subscribed.track_name)
+            {
                 log::info!("serving subscribe from local: {:?}", track.info);
                 return Ok(subscribed.serve(track).await?);
             }
@@ -118,9 +121,10 @@ impl Producer {
             .locals
             .route(&track_status_requested.request_msg.track_namespace)
         {
-            if let Some(track) =
-                local_tracks.get_track_reader(&track_status_requested.request_msg.track_name)
-            {
+            if let Some(track) = local_tracks.get_track_reader(
+                &track_status_requested.request_msg.track_namespace,
+                &track_status_requested.request_msg.track_name,
+            ) {
                 log::info!("serving track_status from local: {:?}", track.info);
                 return Ok(track_status_requested.respond_ok(&track)?);
             }

moq-sub/src/media.rs

@@ -107,7 +107,11 @@ impl<O: AsyncWrite + Send + Unpin + 'static> Media<O> {
                     });
                 });
 
-                tracks.push(self.broadcast.subscribe(&name).context("no track")?);
+                tracks.push(
+                    self.broadcast
+                        .subscribe(self.broadcast.namespace.clone(), &name)
+                        .context("no track")?,
+                );
             }
         }
 
@@ -145,7 +149,7 @@ impl<O: AsyncWrite + Send + Unpin + 'static> Media<O> {
 
         let track = self
             .broadcast
-            .subscribe(track_name)
+            .subscribe(self.broadcast.namespace.clone(), track_name)
             .context(format!("no {alias} track"))?;
         let mut group = match track.mode().await? {
             TrackReaderMode::Subgroups(mut groups) => {

moq-transport/src/serve/tracks.rs

@@ -16,6 +16,13 @@ use super::{ServeError, Track, TrackReader, TrackWriter};
 use crate::coding::TrackNamespace;
 use crate::watch::{Queue, State};
 
+/// Full track identifier: namespace + track name
+#[derive(Hash, Eq, PartialEq, Clone, Debug)]
+pub struct FullTrackName {
+    pub namespace: TrackNamespace,
+    pub name: String,
+}
+
 /// Static information about a broadcast.
 #[derive(Debug)]
 pub struct Tracks {
@@ -42,7 +49,7 @@ impl Tracks {
 
 #[derive(Default)]
 pub struct TracksState {
-    tracks: HashMap<String, TrackReader>,
+    tracks: HashMap<FullTrackName, TrackReader>,
 }
 
 /// Publish new tracks for a broadcast by name.
@@ -57,6 +64,7 @@ impl TracksWriter {
     }
 
     /// Create a new track with the given name, inserting it into the broadcast.
+    /// The track will use this writer's namespace.
     /// None is returned if all [TracksReader]s have been dropped.
     pub fn create(&mut self, track: &str) -> Option<TrackWriter> {
         let (writer, reader) = Track {
@@ -66,17 +74,22 @@ impl TracksWriter {
         .produce();
 
         // NOTE: We overwrite the track if it already exists.
-        self.state
-            .lock_mut()?
-            .tracks
-            .insert(track.to_owned(), reader);
+        let full_name = FullTrackName {
+            namespace: self.namespace.clone(),
+            name: track.to_owned(),
+        };
+        self.state.lock_mut()?.tracks.insert(full_name, reader);
 
         Some(writer)
     }
 
-    /// Remove a track from the broadcast by name.
-    pub fn remove(&mut self, track_name: &str) -> Option<TrackReader> {
-        self.state.lock_mut()?.tracks.remove(track_name)
+    /// Remove a track from the broadcast by full name.
+    pub fn remove(&mut self, namespace: &TrackNamespace, track_name: &str) -> Option<TrackReader> {
+        let full_name = FullTrackName {
+            namespace: namespace.clone(),
+            name: track_name.to_owned(),
+        };
+        self.state.lock_mut()?.tracks.remove(&full_name)
     }
 }
 
@@ -143,28 +156,46 @@ impl TracksReader {
         Self { state, queue, info }
     }
 
-    /// Get a track from the broadcast by name, if it exists.
-    pub fn get_track_reader(&mut self, track_name: &str) -> Option<TrackReader> {
+    /// Get a track from the broadcast by full name, if it exists.
+    pub fn get_track_reader(
+        &mut self,
+        namespace: &TrackNamespace,
+        track_name: &str,
+    ) -> Option<TrackReader> {
         let state = self.state.lock();
+        let full_name = FullTrackName {
+            namespace: namespace.clone(),
+            name: track_name.to_owned(),
+        };
 
-        if let Some(track_reader) = state.tracks.get(track_name) {
+        if let Some(track_reader) = state.tracks.get(&full_name) {
             return Some(track_reader.clone());
         }
         None
     }
 
-    /// Get or request a track from the broadcast by name.
+    /// Get or request a track from the broadcast by full name.
+    /// The namespace parameter should be the full requested namespace, not just the announced prefix.
     /// None is returned if [TracksWriter] or [TracksRequest] cannot fufill the request.
-    pub fn subscribe(&mut self, track_name: &str) -> Option<TrackReader> {
+    pub fn subscribe(
+        &mut self,
+        namespace: TrackNamespace,
+        track_name: &str,
+    ) -> Option<TrackReader> {
         let state = self.state.lock();
+        let full_name = FullTrackName {
+            namespace: namespace.clone(),
+            name: track_name.to_owned(),
+        };
 
-        if let Some(track_reader) = state.tracks.get(track_name) {
+        if let Some(track_reader) = state.tracks.get(&full_name) {
             return Some(track_reader.clone());
         }
 
         let mut state = state.into_mut()?;
+        // Use the full requested namespace, not self.namespace
         let track_writer_reader = Track {
-            namespace: self.namespace.clone(),
+            namespace: namespace.clone(),
             name: track_name.to_owned(),
         }
         .produce();
@@ -173,10 +204,10 @@ impl TracksReader {
             return None;
         }
 
-        // We requested the track sucessfully so we can deduplicate it.
+        // We requested the track sucessfully so we can deduplicate it by full name.
         state
             .tracks
-            .insert(track_name.to_owned(), track_writer_reader.1.clone());
+            .insert(full_name, track_writer_reader.1.clone());
 
         Some(track_writer_reader.1.clone())
     }

moq-transport/src/session/publisher.rs

@@ -167,7 +167,10 @@ impl Publisher {
         subscribed: Subscribed,
         mut tracks: TracksReader,
     ) -> Result<(), SessionError> {
-        if let Some(track) = tracks.subscribe(&subscribed.track_name) {
+        if let Some(track) = tracks.subscribe(
+            subscribed.info.track_namespace.clone(),
+            &subscribed.info.track_name,
+        ) {
             subscribed.serve(track).await?;
         } else {
             subscribed.close(ServeError::NotFound)?;
@@ -181,7 +184,10 @@ impl Publisher {
         mut tracks: TracksReader,
     ) -> Result<(), SessionError> {
         let track = tracks
-            .subscribe(&track_status_request.request_msg.track_name.clone())
+            .subscribe(
+                track_status_request.request_msg.track_namespace.clone(),
+                &track_status_request.request_msg.track_name,
+            )
             .ok_or(ServeError::NotFound)?;
 
         track_status_request.respond_ok(&track)?;