Implement io::net::unix

3 files changed, 271 insertions, 19 deletions

diff --git a/src/libstd/rt/io/net/unix.rs b/src/libstd/rt/io/net/unix.rs
index 1771a963ba7..9428c1f800d 100644
--- a/src/libstd/rt/io/net/unix.rs
+++ b/src/libstd/rt/io/net/unix.rs
@@ -8,44 +8,296 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
+/*!
+
+Named pipes
+
+This module contains the ability to communicate over named pipes with
+synchronous I/O. On windows, this corresponds to talking over a Named Pipe,
+while on Unix it corresponds to UNIX domain sockets.
+
+These pipes are similar to TCP in the sense that you can have both a stream to a
+server and a server itself. The server provided accepts other `UnixStream`
+instances as clients.
+
+*/
+
 use prelude::*;
-use super::super::*;
+
 use super::super::support::PathLike;
+use rt::rtio::{IoFactory, IoFactoryObject, RtioUnixListenerObject};
+use rt::rtio::{RtioUnixAcceptorObject, RtioPipeObject, RtioUnixListener};
+use rt::rtio::RtioUnixAcceptor;
+use rt::io::pipe::PipeStream;
+use rt::io::{io_error, Listener, Acceptor, Reader, Writer};
+use rt::local::Local;
 
