Connectivity events

io/zenoh-link-commons/Cargo.toml

@@ -35,6 +35,7 @@ async-std = { version = "=1.12.0", default-features = false }
 
 async-trait = "0.1.57"
 flume = "0.10.14"
+serde = "1.0.145"
 
 zenoh-buffers = { version = "0.6.0-beta.1", path = "../../commons/zenoh-buffers/" }
 zenoh-cfg-properties = { version = "0.6.0-beta.1", path = "../../commons/zenoh-cfg-properties/" }

io/zenoh-link-commons/src/lib.rs

@@ -18,6 +18,7 @@
 //!
 //! [Click here for Zenoh's documentation](../zenoh/index.html)
 use async_trait::async_trait;
+use serde::Serialize;
 use std::borrow::Cow;
 use std::cmp::PartialEq;
 use std::fmt;
@@ -37,7 +38,7 @@ const WBUF_SIZE: usize = 64;
 /*************************************/
 /*            GENERAL                */
 /*************************************/
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Serialize, Hash, PartialEq, Eq)]
 pub struct Link {
     pub src: Locator,
     pub dst: Locator,

io/zenoh-transport/src/lib.rs

@@ -26,6 +26,7 @@ pub mod unicast;
 pub use manager::*;
 pub use multicast::*;
 pub use primitives::*;
+use serde::Serialize;
 use std::any::Any;
 use std::sync::Arc;
 pub use unicast::*;
@@ -99,12 +100,14 @@ impl TransportMulticastEventHandler for DummyTransportMulticastEventHandler {
 /*************************************/
 /*             CALLBACK              */
 /*************************************/
-#[derive(Clone)]
+#[derive(Clone, Serialize)]
+#[serde(rename = "Transport")]
 pub struct TransportPeer {
     pub zid: ZenohId,
     pub whatami: WhatAmI,
     pub is_qos: bool,
     pub is_shm: bool,
+    #[serde(skip)]
     pub links: Vec<Link>,
 }
 

io/zenoh-transport/src/unicast/establishment/mod.rs

@@ -254,30 +254,28 @@ pub(super) async fn transport_finalize(
     // Assign a callback if the transport is new
     // Keep the lock to avoid concurrent new_transport and closing/closed notifications
     let a_guard = transport.get_alive().await;
-    match transport.get_callback() {
-        Some(callback) => {
-            // Notify the transport handler there is a new link on this transport
-            callback.new_link(Link::from(link));
-        }
-        None => {
-            let peer = TransportPeer {
-                zid: transport.get_zid(),
-                whatami: transport.get_whatami(),
-                is_qos: transport.is_qos(),
-                is_shm: transport.is_shm(),
-                links: vec![Link::from(link)],
-            };
-            // Notify the transport handler that there is a new transport and get back a callback
-            // NOTE: the read loop of the link the open message was sent on remains blocked
-            //       until new_unicast() returns. The read_loop in the various links
-            //       waits for any eventual transport to associate to.
-            let callback = manager
-                .config
-                .handler
-                .new_unicast(peer, input.transport.clone())?;
-            // Set the callback on the transport
-            transport.set_callback(callback);
-        }
+    if transport.get_callback().is_none() {
+        let peer = TransportPeer {
+            zid: transport.get_zid(),
+            whatami: transport.get_whatami(),
+            is_qos: transport.is_qos(),
+            is_shm: transport.is_shm(),
+            links: vec![Link::from(link)],
+        };
+        // Notify the transport handler that there is a new transport and get back a callback
+        // NOTE: the read loop of the link the open message was sent on remains blocked
+        //       until new_unicast() returns. The read_loop in the various links
+        //       waits for any eventual transport to associate to.
+        let callback = manager
+            .config
+            .handler
+            .new_unicast(peer, input.transport.clone())?;
+        // Set the callback on the transport
+        transport.set_callback(callback);
+    }
+    if let Some(callback) = transport.get_callback() {
+        // Notify the transport handler there is a new link on this transport
+        callback.new_link(Link::from(link));
     }
     drop(a_guard);
 

io/zenoh-transport/src/unicast/transport.rs

@@ -331,10 +331,6 @@ impl TransportUnicastInner {
                     let stl = links.remove(index);
                     *guard = links.into_boxed_slice();
                     drop(guard);
-                    // Notify the callback
-                    if let Some(callback) = zread!(self.callback).as_ref() {
-                        callback.del_link(Link::from(link));
-                    }
                     Target::Link(stl.into())
                 }
             } else {
@@ -346,6 +342,11 @@ impl TransportUnicastInner {
             }
         };
 
+        // Notify the callback
+        if let Some(callback) = zread!(self.callback).as_ref() {
+            callback.del_link(Link::from(link));
+        }
+
         match target {
             Target::Transport => self.delete().await,
             Target::Link(stl) => stl.close().await,

zenoh/src/admin.rs

@@ -0,0 +1,210 @@
+use std::{
+    collections::hash_map::DefaultHasher,
+    hash::{Hash, Hasher},
+    sync::Arc,
+};
+
+//
+// Copyright (c) 2022 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, <[email protected]>
+//
+use crate::{
+    keyexpr,
+    prelude::sync::{KeyExpr, Locality},
+    queryable::Query,
+    Sample, Session, ZResult,
+};
+use zenoh_core::SyncResolve;
+use zenoh_protocol::proto::DataInfo;
+use zenoh_protocol_core::{Encoding, KnownEncoding, SampleKind, WireExpr};
+use zenoh_transport::{TransportEventHandler, TransportPeerEventHandler};
+
+macro_rules! ke_for_sure {
+    ($val:expr) => {
+        unsafe { keyexpr::from_str_unchecked($val) }
+    };
+}
+
+lazy_static::lazy_static!(
+    static ref KE_STARSTAR: &'static keyexpr = ke_for_sure!("**");
+    static ref KE_PREFIX: &'static keyexpr = ke_for_sure!("@/session");
+    static ref KE_TRANSPORT_UNICAST: &'static keyexpr = ke_for_sure!("transport/unicast");
+    static ref KE_LINK: &'static keyexpr = ke_for_sure!("link");
+);
+
+pub(crate) fn init(session: &Session) {
+    if let Ok(own_zid) = keyexpr::new(&session.zid().to_string()) {
+        let admin_key = KeyExpr::from(*KE_PREFIX / own_zid / *KE_STARSTAR)
+            .to_wire(session)
+            .to_owned();
+
+        let _admin_qabl = session.declare_queryable_inner(
+            &admin_key,
+            true,
+            Locality::SessionLocal,
+            Arc::new({
+                let session = session.clone();
+                move |q| super::admin::on_admin_query(&session, q)
+            }),
+        );
+    }
+}
+
+pub(crate) fn on_admin_query(session: &Session, query: Query) {
+    if let Ok(own_zid) = keyexpr::new(&session.zid().to_string()) {
+        for transport in session.runtime.manager().get_transports() {
+            if let Ok(zid) = transport.get_zid().map(|zid| zid.to_string()) {
+                if let Ok(zid) = keyexpr::new(&zid) {
+                    let key_expr = *KE_PREFIX / own_zid / *KE_TRANSPORT_UNICAST / zid;
+                    if query.key_expr().intersects(&key_expr) {
+                        if let Ok(value) = transport
+                            .get_peer()
+                            .map_err(|_| ())
+                            .and_then(|p| serde_json::value::to_value(p).map_err(|_| ()))
+                        {
+                            let _ = query.reply(Ok(Sample::new(key_expr, value))).res_sync();
+                        }
+                    }
+
+                    for link in transport.get_links().unwrap_or_else(|_| vec![]) {
+                        let mut s = DefaultHasher::new();
+                        link.hash(&mut s);
+                        if let Ok(lid) = keyexpr::new(&s.finish().to_string()) {
+                            let key_expr =
+                                *KE_PREFIX / own_zid / *KE_TRANSPORT_UNICAST / zid / *KE_LINK / lid;
+                            if query.key_expr().intersects(&key_expr) {
+                                if let Ok(value) = serde_json::value::to_value(link) {
+                                    let _ =
+                                        query.reply(Ok(Sample::new(key_expr, value))).res_sync();
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
+
+pub(crate) struct Handler {
+    pub(crate) session: Arc<Session>,
+}
+
+impl Handler {
+    pub(crate) fn new(session: Session) -> Self {
+        Self {
+            session: Arc::new(session),
+        }
+    }
+}
+
+impl TransportEventHandler for Handler {
+    fn new_unicast(
+        &self,
+        peer: zenoh_transport::TransportPeer,
+        _transport: zenoh_transport::TransportUnicast,
+    ) -> ZResult<Arc<dyn zenoh_transport::TransportPeerEventHandler>> {
+        if let Ok(own_zid) = keyexpr::new(&self.session.zid().to_string()) {
+            if let Ok(zid) = keyexpr::new(&peer.zid.to_string()) {
+                let expr = WireExpr::from(&(*KE_PREFIX / own_zid / *KE_TRANSPORT_UNICAST / zid))
+                    .to_owned();
+                let info = DataInfo {
+                    encoding: Some(Encoding::Exact(KnownEncoding::AppJson)),
+                    ..Default::default()
+                };
+                self.session.handle_data(
+                    true,
+                    &expr,
+                    Some(info),
+                    serde_json::to_vec(&peer).unwrap().into(),
+                );
+                Ok(Arc::new(PeerHandler {
+                    expr,
+                    session: self.session.clone(),
+                }))
+            } else {
+                bail!("Unable to build keyexpr from zid")
+            }
+        } else {
+            bail!("Unable to build keyexpr from zid")
+        }
+    }
+
+    fn new_multicast(
+        &self,
+        _transport: zenoh_transport::TransportMulticast,
+    ) -> ZResult<Arc<dyn zenoh_transport::TransportMulticastEventHandler>> {
+        bail!("unimplemented")
+    }
+}
+
+pub(crate) struct PeerHandler {
+    pub(crate) expr: WireExpr<'static>,
+    pub(crate) session: Arc<Session>,
+}
+
+impl TransportPeerEventHandler for PeerHandler {
+    fn handle_message(&self, _msg: zenoh_protocol::proto::ZenohMessage) -> ZResult<()> {
+        Ok(())
+    }
+
+    fn new_link(&self, link: zenoh_link::Link) {
+        let mut s = DefaultHasher::new();
+        link.hash(&mut s);
+        let info = DataInfo {
+            encoding: Some(Encoding::Exact(KnownEncoding::AppJson)),
+            ..Default::default()
+        };
+        self.session.handle_data(
+            true,
+            &self
+                .expr
+                .clone()
+                .with_suffix(&format!("/link/{}", s.finish())),
+            Some(info),
+            serde_json::to_vec(&link).unwrap().into(),
+        );
+    }
+
+    fn del_link(&self, link: zenoh_link::Link) {
+        let mut s = DefaultHasher::new();
+        link.hash(&mut s);
+        let info = DataInfo {
+            kind: SampleKind::Delete,
+            ..Default::default()
+        };
+        self.session.handle_data(
+            true,
+            &self
+                .expr
+                .clone()
+                .with_suffix(&format!("/link/{}", s.finish())),
+            Some(info),
+            vec![0u8; 0].into(),
+        );
+    }
+
+    fn closing(&self) {}
+
+    fn closed(&self) {
+        let info = DataInfo {
+            kind: SampleKind::Delete,
+            ..Default::default()
+        };
+        self.session
+            .handle_data(true, &self.expr, Some(info), vec![0u8; 0].into());
+    }
+
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+}

zenoh/src/lib.rs

@@ -95,6 +95,7 @@ pub use zenoh_core::Result;
 
 const GIT_VERSION: &str = git_version!(prefix = "v", cargo_prefix = "v");
 
+mod admin;
 #[macro_use]
 mod session;
 pub use session::*;