REANNOUNCE via duplicate ANNOUNCE; announced() returns (path, Announced)

rs/hang/examples/subscribe.rs

@@ -2,6 +2,8 @@
 
 use std::time::Duration;
 
+use anyhow::Context;
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
 	// Optional: Use moq_native to configure a logger.
@@ -40,13 +42,11 @@ async fn run_session(origin: moq_net::OriginProducer) -> anyhow::Result<()> {
 // Subscribe to a broadcast and read media frames.
 async fn run_subscribe(consumer: moq_net::OriginConsumer) -> anyhow::Result<()> {
 	// Wait for a broadcast to be announced.
-	let (path, broadcast) = consumer
-		.announced()
-		.next()
-		.await
-		.ok_or_else(|| anyhow::anyhow!("origin closed"))?;
+	let (path, broadcast) = consumer.announced().next().await.context("origin closed")?;
 
-	let broadcast = broadcast.ok_or_else(|| anyhow::anyhow!("broadcast unannounced: {path}"))?;
+	let broadcast = broadcast
+		.broadcast()
+		.with_context(|| format!("broadcast unannounced: {path}"))?;
 
 	tracing::info!(%path, "broadcast announced");
 

rs/libmoq/src/origin.rs

@@ -76,7 +76,7 @@ impl Origin {
 	) -> Result<(), Error> {
 		loop {
 			// `biased` so a pending close always wins over a ready announcement.
-			let (path, broadcast) = tokio::select! {
+			let (path, event) = tokio::select! {
 				biased;
 				_ = &mut close => return Ok(()),
 				next = consumer.next() => match next {
@@ -86,10 +86,11 @@ impl Origin {
 			};
 
 			// Hold the lock only to buffer the announcement; release it before the callback.
+			// Active and Restart both carry a broadcast; Ended does not.
 			let announced_id = State::lock()
 				.origin
 				.announced
-				.insert((path.to_string(), broadcast.is_some()))?;
+				.insert((path.to_string(), event.broadcast().is_some()))?;
 			callback.call(announced_id);
 		}
 	}

rs/moq-boy/src/input.rs

@@ -71,7 +71,7 @@ pub async fn handle_viewers(
 
 		let viewer_id = path.to_string();
 
-		if let Some(broadcast) = broadcast {
+		if let Some(broadcast) = broadcast.broadcast() {
 			tracing::info!(%viewer_id, "viewer connected");
 			let cmd_tx = cmd_tx.clone();
 			let vid = viewer_id.clone();

rs/moq-ffi/src/origin.rs

@@ -28,14 +28,16 @@ impl Announced {
 	async fn next(&mut self) -> Result<Option<Arc<MoqAnnouncement>>, MoqError> {
 		loop {
 			match self.inner.next().await {
-				Some((path, Some(broadcast))) => {
+				// Active and Restart both carry a broadcast; skip unannounce events.
+				Some((path, event)) => {
+					let Some(broadcast) = event.broadcast() else {
+						continue;
+					};
 					return Ok(Some(Arc::new(MoqAnnouncement {
 						path: path.to_string(),
 						broadcast: Arc::new(MoqBroadcastConsumer::new(broadcast)),
 					})));
 				}
-				// TODO moq-lite will change to not emit None (unannounce) events here.
-				Some((_path, None)) => continue,
 				None => return Ok(None),
 			}
 		}
@@ -44,11 +46,11 @@ impl Announced {
 	async fn available(&mut self) -> Result<Arc<MoqBroadcastConsumer>, MoqError> {
 		loop {
 			match self.inner.next().await {
-				Some((_path, Some(broadcast))) => {
-					return Ok(Arc::new(MoqBroadcastConsumer::new(broadcast)));
-				}
-				// TODO moq-lite will change to not emit None (unannounce) events here.
-				Some((_path, None)) => continue,
+				// Active and Restart both carry a broadcast; skip unannounce events.
+				Some((_path, event)) => match event.broadcast() {
+					Some(broadcast) => return Ok(Arc::new(MoqBroadcastConsumer::new(broadcast))),
+					None => continue,
+				},
 				None => return Err(MoqError::Closed),
 			}
 		}

rs/moq-native/examples/clock.rs

@@ -101,13 +101,13 @@ async fn main() -> anyhow::Result<()> {
 
 			loop {
 				tokio::select! {
-					Some(announce) = origin.next() => match announce {
-						(path, Some(broadcast)) => {
+					Some((path, event)) = origin.next() => match event.broadcast() {
+						Some(broadcast) => {
 							tracing::info!(broadcast = %path, "broadcast is online, subscribing to track");
 							let track = broadcast.subscribe_track(&track)?;
 							clock = Some(Subscriber::new(track));
 						}
-						(path, None) => {
+						None => {
 							tracing::warn!(broadcast = %path, "broadcast is offline, waiting...");
 						}
 					},

rs/moq-native/tests/backend.rs

@@ -67,7 +67,7 @@ async fn backend_test(scheme: &str, backend: moq_native::QuicBackend) {
 		.expect("origin closed");
 
 	assert_eq!(path.as_str(), "test");
-	let bc = bc.expect("expected announce, got unannounce");
+	let bc = bc.broadcast().expect("expected announce, got unannounce");
 
 	let mut track_sub = bc
 		.subscribe_track(&Track::new("video"))
@@ -219,7 +219,7 @@ async fn iroh_connect() {
 		.expect("origin closed");
 
 	assert_eq!(path.as_str(), "test");
-	let bc = bc.expect("expected announce, got unannounce");
+	let bc = bc.broadcast().expect("expected announce, got unannounce");
 
 	let mut track_sub = bc
 		.subscribe_track(&Track::new("video"))

rs/moq-native/tests/broadcast.rs

@@ -84,7 +84,7 @@ async fn broadcast_test(scheme: &str, client_version: Option<&str>, server_versi
 		.expect("origin closed");
 
 	assert_eq!(path.as_str(), "test");
-	let bc = bc.expect("expected announce, got unannounce");
+	let bc = bc.broadcast().expect("expected announce, got unannounce");
 
 	// Subscribe to the track.
 	let mut track_sub = bc
@@ -496,7 +496,7 @@ async fn broadcast_websocket() {
 		.expect("origin closed");
 
 	assert_eq!(path.as_str(), "test");
-	let bc = bc.expect("expected announce, got unannounce");
+	let bc = bc.broadcast().expect("expected announce, got unannounce");
 
 	// Subscribe to the track.
 	let mut track_sub = bc
@@ -603,7 +603,7 @@ async fn broadcast_websocket_fallback() {
 		.expect("origin closed");
 
 	assert_eq!(path.as_str(), "test");
-	let bc = bc.expect("expected announce, got unannounce");
+	let bc = bc.broadcast().expect("expected announce, got unannounce");
 
 	// Subscribe to the track.
 	let mut track_sub = bc
@@ -844,7 +844,7 @@ async fn linger_resubscribe_keeps_flowing_moq_lite_03() {
 		.expect("announce timeout")
 		.expect("origin closed");
 	assert_eq!(path.as_str(), "test");
-	let bc = bc.expect("expected announce");
+	let bc = bc.broadcast().expect("expected announce");
 
 	// First subscription: receive group 0.
 	let mut sub1 = bc.subscribe_track(&Track::new("video")).expect("subscribe1");