zenkey-fleet 0.11.1

Fleet engine for keyspace-v2 Zenoh tooling: disciplined fan-in queries, liveliness roster, registry-slice sets, schema-aware decode, live key-tree monitoring — the shared core of zenctl and zengui
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

//! Matching badges on self-declared entities (#38, RFC 12 §9's allowed
//! half): a [`Publication`] knows when a subscriber matches it, a
//! [`RepeatingQuery`] knows when a queryable serves it — and neither claim
//! reaches beyond the entity this process declared.
//!
//! Event-driven like `lazy.rs`: the listener observes the transition, no
//! status poll races the declaration.
//! Ports are ephemeral (`util::peer_pair`), so two test runs at once
//! cannot collide.

use std::time::Duration;

use zenkey::qos::QosProfile;
use zenkey_fleet::{declare_publication, declare_repeating};

mod util;
use util::peer_pair;

const KEY: &str = "v1/h-cccccccccccc/state/demo/health";

/// A publication's badge flips when a subscriber appears on the far peer,
/// and flips back when it undeclares.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_publication_sees_its_own_subscribers_appear_and_leave() {
    let (a, b) = peer_pair().await;

    let publication = declare_publication(&a, KEY, QosProfile::Transition, None)
        .await
        .expect("declare publication");
    let events = publication.matching_events().await.expect("events");
    assert!(
        !publication.matching_status().await.expect("status"),
        "no subscriber exists yet, anywhere"
    );

    let subscriber = b.declare_subscriber(KEY).await.expect("subscriber");
    let ev = tokio::time::timeout(util::SETTLE, events.recv())
        .await
        .expect("matching event within 5s")
        .expect("listener alive");
    assert!(ev, "a matching subscriber must raise the badge");

    subscriber.undeclare().await.expect("undeclare subscriber");
    let ev = tokio::time::timeout(util::SETTLE, events.recv())
        .await
        .expect("unmatching event within 5s")
        .expect("listener alive");
    assert!(!ev, "the last subscriber leaving must lower it");
}

/// A repeating query's badge flips when a queryable starts serving the
/// keyexpr it asks on.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_repeating_query_sees_a_server_appear() {
    let (a, b) = peer_pair().await;

    let repeating = declare_repeating(
        &zenkey_fleet::Fleet::new(&b, ""),
        KEY,
        Duration::from_secs(5),
    )
    .await
    .expect("declare");
    let events = repeating.matching_events().await.expect("events");

    let _queryable = a
        .declare_queryable(KEY)
        .callback(|_| {})
        .await
        .expect("queryable");
    let ev = tokio::time::timeout(util::SETTLE, events.recv())
        .await
        .expect("matching event within 5s")
        .expect("listener alive");
    assert!(ev, "a serving queryable must raise the badge");
    assert!(
        repeating.matching_status().await.expect("status"),
        "status agrees with the event"
    );
}