moq-net: reshape Track into TrackInfo + an async TrackConsumer handle (#1631)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: kixelated

rs/hang/examples/subscribe.rs

@@ -52,8 +52,8 @@ async fn run_subscribe(consumer: moq_net::OriginConsumer) -> anyhow::Result<()>
 
 	// Read the catalog to discover available tracks.
 	let catalog_track = broadcast
-		.consume_track(hang::Catalog::DEFAULT_NAME)
-		.subscribe(hang::Catalog::default_subscription())
+		.track(hang::Catalog::DEFAULT_NAME)?
+		.subscribe(hang::Catalog::default_subscription())?
 		.await?;
 	let mut catalog = moq_mux::catalog::hang::Consumer::new(catalog_track);
 
@@ -77,11 +77,11 @@ async fn run_subscribe(consumer: moq_net::OriginConsumer) -> anyhow::Result<()>
 
 	// Subscribe to the video track.
 	let track_consumer = broadcast
-		.consume_track(name)
+		.track(name)?
 		.subscribe(moq_net::Subscription {
 			priority: 1,
 			..Default::default()
-		})
+		})?
 		.await?;
 	let mut ordered = moq_mux::container::Consumer::new(track_consumer, moq_mux::catalog::hang::Container::Legacy)
 		.with_latency(Duration::from_millis(500));

rs/hang/examples/video.rs

