zenoh 1.9.0

Zenoh: The Zero Overhead Pub/Sub/Query Protocol.
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
136
137
138
139
140

//
// Copyright (c) 2024 ZettaScale Technology
//
// This program and the accompanying materials are made available under the
// terms of the Eclipse Public License 2.0 which is available at
// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
// which is available at https://www.apache.org/licenses/LICENSE-2.0.
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
//
// Contributors:
//   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
//

//! Connectivity event handler - independent from adminspace
//!
//! This handler subscribes to transport events and broadcasts them
//! to user-registered callbacks through the connectivity API.

use std::sync::Arc;

use zenoh_protocol::core::ZenohIdProto;
use zenoh_result::ZResult;
use zenoh_transport::{
    TransportEventHandler, TransportMulticastEventHandler, TransportPeer, TransportPeerEventHandler,
};

use crate::{api::session::WeakSession, sample::SampleKind};
pub(crate) struct ConnectivityHandler {
    session: WeakSession,
}

impl ConnectivityHandler {
    pub(crate) fn new(session: WeakSession) -> Self {
        Self { session }
    }
}

impl TransportEventHandler for ConnectivityHandler {
    fn new_unicast(
        &self,
        peer: TransportPeer,
        _transport: zenoh_transport::unicast::TransportUnicast,
    ) -> ZResult<Arc<dyn TransportPeerEventHandler>> {
        self.session
            .broadcast_transport_event(SampleKind::Put, &peer, false);

        // Return ConnectivityPeerHandler
        Ok(Arc::new(ConnectivityPeerHandler {
            session: self.session.clone(),
            peer_zid: peer.zid,
            peer,
            is_multicast: false,
        }))
    }

    fn new_multicast(
        &self,
        _transport: zenoh_transport::multicast::TransportMulticast,
    ) -> ZResult<Arc<dyn TransportMulticastEventHandler>> {
        Ok(Arc::new(ConnectivityMulticastHandler {
            session: self.session.clone(),
        }))
    }
}

/// Peer handler for connectivity events
pub(crate) struct ConnectivityPeerHandler {
    session: WeakSession,
    peer_zid: ZenohIdProto,
    peer: TransportPeer,
    is_multicast: bool,
}

impl TransportPeerEventHandler for ConnectivityPeerHandler {
    fn handle_message(&self, _msg: zenoh_protocol::network::NetworkMessageMut) -> ZResult<()> {
        // Connectivity doesn't need to handle messages
        Ok(())
    }

    fn new_link(&self, link: zenoh_link::Link) {
        // Broadcast link added event
        self.session.broadcast_link_event(
            SampleKind::Put,
            self.peer_zid,
            &link,
            self.is_multicast,
            self.peer.is_qos,
        );
    }

    fn del_link(&self, link: zenoh_link::Link) {
        // Broadcast link removed event
        self.session.broadcast_link_event(
            SampleKind::Delete,
            self.peer_zid,
            &link,
            self.is_multicast,
            self.peer.is_qos,
        );
    }

    fn closed(&self) {
        // Broadcast transport closed event
        self.session
            .broadcast_transport_event(SampleKind::Delete, &self.peer, self.is_multicast);
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }
}

/// Multicast handler for connectivity events
pub(crate) struct ConnectivityMulticastHandler {
    session: WeakSession,
}

impl TransportMulticastEventHandler for ConnectivityMulticastHandler {
    fn new_peer(&self, peer: TransportPeer) -> ZResult<Arc<dyn TransportPeerEventHandler>> {
        self.session
            .broadcast_transport_event(SampleKind::Put, &peer, true);

        // Return ConnectivityPeerHandler
        Ok(Arc::new(ConnectivityPeerHandler {
            session: self.session.clone(),
            peer_zid: peer.zid,
            peer,
            is_multicast: true,
        }))
    }

    fn closed(&self) {
        // Nothing to do for multicast group closure
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }
}