moq_room/
room.rs

1//! A room is a path prefix. Participants are discovered from the announce stream.
2
3use std::task::{Poll, ready};
4
5use moq_net::{PathOwned, announce, broadcast, origin};
6
7use crate::path::{Kind, parse};
8
9/// One (un)announce of a participant broadcast.
10#[derive(Clone)]
11pub struct Event {
12	/// Participant identity (everything before `camera`/`screen`).
13	pub identity: PathOwned,
14	/// `camera` or `screen`.
15	pub kind: Kind,
16	/// Broadcast path relative to the room prefix.
17	pub path: PathOwned,
18	/// The live broadcast, or `None` when it went offline.
19	pub broadcast: Option<broadcast::Consumer>,
20}
21
22impl std::fmt::Debug for Event {
23	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
24		f.debug_struct("Event")
25			.field("identity", &self.identity)
26			.field("kind", &self.kind)
27			.field("path", &self.path)
28			.field("online", &self.broadcast.is_some())
29			.finish()
30	}
31}
32
33/// Runs the announce loop and yields remote participant broadcasts.
34///
35/// Skips paths that are not `{identity}/camera.hang` or `{identity}/screen.hang`, and
36/// skips the local identity so a publisher does not see itself as a remote.
37pub struct Room {
38	announced: announce::Consumer,
39	local: Option<PathOwned>,
40}
41
42impl std::fmt::Debug for Room {
43	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
44		f.debug_struct("Room")
45			.field("local", &self.local)
46			.finish_non_exhaustive()
47	}
48}
49
50impl Room {
51	/// Watch announcements on `origin`, skipping `local` when set.
52	pub fn new(origin: &origin::Consumer, local: Option<PathOwned>) -> Self {
53		Self {
54			announced: origin.announced(),
55			local,
56		}
57	}
58
59	/// Poll for the next remote camera/screen (un)announce.
60	pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll<Option<Event>> {
61		loop {
62			let Some(update) = ready!(self.announced.poll_next(waiter)) else {
63				return Poll::Ready(None);
64			};
65			let Some(parsed) = parse(&update.path) else {
66				continue;
67			};
68			if self.local.as_ref().is_some_and(|id| *id == parsed.identity) {
69				continue;
70			}
71			return Poll::Ready(Some(Event {
72				identity: parsed.identity,
73				kind: parsed.kind,
74				path: update.path,
75				broadcast: update.broadcast,
76			}));
77		}
78	}
79
80	/// Wait for the next remote camera/screen (un)announce.
81	pub async fn next(&mut self) -> Option<Event> {
82		kio::wait(|waiter| self.poll_next(waiter)).await
83	}
84}
85
86#[cfg(test)]
87mod tests {
88	use super::*;
89	use moq_net::{Origin, Path, broadcast::Route};
90
91	#[tokio::test]
92	async fn yields_camera_and_skips_local_and_unknown() {
93		let origin = Origin::random().produce();
94		let local = Path::new("alice").to_owned();
95		let mut room = Room::new(&origin.consume(), Some(local));
96
97		let _alice = origin
98			.create_broadcast("alice/camera", Route::announced())
99			.expect("alice camera");
100		let _bob = origin
101			.create_broadcast("bob/camera", Route::announced())
102			.expect("bob camera");
103		let _noise = origin
104			.create_broadcast("bob/chat", Route::announced())
105			.expect("unknown kind");
106
107		let event = room.next().await.expect("bob camera");
108		assert_eq!(event.identity.as_str(), "bob");
109		assert_eq!(event.kind, Kind::Camera);
110		assert!(event.broadcast.is_some());
111
112		drop(_bob);
113		let gone = room.next().await.expect("bob unannounce");
114		assert_eq!(gone.identity.as_str(), "bob");
115		assert_eq!(gone.kind, Kind::Camera);
116		assert!(gone.broadcast.is_none());
117	}
118
119	#[tokio::test]
120	async fn yields_screen() {
121		let origin = Origin::random().produce();
122		let mut room = Room::new(&origin.consume(), None);
123
124		let _screen = origin
125			.create_broadcast("bob/screen", Route::announced())
126			.expect("bob screen");
127
128		let event = room.next().await.expect("bob screen");
129		assert_eq!(event.identity.as_str(), "bob");
130		assert_eq!(event.kind, Kind::Screen);
131		assert!(event.broadcast.is_some());
132	}
133}
moq_room/room.rs

moq_room/
room.rs
