Merge pull request #55 from gabrik/feature-unix-domain-sockets

Unix Domain Socket transport
eclipse-zenoh · Nov 23, 2020 · beda767 · beda767
2 parents 1f8ea92 + f8acca3
commit beda767
Show file tree

Hide file tree

Showing 8 changed files with 1,211 additions and 12 deletions.
diff --git a/zenoh-protocol/Cargo.toml b/zenoh-protocol/Cargo.toml
@@ -35,7 +35,8 @@ maintenance = { status = "actively-developed" }
 tcp = []
 udp = []
 zero-copy = []
-default = ["tcp", "udp", "zero-copy"]
+transport_unixsock-stream = []
+default = ["tcp", "udp", "zero-copy", "transport_unixsock-stream"]
 
 [dependencies]
 async-trait = "0.1.38"

diff --git a/zenoh-protocol/src/link/locator.rs b/zenoh-protocol/src/link/locator.rs
@@ -15,7 +15,9 @@
 use async_std::net::SocketAddr;
 #[cfg(any(feature = "tcp", feature = "udp"))]
 use async_std::net::ToSocketAddrs;
-#[cfg(any(feature = "tcp", feature = "udp"))]
+#[cfg(all(feature = "transport_unixsock-stream", target_family = "unix"))]
+use async_std::path::PathBuf;
+#[cfg(any(feature = "tcp", feature = "udp", feat))]
 use async_std::task;
 
 use std::cmp::PartialEq;
@@ -36,13 +38,17 @@ pub const PORT_SEPARATOR: char = ':';
 pub const STR_TCP: &str = "tcp";
 #[cfg(feature = "udp")]
 pub const STR_UDP: &str = "udp";
+#[cfg(all(feature = "transport_unixsock-stream", target_family = "unix"))]
+pub const STR_UNIXSOCK_STREAM: &str = "unixsock-stream";
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub enum LocatorProtocol {
     #[cfg(feature = "tcp")]
     Tcp,
     #[cfg(feature = "udp")]
     Udp,
+    #[cfg(all(feature = "transport_unixsock-stream", target_family = "unix"))]
+    UnixSockStream,
 }
 
 impl fmt::Display for LocatorProtocol {
@@ -52,6 +58,8 @@ impl fmt::Display for LocatorProtocol {
             LocatorProtocol::Tcp => write!(f, "{}", STR_TCP)?,
             #[cfg(feature = "udp")]
             LocatorProtocol::Udp => write!(f, "{}", STR_UDP)?,
+            #[cfg(all(feature = "transport_unixsock-stream", target_family = "unix"))]
+            LocatorProtocol::UnixSockStream => write!(f, "{}", STR_UNIXSOCK_STREAM)?,
         }
         Ok(())
     }
@@ -63,19 +71,35 @@ pub enum Locator {
     Tcp(SocketAddr),
     #[cfg(feature = "udp")]
     Udp(SocketAddr),
+    #[cfg(all(feature = "transport_unixsock-stream", target_family = "unix"))]
+    UnixSockStream(PathBuf),
 }
 
 impl FromStr for Locator {
     type Err = ZError;
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
-        let split: Vec<&str> = s.split(PROTO_SEPARATOR).collect();
-        if split.len() != 2 {
-            return zerror!(ZErrorKind::InvalidLocator {
-                descr: format!("Missing protocol: {}", s)
-            });
-        }
-        let (proto, addr) = (split[0], split[1]);
+        let sep_index = s.find(PROTO_SEPARATOR);
+
+        let index = match sep_index {
+            Some(index) => index,
+            None => {
+                return zerror!(ZErrorKind::InvalidLocator {
+                    descr: format!("Invalid locator: {}", s)
+                });
+            }
+        };
+
+        let (proto, addr) = s.split_at(index);
+        let addr = match addr.strip_prefix(PROTO_SEPARATOR) {
+            Some(addr) => addr,
+            None => {
+                return zerror!(ZErrorKind::InvalidLocator {
+                    descr: format!("Invalid locator: {}", s)
+                });
+            }
+        };
+
         match proto {
             #[cfg(feature = "tcp")]
             STR_TCP => {
@@ -121,6 +145,20 @@ impl FromStr for Locator {
                 });
                 addr.map(Locator::Udp)
             }
+            #[cfg(all(feature = "transport_unixsock-stream", target_family = "unix"))]
+            STR_UNIXSOCK_STREAM => {
+                let addr = task::block_on(async {
+                    match PathBuf::from(addr).to_str() {
+                        Some(path) => Ok(PathBuf::from(path)),
+                        None => {
+                            let e = format!("Invalid Unix locator: {:?}", addr);
+                            log::warn!("{}", e);
+                            zerror!(ZErrorKind::InvalidLocator { descr: e })
+                        }
+                    }
+                });
+                addr.map(Locator::UnixSockStream)
+            }
             _ => {
                 let e = format!("Invalid protocol locator: {}", proto);
                 log::warn!("{}", e);
@@ -137,6 +175,8 @@ impl Locator {
             Locator::Tcp(..) => LocatorProtocol::Tcp,
             #[cfg(feature = "udp")]
             Locator::Udp(..) => LocatorProtocol::Udp,
+            #[cfg(all(feature = "transport_unixsock-stream", target_family = "unix"))]
+            Locator::UnixSockStream(..) => LocatorProtocol::UnixSockStream,
         }
     }
 }