@@ -42,7 +42,7 @@ async fn run_session(origin: moq_net::OriginProducer) -> anyhow::Result<()> {
 // The catalog can contain multiple tracks, used by the viewer to choose the best track.
 fn create_track(broadcast: &mut moq_net::BroadcastProducer) -> anyhow::Result<moq_net::TrackProducer> {
 	// Basic information about the video track.
-	let video_track = moq_net::Track::new("video");
+	let video_track = "video";
 
 	// Example video configuration
 	// In a real application, you would get this from the encoder
@@ -61,7 +61,7 @@ fn create_track(broadcast: &mut moq_net::BroadcastProducer) -> anyhow::Result<mo
 	// Create a map of video renditions
 	// Multiple renditions allow the viewer to choose based on their capabilities
 	let mut renditions = std::collections::BTreeMap::new();
-	renditions.insert(video_track.name.clone(), video_config);
+	renditions.insert(video_track.to_string(), video_config);
 
 	// Create the catalog describing our video track.
 	let catalog = hang::catalog::Catalog {
@@ -75,21 +75,21 @@ fn create_track(broadcast: &mut moq_net::BroadcastProducer) -> anyhow::Result<mo
 	};
 
 	// Publish the catalog as a "catalog.json" track in the broadcast.
-	let mut catalog_track = broadcast.create_track(hang::Catalog::default_track())?;
+	let mut catalog_track = broadcast.create_track(hang::Catalog::DEFAULT_NAME, hang::Catalog::default_track_info())?;
 	let mut group = catalog_track.append_group()?;
 	group.write_frame(catalog.to_string()?)?;
 	group.finish()?;
 
 	// Actually create the media track now.
-	let track = broadcast.create_track(video_track)?;
+	let track = broadcast.create_track(video_track, None)?;
 
 	Ok(track)
 }
 
 // Produce a broadcast and publish it to the origin.
 async fn run_broadcast(origin: moq_net::OriginProducer) -> anyhow::Result<()> {
 	// Create and publish a broadcast to the origin.
-	let mut broadcast = moq_net::Broadcast::new().produce();
+	let mut broadcast = moq_net::BroadcastInfo::new().produce();
 	let track = create_track(&mut broadcast)?;
 
 	// NOTE: The path is empty because we're using the URL to scope the broadcast.

rs/hang/src/catalog/root.rs

@@ -64,9 +64,12 @@ impl Catalog {
 		Ok(serde_json::to_writer(writer, self)?)
 	}
 
-	pub fn default_track() -> moq_net::Track {
-		// The catalog is JSON and re-sent on every change, so it pays to compress.
-		moq_net::Track::new(Catalog::DEFAULT_NAME).with_compress(true)
+	/// Track properties for creating the catalog track via
+	/// [`create_track`](moq_net::BroadcastProducer::create_track) at
+	/// [`DEFAULT_NAME`](Self::DEFAULT_NAME). The catalog is JSON and re-sent on
+	/// every change, so it pays to compress.
+	pub fn default_track_info() -> moq_net::TrackInfo {
+		moq_net::TrackInfo::default().with_compress(true)
 	}
 
 	/// The subscription preferences used for the catalog track (high priority so

rs/kio/src/consumer.rs

@@ -4,7 +4,7 @@ use std::{
 };
 
 use crate::{
-	Counts, State, Weak,
+	Counts, Mut, State, Weak,
 	lock::*,
 	producer::{Producer, Ref},
 	waiter::*,
@@ -106,6 +106,24 @@ impl<T> Consumer<T> {
 		}
 	}
 
+	/// Acquire mutable access to the shared state.
+	///
+	/// Returns `Ok(Mut)` if the channel is open, or `Err(Ref)` with
+	/// read-only access if closed. Only locks once.
+	pub fn write(&self) -> Result<Mut<'_, T>, Ref<'_, T>> {
+		let state = self.state.lock();
+		if state.closed {
+			Err(Ref { state })
+		} else {
+			Ok(Mut::new(state))
+		}
+	}
+
+	/// Returns `true` if the channel has been closed by the producer.
+	pub fn is_closed(&self) -> bool {
+		self.state.lock().closed
+	}
+
 	/// Returns `true` if both consumers share the same underlying state.
 	pub fn same_channel(&self, other: &Self) -> bool {
 		self.state.is_clone(&other.state)

rs/libmoq/src/consume.rs

@@ -69,8 +69,8 @@ impl Consume {
 		tokio::spawn(async move {
 			let res = async move {
 				let catalog = broadcast
-					.consume_track(hang::catalog::Catalog::DEFAULT_NAME)
-					.subscribe(hang::catalog::Catalog::default_subscription())
+					.track(hang::catalog::Catalog::DEFAULT_NAME)?
+					.subscribe(hang::catalog::Catalog::default_subscription())?
 					.await?;
 				Self::run_catalog(on_catalog, broadcast.clone(), catalog.into(), channel.1).await
 			}
@@ -250,11 +250,11 @@ impl Consume {
 		tokio::spawn(async move {
 			let res = async move {
 				let track = broadcast
-					.consume_track(&name)
+					.track(&name)?
 					.subscribe(moq_net::Subscription {
 						priority: 1,
 						..Default::default()
-					})
+					})?
 					.await?;
 				let track = moq_mux::container::Consumer::new(track, moq_mux::catalog::hang::Container::Legacy)
 					.with_latency(latency);
@@ -302,11 +302,11 @@ impl Consume {
 		tokio::spawn(async move {
 			let res = async move {
 				let track = broadcast
-					.consume_track(&name)
+					.track(&name)?
 					.subscribe(moq_net::Subscription {
 						priority: 2,
 						..Default::default()
-					})
+					})?
 					.await?;
 				let track = moq_mux::container::Consumer::new(track, moq_mux::catalog::hang::Container::Legacy)
 					.with_latency(latency);
@@ -407,7 +407,7 @@ impl Consume {
 		// `subscribe` blocks on SUBSCRIBE_OK, so run it inside the task.
 		tokio::spawn(async move {
 			let res = async move {
-				let track = broadcast.consume_track(&name).subscribe(None).await?;
+				let track = broadcast.track(&name)?.subscribe(None)?.await?;
 				Self::run_raw(on_frame, track, channel.1).await
 			}
 			.await;

rs/libmoq/src/publish.rs

@@ -22,7 +22,7 @@ pub struct Publish {
 
 impl Publish {
 	pub fn create(&mut self) -> Result<Id, Error> {
-		let mut broadcast = moq_net::Broadcast::new().produce();
+		let mut broadcast = moq_net::BroadcastInfo::new().produce();
 		let catalog = moq_mux::catalog::hang::Producer::new(&mut broadcast)?;
 
 		let id = self.broadcasts.insert((broadcast, catalog))?;
@@ -128,7 +128,7 @@ impl Publish {
 	/// if you want to describe the track in the catalog as well.
 	pub fn track(&mut self, broadcast: Id, name: &str) -> Result<Id, Error> {
 		let (broadcast, _) = self.broadcasts.get_mut(broadcast).ok_or(Error::BroadcastNotFound)?;
-		let track = broadcast.create_track(moq_net::Track::new(name))?;
+		let track = broadcast.create_track(name, None)?;
 		self.tracks.insert(track)
 	}
 

rs/moq-audio/src/consumer.rs

@@ -53,7 +53,7 @@ impl AudioConsumer {
 		};
 
 		let name = name.into();
-		let track = broadcast.consume_track(&name).subscribe(None).await?;
+		let track = broadcast.track(&name)?.subscribe(None)?.await?;
 		let mut track = moq_mux::container::Consumer::new(track, moq_mux::container::legacy::Wire);
 		if let Some(latency) = output.latency_max {
 			track = track.with_latency(latency);

rs/moq-audio/src/producer.rs

@@ -61,8 +61,10 @@ impl AudioProducer {
 		// Audio hang frames carry microsecond timestamps; advertise that on the
 		// track so Lite05 subscribers know what scale to expect and the model
 		// layer accepts Frame::timestamp on append.
-		let track =
-			broadcast.create_track(moq_net::Track::new(name.clone()).with_timescale(hang::container::TIMESCALE))?;
+		let track = broadcast.create_track(
+			name.clone(),
+			moq_net::TrackInfo::default().with_timescale(hang::container::TIMESCALE),
+		)?;
 		let track = moq_mux::container::Producer::new(track, moq_mux::container::legacy::Wire);
 
 		let mut catalog_mut = catalog.clone();

rs/moq-audio/tests/roundtrip.rs

@@ -30,7 +30,7 @@ fn f32_bytes(samples: &[f32]) -> Bytes {
 
 #[tokio::test]
 async fn opus_round_trip_48k_stereo() {
-	let mut broadcast = moq_net::Broadcast::new().produce();
+	let mut broadcast = moq_net::BroadcastInfo::new().produce();
 	let catalog = moq_mux::catalog::hang::Producer::new(&mut broadcast).unwrap();
 	let mut catalog_consumer = catalog.consume().unwrap();
 	let broadcast_consumer = broadcast.consume();
@@ -110,7 +110,7 @@ async fn opus_round_trip_48k_stereo() {
 
 #[tokio::test]
 async fn opus_round_trip_44100_s16_resampled() {
-	let mut broadcast = moq_net::Broadcast::new().produce();
+	let mut broadcast = moq_net::BroadcastInfo::new().produce();
 	let catalog = moq_mux::catalog::hang::Producer::new(&mut broadcast).unwrap();
 	let mut catalog_consumer = catalog.consume().unwrap();
 	let broadcast_consumer = broadcast.consume();

rs/moq-boy/src/input.rs

@@ -99,7 +99,7 @@ async fn handle_viewer_commands(
 	broadcast: moq_net::BroadcastConsumer,
 	cmd_tx: &tokio::sync::mpsc::Sender<Command>,
 ) -> anyhow::Result<()> {
-	let mut track = broadcast.consume_track("command").subscribe(None).await?;
+	let mut track = broadcast.track("command")?.subscribe(None)?.await?;
 
 	while let Some(mut group) = track.recv_group().await? {
 		while let Some(frame) = group.read_frame().await? {