Implement Unix domain sockets in libnative

2 files changed, 358 insertions, 2 deletions

diff --git a/src/libnative/io/mod.rs b/src/libnative/io/mod.rs
index 69ef10ac11b..dc134aa95fe 100644
--- a/src/libnative/io/mod.rs
+++ b/src/libnative/io/mod.rs
@@ -197,10 +197,10 @@ impl rtio::IoFactory for IoFactory {
         net::UdpSocket::bind(addr).map(|u| ~u as ~RtioUdpSocket)
     }
     fn unix_bind(&mut self, _path: &CString) -> IoResult<~RtioUnixListener> {
-        Err(unimpl())
+        net::UnixListener::bind(_path).map(|s| ~s as ~RtioUnixListener)
     }
     fn unix_connect(&mut self, _path: &CString) -> IoResult<~RtioPipe> {
-        Err(unimpl())
+        net::UnixStream::connect(_path, libc::SOCK_STREAM).map(|s| ~s as ~RtioPipe)
     }
     fn get_host_addresses(&mut self, host: Option<&str>, servname: Option<&str>,
                           hint: Option<ai::Hint>) -> IoResult<~[ai::Info]> {
diff --git a/src/libnative/io/net.rs b/src/libnative/io/net.rs
index b33b54862dc..62753b66af0 100644
--- a/src/libnative/io/net.rs
+++ b/src/libnative/io/net.rs
@@ -8,6 +8,7 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
+use std::c_str::CString;
 use std::cast;
 use std::io::net::ip;
 use std::io;
@@ -15,6 +16,7 @@ use std::libc;
 use std::mem;
 use std::rt::rtio;
 use std::sync::arc::UnsafeArc;
+use std::unstable::intrinsics;
 
 use super::{IoResult, retry};
 use super::file::keep_going;
@@ -88,6 +90,30 @@ fn addr_to_sockaddr(addr: ip::SocketAddr) -> (libc::sockaddr_storage, uint) {
     }
 }
 
