1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
//! What the peer declared in its SETUP, and the slot the session tasks read it from.
use super::cluster;
/// The Setup Options the peer sent us.
///
/// One value per session: both extensions ride the same SETUP, so they arrive together
/// and are read together.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(crate) struct Peer {
/// The MoQ Cluster options: the peer's Hop ID and what it charges to cross.
pub cluster: cluster::Peer,
/// MoQ Solicit: whether the peer requires advertisements to be solicited, so an
/// unsolicited PUBLISH_NAMESPACE is unwanted. `None` when it declared nothing, which
/// is the one case where sending us one anyway is not a protocol violation.
pub solicit: Option<bool>,
}
/// Shared slot for [`Peer`], filled when the peer's SETUP is read.
///
/// The publisher blocks on this before its first advertisement: the SETUP decides both
/// what an advertisement carries (MoQ Cluster) and whether one may be sent unasked (MoQ
/// Solicit), so nothing can go out until it arrives. Cheap to clone; every handle
/// shares the same slot.
#[derive(Clone, Default)]
pub(crate) struct PeerSetup(kio::Shared<Option<Peer>>);
impl PeerSetup {
/// Record what the peer declared. A SETUP carrying no options records the default
/// (no extension negotiated, no requirements), which is what unblocks a waiter.
///
/// First write wins. The announce loops read this once and hold it for their
/// lifetime while subscription serving re-reads it, so letting a later SETUP
/// overwrite the identity would advertise under one exclusion and serve under
/// another, which is how a routing loop gets back in.
pub fn set(&self, peer: Peer) {
let mut slot = self.0.lock();
if slot.is_none() {
*slot = Some(peer);
}
}
/// Await the peer's SETUP.
///
/// The peer MUST send exactly one, so this resolves once that stream is read. Waits
/// forever if it never does; the caller is a session task, cancelled when the driver
/// drops.
pub async fn get(&self) -> Peer {
let slot = self
.0
.wait(|peer| match peer.is_some() {
true => std::task::Poll::Ready(()),
false => std::task::Poll::Pending,
})
.await;
(*slot).expect("waited for Some")
}
}
#[cfg(test)]
mod tests {
use super::*;
/// The announce loops read the peer's declaration once and hold it, while
/// subscription serving re-reads it. A second SETUP overwriting the identity would
/// split those two apart, so the first write is the one that counts.
#[tokio::test]
async fn first_write_wins() {
let first = Peer {
cluster: cluster::Peer {
origin: Some(crate::Origin::new(42).unwrap()),
cost: Some(3),
},
solicit: None,
};
let slot = PeerSetup::default();
slot.set(first);
slot.set(Peer {
cluster: cluster::Peer {
origin: Some(crate::Origin::new(99).unwrap()),
cost: Some(0),
},
solicit: Some(true),
});
assert_eq!(slot.get().await, first);
}
}