@@ -148,6 +188,11 @@ impl fmt::Display for Locator {
             Locator::Tcp(addr) => write!(f, "{}/{}", STR_TCP, addr)?,
             #[cfg(feature = "udp")]
             Locator::Udp(addr) => write!(f, "{}/{}", STR_UDP, addr)?,
+            #[cfg(all(feature = "transport_unixsock-stream", target_family = "unix"))]
+            Locator::UnixSockStream(addr) => {
+                let path = addr.to_str().unwrap_or("None");
+                write!(f, "{}/{}", STR_UNIXSOCK_STREAM, path)?
+            }
         }
         Ok(())
     }
@@ -160,6 +205,11 @@ impl fmt::Debug for Locator {
             Locator::Tcp(addr) => (STR_TCP, addr.to_string()),
             #[cfg(feature = "udp")]
             Locator::Udp(addr) => (STR_UDP, addr.to_string()),
+            #[cfg(all(feature = "transport_unixsock-stream", target_family = "unix"))]
+            Locator::UnixSockStream(addr) => {
+                let path = addr.to_str().unwrap_or("None");
+                (STR_UNIXSOCK_STREAM, path.to_string())
+            }
         };
 
         f.debug_struct("Locator")

diff --git a/zenoh-protocol/src/link/manager.rs b/zenoh-protocol/src/link/manager.rs
@@ -17,6 +17,8 @@ use async_std::sync::Arc;
 use crate::link::tcp::ManagerTcp;
 #[cfg(feature = "udp")]
 use crate::link::udp::ManagerUdp;
+#[cfg(all(feature = "transport_unixsock-stream", target_family = "unix"))]
+use crate::link::unixsock_stream::ManagerUnixSockStream;
 use crate::link::{LinkManager, LocatorProtocol};
 use crate::session::SessionManagerInner;
 
@@ -32,6 +34,8 @@ impl LinkManagerBuilder {
             LocatorProtocol::Tcp => Arc::new(ManagerTcp::new(manager)),
             #[cfg(feature = "udp")]
             LocatorProtocol::Udp => Arc::new(ManagerUdp::new(manager)),
+            #[cfg(all(feature = "transport_unixsock-stream", target_family = "unix"))]
+            LocatorProtocol::UnixSockStream => Arc::new(ManagerUnixSockStream::new(manager)),
         }
     }
 }
diff --git a/zenoh-protocol/src/link/mod.rs b/zenoh-protocol/src/link/mod.rs
@@ -22,6 +22,8 @@ pub use manager::*;
 mod tcp;
 #[cfg(feature = "udp")]
 mod udp;
+#[cfg(all(feature = "transport_unixsock-stream", target_family = "unix"))]
+mod unixsock_stream;
 
 /* General imports */
 use async_std::sync::{Arc, Weak};