moq-net: accept impl Into<Option<Subscription>> for subscribe, drop subscribe_default

rs/libmoq/src/consume.rs

@@ -407,10 +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(moq_net::Subscription::default())
-					.await?;
+				let track = broadcast.consume_track(&name).subscribe(None).await?;
 				Self::run_raw(on_frame, track, channel.1).await
 			}
 			.await;

rs/moq-audio/src/consumer.rs

@@ -53,10 +53,7 @@ impl AudioConsumer {
 		};
 
 		let name = name.into();
-		let track = broadcast
-			.consume_track(&name)
-			.subscribe(moq_net::Subscription::default())
-			.await?;
+		let track = broadcast.consume_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-boy/src/input.rs

@@ -99,10 +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(moq_net::Subscription::default())
-		.await?;
+	let mut track = broadcast.consume_track("command").subscribe(None).await?;
 
 	while let Some(mut group) = track.recv_group().await? {
 		while let Some(frame) = group.read_frame().await? {

rs/moq-ffi/src/consumer.rs

@@ -97,11 +97,7 @@ impl MoqBroadcastConsumer {
 	///
 	/// Frames are returned as plain byte payloads with no codec or container parsing.
 	pub async fn subscribe_track(&self, name: String) -> Result<Arc<MoqTrackConsumer>, MoqError> {
-		let track = self
-			.inner
-			.consume_track(&name)
-			.subscribe(moq_net::Subscription::default())
-			.await?;
+		let track = self.inner.consume_track(&name).subscribe(None).await?;
 		Ok(Arc::new(MoqTrackConsumer::new(track)))
 	}
 
@@ -121,11 +117,7 @@ impl MoqBroadcastConsumer {
 		let media: moq_mux::catalog::hang::Container = (&container)
 			.try_into()
 			.map_err(|e| MoqError::Codec(format!("invalid container: {e}")))?;
-		let track = self
-			.inner
-			.consume_track(&name)
-			.subscribe(moq_net::Subscription::default())
-			.await?;
+		let track = self.inner.consume_track(&name).subscribe(None).await?;
 		let latency = std::time::Duration::from_millis(max_latency_ms);
 		let consumer = moq_mux::container::Consumer::new(track, media).with_latency(latency);
 		Ok(Arc::new(MoqMediaConsumer {

rs/moq-ffi/src/producer.rs

@@ -203,7 +203,7 @@ impl MoqTrackProducer {
 		let _guard = crate::ffi::RUNTIME.enter();
 		let guard = self.inner.lock().unwrap();
 		let track = guard.as_ref().ok_or_else(|| MoqError::Closed)?;
-		Ok(Arc::new(MoqTrackConsumer::new(track.subscribe_default())))
+		Ok(Arc::new(MoqTrackConsumer::new(track.subscribe(None))))
 	}
 
 	/// Append a new group to the track, returning a producer for writing frames into it.

rs/moq-gst/src/source/imp.rs

@@ -448,10 +448,7 @@ async fn run_session(
 		};
 		let caps = video_caps(&config)?;
 		let endpoint = request_pad(&control_tx, descriptor.clone(), caps).await?;
-		let track_consumer = broadcast
-			.consume_track(&track_name)
-			.subscribe(moq_net::Subscription::default())
-			.await?;
+		let track_consumer = broadcast.consume_track(&track_name).subscribe(None).await?;
 		let track = moq_mux::container::Consumer::new(track_consumer, moq_mux::catalog::hang::Container::Legacy)
 			.with_latency(Duration::from_secs(1));
 		tasks.push(spawn_track_pump(track, descriptor, endpoint, shutdown.clone()));
@@ -464,10 +461,7 @@ async fn run_session(
 		};
 		let caps = audio_caps(&config)?;
 		let endpoint = request_pad(&control_tx, descriptor.clone(), caps).await?;
-		let track_consumer = broadcast
-			.consume_track(&track_name)
-			.subscribe(moq_net::Subscription::default())
-			.await?;
+		let track_consumer = broadcast.consume_track(&track_name).subscribe(None).await?;
 		let track = moq_mux::container::Consumer::new(track_consumer, moq_mux::catalog::hang::Container::Legacy)
 			.with_latency(Duration::from_secs(1));
 		tasks.push(spawn_track_pump(track, descriptor, endpoint, shutdown.clone()));

rs/moq-mux/src/catalog/consumer.rs

@@ -32,10 +32,7 @@ impl Consumer {
 				Self::Hang(super::hang::Consumer::new(track))
 			}
 			CatalogFormat::Msf => {
-				let track = broadcast
-					.consume_track(moq_msf::DEFAULT_NAME)
-					.subscribe(moq_net::Subscription::default())
-					.await?;
+				let track = broadcast.consume_track(moq_msf::DEFAULT_NAME).subscribe(None).await?;
 				Self::Msf(super::msf::Consumer::new(track))
 			}
 		})

rs/moq-mux/src/catalog/hang/consumer.rs

@@ -94,7 +94,7 @@ mod test {
 	#[test]
 	fn waits_for_pending_catalog_group_payload() {
 		let mut track = Catalog::default_track().produce();
-		let mut consumer = Consumer::new(track.subscribe_default());
+		let mut consumer = Consumer::new(track.subscribe(None));
 		let mut group = track.append_group().expect("catalog group should append");
 
 		let waiter = kio::Waiter::noop();
@@ -110,7 +110,7 @@ mod test {
 	#[test]
 	fn waits_for_pending_catalog_group_payload_after_track_finish() {
 		let mut track = Catalog::default_track().produce();
-		let mut consumer = Consumer::new(track.subscribe_default());
+		let mut consumer = Consumer::new(track.subscribe(None));
 		let mut group = track.append_group().expect("catalog group should append");
 
 		track.finish().expect("catalog track should finish");
@@ -128,7 +128,7 @@ mod test {
 	#[test]
 	fn returns_latest_complete_catalog_group() {
 		let mut track = Catalog::default_track().produce();
-		let mut consumer = Consumer::new(track.subscribe_default());
+		let mut consumer = Consumer::new(track.subscribe(None));
 		let waiter = kio::Waiter::noop();
 
 		let (_old, old_payload) = catalog_payload("old");
@@ -154,7 +154,7 @@ mod test {
 	#[test]
 	fn waits_for_newer_pending_group_instead_of_returning_older_ready_group() {
 		let mut track = Catalog::default_track().produce();
-		let mut consumer = Consumer::new(track.subscribe_default());
+		let mut consumer = Consumer::new(track.subscribe(None));
 		let waiter = kio::Waiter::noop();
 
 		let (_old, old_payload) = catalog_payload("old");
@@ -181,7 +181,7 @@ mod test {
 	#[test]
 	fn retained_pending_group_is_superseded_by_newer_group() {
 		let mut track = Catalog::default_track().produce();
-		let mut consumer = Consumer::new(track.subscribe_default());
+		let mut consumer = Consumer::new(track.subscribe(None));
 		let waiter = kio::Waiter::noop();
 
 		let (_old, old_payload) = catalog_payload("old");
@@ -211,7 +211,7 @@ mod test {
 	#[test]
 	fn returns_none_when_empty_track_finishes() {
 		let mut track = Catalog::default_track().produce();
-		let mut consumer = Consumer::new(track.subscribe_default());
+		let mut consumer = Consumer::new(track.subscribe(None));
 		let waiter = kio::Waiter::noop();
 
 		track.finish().expect("catalog track should finish");

rs/moq-mux/src/catalog/hang/producer.rs

@@ -54,7 +54,7 @@ impl Producer {
 
 	/// Create a consumer for this catalog, receiving updates as they're published.
 	pub fn consume(&self) -> Result<super::Consumer, moq_net::Error> {
-		let track = self.hang_track.subscribe_default();
+		let track = self.hang_track.subscribe(None);
 		let subscriber = track;
 		Ok(super::Consumer::new(subscriber))
 	}