+fn addr_to_sockaddr_un(addr: &CString) -> IoResult<(libc::sockaddr_storage, uint)> {
+    // the sun_path length is limited to SUN_LEN (with null)
+    if addr.len() > libc::sun_len -1 {
+        return Err(io::IoError {
+            kind: io::OtherIoError,
+            desc: "path must be smaller than SUN_LEN",
+            detail: None,
+        })
+    }
+    unsafe {
+        let storage: libc::sockaddr_storage = intrinsics::init();
+        let s: *mut libc::sockaddr_un = cast::transmute(&storage);
+        (*s).sun_family = libc::AF_UNIX as libc::sa_family_t;
+        let mut i = 0;
+        for c in addr.iter() {
+          (*s).sun_path[i] = c;
+          i += 1;
+        }
+
+        let len = mem::size_of::<libc::sa_family_t>() + i + 1; //count the null terminator
+        return Ok((storage, len));
+    }
+}
+
 fn socket(addr: ip::SocketAddr, ty: libc::c_int) -> IoResult<sock_t> {
     unsafe {
         let fam = match addr.ip {
@@ -101,6 +127,15 @@ fn socket(addr: ip::SocketAddr, ty: libc::c_int) -> IoResult<sock_t> {
     }
 }
 
+fn unix_socket(ty: libc::c_int) -> IoResult<sock_t> {
+    unsafe {
+        match libc::socket(libc::AF_UNIX, ty, 0) {
+            -1 => Err(super::last_error()),
+            fd => Ok(fd)
+        }
+    }
+}
+
 fn setsockopt<T>(fd: sock_t, opt: libc::c_int, val: libc::c_int,
                  payload: T) -> IoResult<()> {
     unsafe {
@@ -193,6 +228,24 @@ pub fn sockaddr_to_addr(storage: &libc::sockaddr_storage,
     }
 }
 
+fn sockaddr_to_unix(storage: &libc::sockaddr_storage,
+                    len: uint) -> IoResult<CString> {
+    match storage.ss_family as libc::c_int {
+        libc::AF_UNIX => {
+            assert!(len as uint <= mem::size_of::<libc::sockaddr_un>());
+            let storage: &libc::sockaddr_un = unsafe {
+                cast::transmute(storage)
+            };
+            unsafe {
+                Ok(CString::new(storage.sun_path.to_owned().as_ptr(), false))
+            }
+        }
+        _ => {
+            Err(io::standard_error(io::OtherIoError))
+        }
+    }
+}
+
 #[cfg(unix)]
 pub fn init() {}
 
@@ -632,3 +685,306 @@ impl rtio::RtioUdpSocket for UdpSocket {
         ~UdpSocket { inner: self.inner.clone() } as ~rtio::RtioUdpSocket
     }
 }
+
+
+#[cfg(not(windows))]
+////////////////////////////////////////////////////////////////////////////////
+// Unix
+////////////////////////////////////////////////////////////////////////////////
+
+
+////////////////////////////////////////////////////////////////////////////////
+// Unix streams
+////////////////////////////////////////////////////////////////////////////////
+
+pub struct UnixStream {
+    priv fd: sock_t,
+}
+
+impl UnixStream {
+    pub fn connect(addr: &CString, ty: libc::c_int) -> IoResult<UnixStream> {
+        unsafe {
+            unix_socket(ty).and_then(|fd| {
+                match addr_to_sockaddr_un(addr) {
+                    Err(e)          => return Err(e),
+                    Ok((addr, len)) => {
+                        let ret = UnixStream{ fd: fd };
+                        let addrp = &addr as *libc::sockaddr_storage;
+                        match retry(|| {
+                          libc::connect(fd, addrp as *libc::sockaddr,
+                                         len as libc::socklen_t)
+                        }) {
+                            -1 => return Err(super::last_error()),
+                            _  => return Ok(ret)
+                        }
+                    }
+                }
+            })
+        }
+    }
+
+    pub fn fd(&self) -> sock_t { self.fd }
+}
+
+impl rtio::RtioPipe for UnixStream {
+    fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> {
+        let ret = retry(|| {
+            unsafe {
+                libc::recv(self.fd,
+                           buf.as_ptr() as *mut libc::c_void,
+                           buf.len() as wrlen,
+                           0) as libc::c_int
+            }
+        });
+        if ret == 0 {
+            Err(io::standard_error(io::EndOfFile))
+        } else if ret < 0 {
+            Err(super::last_error())
+        } else {
+            Ok(ret as uint)
+        }
+    }
+    fn write(&mut self, buf: &[u8]) -> IoResult<()> {
+        let ret = keep_going(buf, |buf, len| {
+            unsafe {
+                libc::send(self.fd,
+                           buf as *mut libc::c_void,
+                           len as wrlen,
+                           0) as i64
+            }
+        });
+        if ret < 0 {
+            Err(super::last_error())
+        } else {
+            Ok(())
+        }
+    }
+}
+
+impl Drop for UnixStream {
+    fn drop(&mut self) { unsafe { close(self.fd); } }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+// Unix Datagram
+////////////////////////////////////////////////////////////////////////////////
+
+pub struct UnixDatagram {
+    priv fd: sock_t,
+}
+
+impl UnixDatagram {
+    pub fn connect(addr: &CString, ty: libc::c_int) -> IoResult<UnixDatagram> {
+        unsafe {
+            unix_socket(ty).and_then(|fd| {
+                match addr_to_sockaddr_un(addr) {
+                    Err(e)          => return Err(e),
+                    Ok((addr, len)) => {
+                        let ret = UnixDatagram{ fd: fd };
+                        let addrp = &addr as *libc::sockaddr_storage;
+                        match retry(|| {
+                          libc::connect(fd, addrp as *libc::sockaddr,
+                                         len as libc::socklen_t)
+                        }) {
+                            -1 => return Err(super::last_error()),
+                            _  => return Ok(ret)
+                        }
+                    }
+                }
+            })
+        }
+    }
+
+    pub fn bind(addr: &CString) -> IoResult<UnixDatagram> {
+        unsafe {
+            unix_socket(libc::SOCK_DGRAM).and_then(|fd| {
+                match addr_to_sockaddr_un(addr) {
+                    Err(e)          => return Err(e),
+                    Ok((addr, len)) => {
+                        let ret = UnixDatagram{ fd: fd };
+                        let addrp = &addr as *libc::sockaddr_storage;
+                        match libc::bind(fd, addrp as *libc::sockaddr,
+                                         len as libc::socklen_t) {
+                            -1 => return Err(super::last_error()),
+                            _  => return Ok(ret)
+                        }
+                    }
+                }
+            })
+        }
+    }
+
+    pub fn fd(&self) -> sock_t { self.fd }
+}
+
+impl rtio::RtioPipe for UnixDatagram {
+    fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> {
+        let ret = retry(|| {
+            unsafe {
+                libc::recv(self.fd,
+                           buf.as_ptr() as *mut libc::c_void,
+                           buf.len() as wrlen,
+                           0) as libc::c_int
+            }
+        });
+        if ret == 0 {
+            Err(io::standard_error(io::EndOfFile))
+        } else if ret < 0 {
+            Err(super::last_error())
+        } else {
+            Ok(ret as uint)
+        }
+    }
+    fn write(&mut self, buf: &[u8]) -> IoResult<()> {
+        let ret = keep_going(buf, |buf, len| {
+            unsafe {
+                libc::send(self.fd,
+                           buf as *mut libc::c_void,
+                           len as wrlen,
+                           0) as i64
+            }
+        });
+        if ret < 0 {
+            Err(super::last_error())
+        } else {
+            Ok(())
+        }
+    }
+}
+
+impl rtio::RtioDatagramPipe for UnixDatagram {
+    fn recvfrom(&mut self, buf: &mut [u8]) -> IoResult<(uint, CString)> {
+        unsafe {
+            let mut storage: libc::sockaddr_storage = intrinsics::init();
+            let storagep = &mut storage as *mut libc::sockaddr_storage;
+            let mut addrlen: libc::socklen_t =
+                    mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
+            let ret = retry(|| {
+                libc::recvfrom(self.fd,
+                               buf.as_ptr() as *mut libc::c_void,
+                               buf.len() as msglen_t,
+                               0,
+                               storagep as *mut libc::sockaddr,
+                               &mut addrlen) as libc::c_int
+            });
+            if ret < 0 { return Err(super::last_error()) }
+            sockaddr_to_unix(&storage, addrlen as uint).and_then(|addr| {
+                Ok((ret as uint, addr))
+            })
+        }
+    }
+
+    fn sendto(&mut self, buf: &[u8], dst: &CString) -> IoResult<()> {
+        match addr_to_sockaddr_un(dst) {
+            Err(e)         => Err(e),
+            Ok((dst, len)) => {
+                let dstp = &dst as *libc::sockaddr_storage;
+                unsafe {
+                    let ret = retry(|| {
+                        libc::sendto(self.fd,
+                                     buf.as_ptr() as *libc::c_void,
+                                     buf.len() as msglen_t,
+                                     0,
+                                     dstp as *libc::sockaddr,
+                                     len as libc::socklen_t) as libc::c_int
+                    });
+                    match ret {
+                        -1 => Err(super::last_error()),
+                        n if n as uint != buf.len() => {
+                            Err(io::IoError {
+                                kind: io::OtherIoError,
+                                desc: "couldn't send entire packet at once",
+                                detail: None,
+                            })
+                        }
+                        _ => Ok(())
+                    }
+                }
+            }
+        }
+    }
+}
+
+impl Drop for UnixDatagram {
+    fn drop(&mut self) { unsafe { close(self.fd); } }
+}
+////////////////////////////////////////////////////////////////////////////////
+// Unix Listener
+////////////////////////////////////////////////////////////////////////////////
+
+pub struct UnixListener {
+    priv fd: sock_t,
+}
+
+impl UnixListener {
+    pub fn bind(addr: &CString) -> IoResult<UnixListener> {
+        unsafe {
+            unix_socket(libc::SOCK_STREAM).and_then(|fd| {
+                match addr_to_sockaddr_un(addr) {
+                    Err(e)          => return Err(e),
+                    Ok((addr, len)) => {
+                        let ret = UnixListener{ fd: fd };
+                        let addrp = &addr as *libc::sockaddr_storage;
+                        match libc::bind(fd, addrp as *libc::sockaddr,
+                                         len as libc::socklen_t) {
+                            -1 => return Err(super::last_error()),
+                            _  => return Ok(ret)
+                        }
+                    }
+                }
+            })
+        }
+    }
+
+    pub fn fd(&self) -> sock_t { self.fd }
+
+    pub fn native_listen(self, backlog: int) -> IoResult<UnixAcceptor> {
+        match unsafe { libc::listen(self.fd, backlog as libc::c_int) } {
+            -1 => Err(super::last_error()),
+            _ => Ok(UnixAcceptor { listener: self })
+        }
+    }
+}
+
+impl rtio::RtioUnixListener for UnixListener {
+    fn listen(~self) -> IoResult<~rtio::RtioUnixAcceptor> {
+        self.native_listen(128).map(|a| ~a as ~rtio::RtioUnixAcceptor)
+    }
+}
+
+impl Drop for UnixListener {
+    fn drop(&mut self) { unsafe { close(self.fd); } }
+}
+
+pub struct UnixAcceptor {
+    priv listener: UnixListener,
+}
+
+impl UnixAcceptor {
+    pub fn fd(&self) -> sock_t { self.listener.fd }
+
+    pub fn native_accept(&mut self) -> IoResult<UnixStream> {
+        unsafe {
+            let mut storage: libc::sockaddr_storage = intrinsics::init();
+            let storagep = &mut storage as *mut libc::sockaddr_storage;
+            let size = mem::size_of::<libc::sockaddr_storage>();
+            let mut size = size as libc::socklen_t;
+            match retry(|| {
+                libc::accept(self.fd(),
+                             storagep as *mut libc::sockaddr,
+                             &mut size as *mut libc::socklen_t) as libc::c_int
+            }) as sock_t {
+                -1 => Err(super::last_error()),
+                fd => Ok(UnixStream { fd: fd })
+            }
+        }
+    }
+}
+
+impl rtio::RtioUnixAcceptor for UnixAcceptor {
+    fn accept(&mut self) -> IoResult<~rtio::RtioPipe> {
+        self.native_accept().map(|s| ~s as ~rtio::RtioPipe)
+    }
+}
+
+