-pub struct UnixStream;
+/// A stream which communicates over a named pipe.
+pub struct UnixStream {
+    priv obj: PipeStream,
+}
 
 impl UnixStream {
-    pub fn connect<P: PathLike>(_path: &P) -> Option<UnixStream> {
-        fail!()
+    fn new(obj: ~RtioPipeObject) -> UnixStream {
+        UnixStream { obj: PipeStream::new_bound(obj) }
+    }
+
+    /// Connect to a pipe named by `path`. This will attempt to open a
+    /// connection to the underlying socket.
+    ///
+    /// The returned stream will be closed when the object falls out of scope.
+    ///
+    /// # Failure
+    ///
+    /// This function will raise on the `io_error` condition if the connection
+    /// could not be made.
+    ///
+    /// # Example
+    ///
+    ///     use std::rt::io::net::unix::UnixStream;
+    ///
+    ///     let server = Path("path/to/my/socket");
+    ///     let mut stream = UnixStream::connect(&server);
+    ///     stream.write([1, 2, 3]);
+    ///
+    pub fn connect<P: PathLike>(path: &P) -> Option<UnixStream> {
+        let pipe = unsafe {
+            let io: *mut IoFactoryObject = Local::unsafe_borrow();
+            (*io).unix_connect(path)
+        };
+
+        match pipe {
+            Ok(s) => Some(UnixStream::new(s)),
+            Err(ioerr) => {
+                io_error::cond.raise(ioerr);
+                None
+            }
+        }
     }
 }
 
 impl Reader for UnixStream {
-    fn read(&mut self, _buf: &mut [u8]) -> Option<uint> { fail!() }
-
-    fn eof(&mut self) -> bool { fail!() }
+    fn read(&mut self, buf: &mut [u8]) -> Option<uint> { self.obj.read(buf) }
+    fn eof(&mut self) -> bool { self.obj.eof() }
 }
 
 impl Writer for UnixStream {
-    fn write(&mut self, _v: &[u8]) { fail!() }
-
-    fn flush(&mut self) { fail!() }
+    fn write(&mut self, buf: &[u8]) { self.obj.write(buf) }
+    fn flush(&mut self) { self.obj.flush() }
 }
 
-pub struct UnixListener;
+pub struct UnixListener {
+    priv obj: ~RtioUnixListenerObject,
+}
 
 impl UnixListener {
-    pub fn bind<P: PathLike>(_path: &P) -> Option<UnixListener> {
-        fail!()
+
+    /// Creates a new listener, ready to receive incoming connections on the
+    /// specified socket. The server will be named by `path`.
+    ///
+    /// This listener will be closed when it falls out of scope.
+    ///
+    /// # Failure
+    ///
+    /// This function will raise on the `io_error` condition if the specified
+    /// path could not be bound.
+    ///
+    /// # Example
+    ///
+    ///     use std::rt::io::net::unix::UnixListener;
+    ///
+    ///     let server = Path("path/to/my/socket");
+    ///     let mut stream = UnixListener::bind(&server);
+    ///     for client in stream.incoming() {
+    ///         let mut client = client;
+    ///         client.write([1, 2, 3, 4]);
+    ///     }
+    ///
+    pub fn bind<P: PathLike>(path: &P) -> Option<UnixListener> {
+        let listener = unsafe {
+            let io: *mut IoFactoryObject = Local::unsafe_borrow();
+            (*io).unix_bind(path)
+        };
+        match listener {
+            Ok(s) => Some(UnixListener{ obj: s }),
+            Err(ioerr) => {
+                io_error::cond.raise(ioerr);
+                None
+            }
+        }
     }
 }
 
 impl Listener<UnixStream, UnixAcceptor> for UnixListener {
-    fn listen(self) -> Option<UnixAcceptor> { fail!() }
+    fn listen(self) -> Option<UnixAcceptor> {
+        match self.obj.listen() {
+            Ok(acceptor) => Some(UnixAcceptor { obj: acceptor }),
+            Err(ioerr) => {
+                io_error::cond.raise(ioerr);
+                None
+            }
+        }
+    }
 }
 
-pub struct UnixAcceptor;
+pub struct UnixAcceptor {
+    priv obj: ~RtioUnixAcceptorObject,
+}
 
 impl Acceptor<UnixStream> for UnixAcceptor {
-    fn accept(&mut self) -> Option<UnixStream> { fail!() }
+    fn accept(&mut self) -> Option<UnixStream> {
+        match self.obj.accept() {
+            Ok(s) => Some(UnixStream::new(s)),
+            Err(ioerr) => {
+                io_error::cond.raise(ioerr);
+                None
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use prelude::*;
+    use super::*;
+    use cell::Cell;
+    use rt::test::*;
+    use rt::io::*;
+    use rt::comm::oneshot;
+    use os;
+
+    fn smalltest(server: ~fn(UnixStream), client: ~fn(UnixStream)) {
+        let server = Cell::new(server);
+        let client = Cell::new(client);
+        do run_in_mt_newsched_task {
+            let server = Cell::new(server.take());
+            let client = Cell::new(client.take());
+            let path1 = next_test_unix();
+            let path2 = path1.clone();
+            let (port, chan) = oneshot();
+            let port = Cell::new(port);
+            let chan = Cell::new(chan);
+
+            do spawntask {
+                let mut acceptor = UnixListener::bind(&path1).listen();
+                chan.take().send(());
+                server.take()(acceptor.accept().unwrap());
+            }
+
+            do spawntask {
+                port.take().recv();
+                client.take()(UnixStream::connect(&path2).unwrap());
+            }
+        }
+    }
+
+    #[test]
+    fn bind_error() {
+        do run_in_mt_newsched_task {
+            let mut called = false;
+            do io_error::cond.trap(|e| {
+                assert!(e.kind == PermissionDenied);
+                called = true;
+            }).inside {
+                let listener = UnixListener::bind(&("path/to/nowhere"));
+                assert!(listener.is_none());
+            }
+            assert!(called);
+        }
+    }
+
+    #[test]
+    fn connect_error() {
+        do run_in_mt_newsched_task {
+            let mut called = false;
+            do io_error::cond.trap(|e| {
+                assert_eq!(e.kind, OtherIoError);
+                called = true;
+            }).inside {
+                let stream = UnixStream::connect(&("path/to/nowhere"));
+                assert!(stream.is_none());
+            }
+            assert!(called);
+        }
+    }
+
+    #[test]
+    fn smoke() {
+        smalltest(|mut server| {
+            let mut buf = [0];
+            server.read(buf);
+            assert!(buf[0] == 99);
+        }, |mut client| {
+            client.write([99]);
+        })
+    }
+
+    #[test]
+    fn read_eof() {
+        smalltest(|mut server| {
+            let mut buf = [0];
+            assert!(server.read(buf).is_none());
+            assert!(server.read(buf).is_none());
+        }, |_client| {
+            // drop the client
+        })
+    }
+
+    #[test]
+    fn write_begone() {
+        smalltest(|mut server| {
+            let buf = [0];
+            let mut stop = false;
+            while !stop{
+                do io_error::cond.trap(|e| {
+                    assert_eq!(e.kind, BrokenPipe);
+                    stop = true;
+                }).inside {
+                    server.write(buf);
+                }
+            }
+        }, |_client| {
+            // drop the client
+        })
+    }
+
+    #[test]
+    fn accept_lots() {
+        do run_in_mt_newsched_task {
+            let times = 10;
+            let path1 = next_test_unix();
+            let path2 = path1.clone();
+            let (port, chan) = oneshot();
+            let port = Cell::new(port);
+            let chan = Cell::new(chan);
+
+            do spawntask {
+                let mut acceptor = UnixListener::bind(&path1).listen();
+                chan.take().send(());
+                do times.times {
+                    let mut client = acceptor.accept();
+                    let mut buf = [0];
+                    client.read(buf);
+                    assert_eq!(buf[0], 100);
+                }
+            }
+
+            do spawntask {
+                port.take().recv();
+                do times.times {
+                    let mut stream = UnixStream::connect(&path2);
+                    stream.write([100]);
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn path_exists() {
+        do run_in_mt_newsched_task {
+            let path = next_test_unix();
+            let _acceptor = UnixListener::bind(&path).listen();
+            assert!(os::path_exists(&path));
+        }
+    }
 }
diff --git a/src/libstd/rt/io/pipe.rs b/src/libstd/rt/io/pipe.rs
index d2cd531ed26..ff1bd55d594 100644
--- a/src/libstd/rt/io/pipe.rs
+++ b/src/libstd/rt/io/pipe.rs
@@ -21,7 +21,7 @@ use rt::rtio::{RtioPipe, RtioPipeObject, IoFactoryObject, IoFactory};
 use rt::rtio::RtioUnboundPipeObject;
 
 pub struct PipeStream {
-    priv obj: RtioPipeObject
+    priv obj: ~RtioPipeObject
 }
 
 // This should not be a newtype, but rt::uv::process::set_stdio needs to reach
@@ -45,7 +45,7 @@ impl PipeStream {
         }
     }
 
-    pub fn bind(inner: RtioPipeObject) -> PipeStream {
+    pub fn new_bound(inner: ~RtioPipeObject) -> PipeStream {
         PipeStream { obj: inner }
     }
 }
diff --git a/src/libstd/rt/io/process.rs b/src/libstd/rt/io/process.rs
index 5f2453852ee..e0ffa82b59f 100644
--- a/src/libstd/rt/io/process.rs
+++ b/src/libstd/rt/io/process.rs
@@ -100,7 +100,7 @@ impl Process {
             Ok((p, io)) => Some(Process{
                 handle: p,
                 io: io.move_iter().map(|p|
-                    p.map(|p| io::PipeStream::bind(p))
+                    p.map(|p| io::PipeStream::new_bound(p))
                 ).collect()
             }),
             Err(ioerr) => {