phoxal 0.10.0

Phoxal — production-oriented autonomous robot framework (engine, model, typed bus, contracts).
Documentation 
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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135

use zenoh::{
    Result as ZResult,
    handlers::FifoChannelHandler,
    key_expr::OwnedKeyExpr,
    liveliness::LivelinessToken,
    pubsub::Subscriber,
    sample::{Sample, SampleKind},
};

use crate::bus::Bus;

/// Event for a liveliness key appearing or disappearing.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum LivelinessEvent {
    Alive(String),
    Dropped(String),
}

/// Declare a liveliness token for a key.
///
/// The announcement remains alive until the returned token is dropped or
/// undeclared.
pub async fn declare_liveliness_token(bus: &Bus, key: &str) -> ZResult<LivelinessToken> {
    let key_expr = OwnedKeyExpr::new(bus.topic(key))?;
    bus.session().liveliness().declare_token(key_expr).await
}

/// Subscribe to liveliness changes under a key expression.
///
/// `key_prefix` is a Zenoh key expression, so wildcard prefixes such as
/// `component/front_camera/rgb/profile/**` are accepted.
pub async fn liveliness_subscriber(bus: &Bus, key_prefix: &str) -> ZResult<LivelinessSubscriber> {
    let key_expr = OwnedKeyExpr::new(bus.topic(key_prefix))?;
    let inner = bus
        .session()
        .liveliness()
        .declare_subscriber(key_expr)
        .history(true)
        .await?;
    Ok(LivelinessSubscriber { inner })
}

pub struct LivelinessSubscriber {
    inner: Subscriber<FifoChannelHandler<Sample>>,
}

impl LivelinessSubscriber {
    pub async fn recv(&self) -> ZResult<LivelinessEvent> {
        self.recv_async().await
    }

    pub async fn recv_async(&self) -> ZResult<LivelinessEvent> {
        let sample = self.inner.recv_async().await?;
        Ok(event_from_sample(&sample))
    }

    pub fn key_expr(&self) -> &zenoh::key_expr::KeyExpr<'static> {
        self.inner.key_expr()
    }
}

fn event_from_sample(sample: &Sample) -> LivelinessEvent {
    let key = sample.key_expr().as_str().to_string();
    match sample.kind() {
        SampleKind::Put => LivelinessEvent::Alive(key),
        SampleKind::Delete => LivelinessEvent::Dropped(key),
    }
}

#[cfg(test)]
mod tests {
    use std::time::{Duration, SystemTime, UNIX_EPOCH};

    use serial_test::serial;
    use tokio::time::timeout;

    use super::{LivelinessEvent, declare_liveliness_token, liveliness_subscriber};
    use crate::bus::Bus;

    fn unique_key(suffix: &str) -> String {
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos();
        format!("phoxal/tests/liveliness/{suffix}/{nanos}")
    }

    fn local_test_config() -> zenoh::Config {
        let mut config = zenoh::Config::default();
        config.insert_json5("listen/endpoints", "[]").unwrap();
        config
            .insert_json5("scouting/multicast/enabled", "false")
            .unwrap();
        config
    }

    async fn open_bus() -> Bus {
        let session = zenoh::open(local_test_config())
            .await
            .expect("test zenoh session should open");
        Bus::new(session, String::new())
    }

    #[serial]
    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
    async fn liveliness_subscriber_receives_alive_and_dropped_events() {
        let bus = open_bus().await;
        let prefix = unique_key("profile");
        let token_key = format!("{prefix}/rgb_640x480_30");
        let subscriber = liveliness_subscriber(&bus, &format!("{prefix}/**"))
            .await
            .expect("liveliness subscriber should declare");

        let token = declare_liveliness_token(&bus, &token_key)
            .await
            .expect("liveliness token should declare");

        let alive = timeout(Duration::from_secs(5), subscriber.recv())
            .await
            .expect("subscriber should receive alive event")
            .expect("liveliness subscriber should stay open");
        assert_eq!(alive, LivelinessEvent::Alive(token_key.clone()));

        drop(token);

        let dropped = timeout(Duration::from_secs(5), subscriber.recv())
            .await
            .expect("subscriber should receive dropped event")
            .expect("liveliness subscriber should stay open");
        assert_eq!(dropped, LivelinessEvent::Dropped(token_key));

        drop(subscriber);
        bus.close().await.expect("bus should close");
    }
}