libmoq consume API (#777)

Co-authored-by: Dave Gullo <[email protected]>

Author: kixelated

Cargo.lock

@@ -76,7 +76,7 @@ fn create_track(broadcast: &mut moq_lite::BroadcastProducer) -> hang::TrackProdu
 
 	// Create a map of video renditions
 	// Multiple renditions allow the viewer to choose based on their capabilities
-	let mut renditions = std::collections::HashMap::new();
+	let mut renditions = std::collections::BTreeMap::new();
 	renditions.insert(video_track.name.clone(), video_config);
 
 	// Create the video catalog entry with the renditions

deno.lock

@@ -4,7 +4,7 @@ mod codec;
 pub use aac::*;
 pub use codec::*;
 
-use std::collections::HashMap;
+use std::collections::BTreeMap;
 
 use bytes::Bytes;
 
@@ -21,7 +21,8 @@ use serde_with::{hex::Hex, DisplayFromStr};
 pub struct Audio {
 	/// A map of track name to rendition configuration.
 	/// This is not an array so it will work with JSON Merge Patch.
-	pub renditions: HashMap<String, AudioConfig>,
+	/// We use a BTreeMap so keys are sorted alphabetically for *some* deterministic behavior.
+	pub renditions: BTreeMap<String, AudioConfig>,
 
 	/// The priority of the audio track, relative to other tracks in the broadcast.
 	pub priority: u8,

rs/hang/examples/video.rs

@@ -7,7 +7,7 @@ use serde::{Deserialize, Serialize};
 
 use crate::catalog::{Audio, AudioConfig, Chat, Track, User, Video, VideoConfig};
 use crate::Result;
-use moq_lite::Produce;
+use moq_lite::Pair;
 
 /// A catalog track, created by a broadcaster to describe the tracks available in a broadcast.
 #[serde_with::serde_as]
@@ -83,10 +83,10 @@ impl Catalog {
 	}
 
 	/// Produce a catalog track that describes the available media tracks.
-	pub fn produce(self) -> Produce<CatalogProducer, CatalogConsumer> {
+	pub fn produce(self) -> Pair<CatalogProducer, CatalogConsumer> {
 		let track = Catalog::default_track().produce();
 
-		Produce {
+		Pair {
 			producer: CatalogProducer::new(track.producer, self),
 			consumer: track.consumer.into(),
 		}
@@ -267,14 +267,14 @@ impl From<moq_lite::TrackConsumer> for CatalogConsumer {
 
 #[cfg(test)]
 mod test {
+	use std::collections::BTreeMap;
+
 	use crate::catalog::{AudioCodec::Opus, AudioConfig, VideoConfig, H264};
 
 	use super::*;
 
 	#[test]
 	fn simple() {
-		use std::collections::HashMap;
-
 		let mut encoded = r#"{
 			"video": {
 				"renditions": {
@@ -304,7 +304,7 @@ mod test {
 
 		encoded.retain(|c| !c.is_whitespace());
 
-		let mut video_renditions = HashMap::new();
+		let mut video_renditions = BTreeMap::new();
 		video_renditions.insert(
 			"video".to_string(),
 			VideoConfig {
@@ -326,7 +326,7 @@ mod test {
 			},
 		);
 
-		let mut audio_renditions = HashMap::new();
+		let mut audio_renditions = BTreeMap::new();
 		audio_renditions.insert(
 			"audio".to_string(),
 			AudioConfig {

rs/hang/src/catalog/audio/mod.rs

@@ -10,7 +10,7 @@ pub use h264::*;
 pub use h265::*;
 pub use vp9::*;
 
-use std::collections::HashMap;
+use std::collections::BTreeMap;
 
 use bytes::Bytes;
 use serde::{Deserialize, Serialize};
@@ -27,7 +27,8 @@ use serde_with::{hex::Hex, DisplayFromStr};
 pub struct Video {
 	/// A map of track name to rendition configuration.
 	/// This is not an array in order for it to work with JSON Merge Patch.
-	pub renditions: HashMap<String, VideoConfig>,
+	/// We use a BTreeMap so keys are sorted alphabetically for *some* deterministic behavior.
+	pub renditions: BTreeMap<String, VideoConfig>,
 
 	/// The priority of the video track, relative to other tracks in the broadcast.
 	pub priority: u8,

rs/hang/src/catalog/root.rs

@@ -3,7 +3,10 @@ use std::{
 	sync::{atomic, Arc},
 };
 
-use crate::catalog::{Catalog, CatalogProducer};
+use crate::{
+	catalog::{Catalog, CatalogConsumer, CatalogProducer},
+	TrackConsumer,
+};
 
 #[derive(Clone)]
 pub struct BroadcastProducer {
@@ -32,6 +35,12 @@ impl BroadcastProducer {
 	}
 }
 
+impl Default for BroadcastProducer {
+	fn default() -> Self {
+		Self::new(moq_lite::BroadcastProducer::default())
+	}
+}
+
 impl Deref for BroadcastProducer {
 	type Target = moq_lite::BroadcastProducer;
 
@@ -58,4 +67,33 @@ impl From<BroadcastProducer> for moq_lite::BroadcastProducer {
 	}
 }
 
-// TODO BroadcastConsumer
+#[derive(Clone)]
+pub struct BroadcastConsumer {
+	pub inner: moq_lite::BroadcastConsumer,
+	pub catalog: CatalogConsumer,
+}
+
+impl BroadcastConsumer {
+	pub fn new(inner: moq_lite::BroadcastConsumer) -> Self {
+		let catalog = inner.subscribe_track(&Catalog::default_track()).into();
+		Self { inner, catalog }
+	}
+
+	pub fn subscribe(&self, track: &moq_lite::Track, latency: std::time::Duration) -> TrackConsumer {
+		TrackConsumer::new(self.inner.subscribe_track(track), latency)
+	}
+}
+
+impl Deref for BroadcastConsumer {
+	type Target = moq_lite::BroadcastConsumer;
+
+	fn deref(&self) -> &Self::Target {
+		&self.inner
+	}
+}
+
+impl From<moq_lite::BroadcastConsumer> for BroadcastConsumer {
+	fn from(inner: moq_lite::BroadcastConsumer) -> Self {
+		Self::new(inner)
+	}
+}

rs/hang/src/catalog/video/mod.rs

@@ -1,8 +1,9 @@
-use buf_list::BufList;
 use derive_more::Debug;
 
 use crate::Timestamp;
 
+pub use buf_list::BufList;
+
 /// A media frame with a timestamp and codec-specific payload.
 ///
 /// Frames are the fundamental unit of media data in hang. Each frame contains:

rs/hang/src/model/broadcast.rs

@@ -91,8 +91,8 @@ impl TrackProducer {
 	///
 	/// Multiple consumers can be created from the same producer, each receiving
 	/// a copy of all data written to the track.
-	pub fn consume(&self) -> TrackConsumer {
-		TrackConsumer::new(self.inner.consume())
+	pub fn consume(&self, max_latency: std::time::Duration) -> TrackConsumer {
+		TrackConsumer::new(self.inner.consume(), max_latency)
 	}
 }
 
@@ -132,18 +132,18 @@ pub struct TrackConsumer {
 	max_timestamp: Timestamp,
 
 	// The maximum buffer size before skipping a group.
-	latency: std::time::Duration,
+	max_latency: std::time::Duration,
 }
 
 impl TrackConsumer {
 	/// Create a new TrackConsumer wrapping the given moq-lite consumer.
-	pub fn new(inner: moq_lite::TrackConsumer) -> Self {
+	pub fn new(inner: moq_lite::TrackConsumer, max_latency: std::time::Duration) -> Self {
 		Self {
 			inner,
 			current: None,
 			pending: VecDeque::new(),
 			max_timestamp: Timestamp::default(),
-			latency: std::time::Duration::ZERO,
+			max_latency,
 		}
 	}
 
@@ -155,7 +155,7 @@ impl TrackConsumer {
 	///
 	/// Returns `None` when the track has ended.
 	pub async fn read_frame(&mut self) -> Result<Option<Frame>, Error> {
-		let latency = self.latency.try_into()?;
+		let latency = self.max_latency.try_into()?;
 		loop {
 			let cutoff = self
 				.max_timestamp
@@ -209,7 +209,7 @@ impl TrackConsumer {
 				},
 				Some((index, timestamp)) = buffering.next() => {
 					if self.current.is_some() {
-						tracing::debug!(old = ?self.max_timestamp, new = ?timestamp, buffer = ?self.latency, "skipping slow group");
+						tracing::debug!(old = ?self.max_timestamp, new = ?timestamp, buffer = ?self.max_latency, "skipping slow group");
 					}
 
 					drop(buffering);
@@ -228,9 +228,9 @@ impl TrackConsumer {
 
 	/// Set the maximum latency tolerance for this consumer.
 	///
-	/// Groups with timestamps older than `max_timestamp - latency` will be skipped.
-	pub fn set_latency(&mut self, max: std::time::Duration) {
-		self.latency = max;
+	/// Groups with timestamps older than `max_timestamp - max_latency` will be skipped.
+	pub fn set_max_latency(&mut self, max: std::time::Duration) {
+		self.max_latency = max;
 	}
 
 	/// Wait until the track is closed.
@@ -239,12 +239,6 @@ impl TrackConsumer {
 	}
 }
 
-impl From<moq_lite::TrackConsumer> for TrackConsumer {
-	fn from(inner: moq_lite::TrackConsumer) -> Self {
-		Self::new(inner)
-	}
-}
-
 impl From<TrackConsumer> for moq_lite::TrackConsumer {
 	fn from(inner: TrackConsumer) -> Self {
 		inner.inner

rs/hang/src/model/frame.rs

@@ -44,12 +44,17 @@ endif()
 file(MAKE_DIRECTORY ${RUST_DIR})
 file(MAKE_DIRECTORY ${RUST_TARGET_DIR}/include)
 
-# Create imported library target
-add_library(moq STATIC IMPORTED GLOBAL)
-set_target_properties(moq PROPERTIES
-    IMPORTED_LOCATION ${RUST_LIB}
-    INTERFACE_INCLUDE_DIRECTORIES ${RUST_TARGET_DIR}/include
-)
+# Create interface library target that wraps the Rust static library
+add_library(moq INTERFACE)
+target_include_directories(moq INTERFACE ${RUST_TARGET_DIR}/include)
+
+# Force absolute path to be used in linker command
+target_link_options(moq INTERFACE "$<LINK_ONLY:${RUST_LIB}>")
+
+# Link required system frameworks on macOS
+if(APPLE)
+    target_link_options(moq INTERFACE "LINKER:-framework,CoreFoundation" "LINKER:-framework,Security")
+endif()
 
 if(BUILD_RUST_LIB)
     add_dependencies(moq rust_build)
@@ -58,6 +63,7 @@ endif()
 if(PROJECT_IS_TOP_LEVEL)
     include(GNUInstallDirs)
 
+    # Install header and library files
     install(FILES ${RUST_HEADER}
         DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}
     )
@@ -66,25 +72,14 @@ if(PROJECT_IS_TOP_LEVEL)
         DESTINATION ${CMAKE_INSTALL_LIBDIR}
     )
 
-    install(TARGETS moq
-        EXPORT moq-targets
-        LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR}
-        ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR}
-        RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR}
-    )
-
-    install(EXPORT moq-targets
-        FILE moq-targets.cmake
-        NAMESPACE moq::
-        DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/moq
-    )
-
+    # Generate and install CMake package config files
     include(CMakePackageConfigHelpers)
 
     configure_package_config_file(
         ${CMAKE_CURRENT_SOURCE_DIR}/cmake/moq-config.cmake.in
         ${CMAKE_CURRENT_BINARY_DIR}/moq-config.cmake
         INSTALL_DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/moq
+        PATH_VARS CMAKE_INSTALL_LIBDIR CMAKE_INSTALL_INCLUDEDIR
     )
 
     write_basic_package_version_file(