zenkey_fleet/bus/
scout.rs

1//! Raw scouting: the Hello layer, below sessions and liveliness (#116).
2//!
3//! `discover` answers "which deployments hold liveliness tokens or storages"
4//! — a question that presumes a working session. Scouting answers the two
5//! questions that come *before* one: "is anything out there at all" and "is
6//! multicast scouting working on this segment". It is a third, independent
7//! signal, not a replacement for either.
8//!
9//! Multicast is deliberately **on** here — scouting is what this module *is*.
10//! The session-opening default stays off (see `session::open`'s contamination
11//! warning): a scout only listens for Hellos and joins nothing.
12
13use crate::report::HelloView;
14use crate::{Error, Result};
15use zenoh::config::WhatAmIMatcher;
16use zenoh::handlers::FifoChannelHandler;
17use zenoh::scouting::Hello;
18
19impl HelloView {
20    fn of(hello: &Hello) -> Self {
21        HelloView {
22            zid: hello.zid().to_string(),
23            whatami: hello.whatami().to_string(),
24            locators: hello.locators().iter().map(|l| l.to_string()).collect(),
25        }
26    }
27}
28
29/// A running scout. Hellos arrive as the segment answers; the caller owns the
30/// deadline (wrap [`ScoutStream::recv`] in a timeout), because how long to
31/// listen is a question about the network, not about this crate.
32pub struct ScoutStream {
33    inner: zenoh::scouting::Scout<FifoChannelHandler<Hello>>,
34}
35
36impl std::fmt::Debug for ScoutStream {
37    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
38        f.debug_struct("ScoutStream").finish_non_exhaustive()
39    }
40}
41
42impl ScoutStream {
43    /// The next Hello, or `None` once the scout has stopped.
44    pub async fn recv(&self) -> Option<HelloView> {
45        self.inner
46            .recv_async()
47            .await
48            .ok()
49            .map(|h| HelloView::of(&h))
50    }
51
52    /// Stop scouting, explicitly — a drop would stop it too, but silently.
53    pub fn stop(self) {
54        self.inner.stop();
55    }
56}
57
58/// Listen for scouting Hellos: multicast on, plus gossip via any `connect`
59/// endpoints, so a segment with filtered multicast can still answer.
60///
61/// `what` filters by advertised kind; combine with `|`
62/// (`WhatAmI::Router | WhatAmI::Peer`) or pass a parsed [`WhatAmIMatcher`].
63pub async fn scout(
64    what: WhatAmIMatcher,
65    connect: &[String],
66    listen: &[String],
67) -> Result<ScoutStream> {
68    let config = crate::bus::session::explorer_config(connect, listen, true);
69
70    let inner = zenoh::scout(what, config)
71        .await
72        .map_err(|e| Error::bus("scout", "", e))?;
73    Ok(ScoutStream { inner })
74}
75
76#[cfg(test)]
77mod tests {
78    use super::*;
79
80    /// The CLI's ndjson row and the GUI both hang off this exact shape; a
81    /// field rename here is a wire-format change for every script piping
82    /// `zenctl scout --format ndjson`.
83    #[test]
84    fn a_hello_serializes_flat_and_stable() {
85        let view = HelloView {
86            zid: "a1b2".into(),
87            whatami: "router".into(),
88            locators: vec!["tcp/10.0.0.1:7447".into()],
89        };
90        let json = serde_json::to_value(&view).unwrap();
91        assert_eq!(
92            json,
93            serde_json::json!({
94                "zid": "a1b2",
95                "whatami": "router",
96                "locators": ["tcp/10.0.0.1:7447"],
97            })
98        );
99    }
100}
zenkey_fleet/bus/scout.rs

zenkey_fleet/bus/
scout.rs
