Implement io::net::unix

5 files changed, 288 insertions, 66 deletions

diff --git a/src/libstd/rt/uv/net.rs b/src/libstd/rt/uv/net.rs
index a2608bf6b24..2e85900a3f2 100644
--- a/src/libstd/rt/uv/net.rs
+++ b/src/libstd/rt/uv/net.rs
@@ -206,12 +206,6 @@ impl StreamWatcher {
         }
     }
 
-    pub fn accept(&mut self, stream: StreamWatcher) {
-        let self_handle = self.native_handle() as *c_void;
-        let stream_handle = stream.native_handle() as *c_void;
-        assert_eq!(0, unsafe { uvll::accept(self_handle, stream_handle) } );
-    }
-
     pub fn close(self, cb: NullCallback) {
         {
             let mut this = self;
@@ -230,6 +224,36 @@ impl StreamWatcher {
             cb();
         }
     }
+
+    pub fn listen(&mut self, cb: ConnectionCallback) -> Result<(), UvError> {
+        {
+            let data = self.get_watcher_data();
+            assert!(data.connect_cb.is_none());
+            data.connect_cb = Some(cb);
+        }
+
+        unsafe {
+            static BACKLOG: c_int = 128; // XXX should be configurable
+            match uvll::listen(self.native_handle(), BACKLOG, connection_cb) {
+                0 => Ok(()),
+                n => Err(UvError(n))
+            }
+        }
+
+        extern fn connection_cb(handle: *uvll::uv_stream_t, status: c_int) {
+            rtdebug!("connection_cb");
+            let mut stream_watcher: StreamWatcher = NativeHandle::from_native_handle(handle);
+            let cb = stream_watcher.get_watcher_data().connect_cb.get_ref();
+            let status = status_to_maybe_uv_error(status);
+            (*cb)(stream_watcher, status);
+        }
+    }
+
+    pub fn accept(&mut self, stream: StreamWatcher) {
+        let self_handle = self.native_handle() as *c_void;
+        let stream_handle = stream.native_handle() as *c_void;
+        assert_eq!(0, unsafe { uvll::accept(self_handle, stream_handle) } );
+    }
 }
 
 impl NativeHandle<*uvll::uv_stream_t> for StreamWatcher {
@@ -300,28 +324,6 @@ impl TcpWatcher {
         }
     }
 
-    pub fn listen(&mut self, cb: ConnectionCallback) {
-        {
-            let data = self.get_watcher_data();
-            assert!(data.connect_cb.is_none());
-            data.connect_cb = Some(cb);
-        }
-
-        unsafe {
-            static BACKLOG: c_int = 128; // XXX should be configurable
-            // XXX: This can probably fail
-            assert_eq!(0, uvll::listen(self.native_handle(), BACKLOG, connection_cb));
-        }
-
-        extern fn connection_cb(handle: *uvll::uv_stream_t, status: c_int) {
-            rtdebug!("connection_cb");
-            let mut stream_watcher: StreamWatcher = NativeHandle::from_native_handle(handle);
-            let cb = stream_watcher.get_watcher_data().connect_cb.get_ref();
-            let status = status_to_maybe_uv_error(status);
-            (*cb)(stream_watcher, status);
-        }
-    }
-
     pub fn as_stream(&self) -> StreamWatcher {
         NativeHandle::from_native_handle(self.native_handle() as *uvll::uv_stream_t)
     }
@@ -644,7 +646,8 @@ mod test {
             server_tcp_watcher.bind(addr);
             let loop_ = loop_;
             rtdebug!("listening");
-            do server_tcp_watcher.listen |mut server_stream_watcher, status| {
+            let mut stream = server_tcp_watcher.as_stream();
+            let res = do stream.listen |mut server_stream_watcher, status| {
                 rtdebug!("listened!");
                 assert!(status.is_none());
                 let mut loop_ = loop_;
@@ -678,7 +681,9 @@ mod test {
                     }
                     count_cell.put_back(count);
                 }
-            }
+            };
+
+            assert!(res.is_ok());
 
             let client_thread = do Thread::start {
                 rtdebug!("starting client thread");
@@ -705,7 +710,7 @@ mod test {
             loop_.run();
             loop_.close();
             client_thread.join();
-        }
+        };
     }
 
     #[test]
@@ -718,7 +723,8 @@ mod test {
             server_tcp_watcher.bind(addr);
             let loop_ = loop_;
             rtdebug!("listening");
-            do server_tcp_watcher.listen |mut server_stream_watcher, status| {
+            let mut stream = server_tcp_watcher.as_stream();
+            let res = do stream.listen |mut server_stream_watcher, status| {
                 rtdebug!("listened!");
                 assert!(status.is_none());
                 let mut loop_ = loop_;
@@ -754,7 +760,8 @@ mod test {
                     }
                     count_cell.put_back(count);
                 }
-            }
+            };
+            assert!(res.is_ok());
 
             let client_thread = do Thread::start {
                 rtdebug!("starting client thread");
diff --git a/src/libstd/rt/uv/pipe.rs b/src/libstd/rt/uv/pipe.rs
index 1147c731a60..1cb86d4df2c 100644
--- a/src/libstd/rt/uv/pipe.rs
+++ b/src/libstd/rt/uv/pipe.rs
@@ -10,6 +10,7 @@
 
 use prelude::*;
 use libc;
+use c_str::CString;
 
 use rt::uv;
 use rt::uv::net;
@@ -37,6 +38,54 @@ impl Pipe {
         net::StreamWatcher(**self as *uvll::uv_stream_t)
     }
 
+    #[fixed_stack_segment] #[inline(never)]
+    pub fn open(&mut self, file: libc::c_int) -> Result<(), uv::UvError> {
+        match unsafe { uvll::uv_pipe_open(self.native_handle(), file) } {
+            0 => Ok(()),
+            n => Err(uv::UvError(n))
+        }
+    }
+
+    #[fixed_stack_segment] #[inline(never)]
+    pub fn bind(&mut self, name: &CString) -> Result<(), uv::UvError> {
+        do name.with_ref |name| {
+            match unsafe { uvll::uv_pipe_bind(self.native_handle(), name) } {
+                0 => Ok(()),
+                n => Err(uv::UvError(n))
+            }
+        }
+    }
+
+    #[fixed_stack_segment] #[inline(never)]
+    pub fn connect(&mut self, name: &CString, cb: uv::ConnectionCallback) {
+        {
+            let data = self.get_watcher_data();
+            assert!(data.connect_cb.is_none());
+            data.connect_cb = Some(cb);
+        }
+
+        let connect = net::ConnectRequest::new();
+        let name = do name.with_ref |p| { p };
+
+        unsafe {
+            uvll::uv_pipe_connect(connect.native_handle(),
+                                  self.native_handle(),
+                                  name,
+                                  connect_cb)
+        }
+
+        extern "C" fn connect_cb(req: *uvll::uv_connect_t, status: libc::c_int) {
+            let connect_request: net::ConnectRequest =
+                    uv::NativeHandle::from_native_handle(req);
+            let mut stream_watcher = connect_request.stream();
+            connect_request.delete();
+
+            let cb = stream_watcher.get_watcher_data().connect_cb.take_unwrap();
+            let status = uv::status_to_maybe_uv_error(status);
+            cb(stream_watcher, status);
+        }
+    }
+
     pub fn close(self, cb: uv::NullCallback) {
         {
             let mut this = self;
@@ -47,7 +96,7 @@ impl Pipe {
 
         unsafe { uvll::close(self.native_handle(), close_cb); }
 
-        extern fn close_cb(handle: *uvll::uv_pipe_t) {
+        extern "C" fn close_cb(handle: *uvll::uv_pipe_t) {
             let mut process: Pipe = uv::NativeHandle::from_native_handle(handle);
             process.get_watcher_data().close_cb.take_unwrap()();
             process.drop_watcher_data();
diff --git a/src/libstd/rt/uv/process.rs b/src/libstd/rt/uv/process.rs
index 176754de8f7..3c629a783cf 100644
--- a/src/libstd/rt/uv/process.rs
+++ b/src/libstd/rt/uv/process.rs
@@ -44,7 +44,7 @@ impl Process {
     /// occurred.
     pub fn spawn(&mut self, loop_: &uv::Loop, mut config: ProcessConfig,
                  exit_cb: uv::ExitCallback)
-                    -> Result<~[Option<UvPipeStream>], uv::UvError>
+                    -> Result<~[Option<~UvPipeStream>], uv::UvError>
     {
         let cwd = config.cwd.map(|s| s.to_c_str());
 
@@ -144,7 +144,7 @@ impl Process {
 }
 
 unsafe fn set_stdio(dst: *uvll::uv_stdio_container_t,
-                    io: StdioContainer) -> Option<UvPipeStream> {
+                    io: StdioContainer) -> Option<~UvPipeStream> {
     match io {
         Ignored => {
             uvll::set_stdio_container_flags(dst, uvll::STDIO_IGNORE);
@@ -166,7 +166,7 @@ unsafe fn set_stdio(dst: *uvll::uv_stdio_container_t,
             let handle = pipe.pipe.as_stream().native_handle();
             uvll::set_stdio_container_flags(dst, flags);
             uvll::set_stdio_container_stream(dst, handle);
-            Some(pipe.bind())
+            Some(~UvPipeStream::new(**pipe))
         }
     }
 }
diff --git a/src/libstd/rt/uv/uvio.rs b/src/libstd/rt/uv/uvio.rs
index 8dd0f8a6b10..6888aa23e99 100644
--- a/src/libstd/rt/uv/uvio.rs
+++ b/src/libstd/rt/uv/uvio.rs
@@ -746,11 +746,11 @@ impl IoFactory for UvIoFactory {
 
     fn pipe_init(&mut self, ipc: bool) -> Result<~RtioUnboundPipeObject, IoError> {
         let home = get_handle_to_current_scheduler!();
-        Ok(~UvUnboundPipe { pipe: Pipe::new(self.uv_loop(), ipc), home: home })
+        Ok(~UvUnboundPipe::new(Pipe::new(self.uv_loop(), ipc), home))
     }
 
     fn spawn(&mut self, config: ProcessConfig)
-            -> Result<(~RtioProcessObject, ~[Option<RtioPipeObject>]), IoError>
+            -> Result<(~RtioProcessObject, ~[Option<~RtioPipeObject>]), IoError>
     {
         // Sadly, we must create the UvProcess before we actually call uv_spawn
         // so that the exit_cb can close over it and notify it when the process
@@ -801,6 +801,74 @@ impl IoFactory for UvIoFactory {
             }
         }
     }
+
+    fn unix_bind<P: PathLike>(&mut self, path: &P) ->
+        Result<~RtioUnixListenerObject, IoError> {
+        let mut pipe = Pipe::new(self.uv_loop(), false);
+        match pipe.bind(&path.path_as_str(|s| s.to_c_str())) {
+            Ok(()) => {
+                let handle = get_handle_to_current_scheduler!();
+                let pipe = UvUnboundPipe::new(pipe, handle);
+                Ok(~UvUnixListener::new(pipe))
+            }
+            Err(e) => {
+                let scheduler: ~Scheduler = Local::take();
+                do scheduler.deschedule_running_task_and_then |_, task| {
+                    let task_cell = Cell::new(task);
+                    do pipe.close {
+                        let scheduler: ~Scheduler = Local::take();
+                        scheduler.resume_blocked_task_immediately(
+                            task_cell.take());
+                    }
+                }
+                Err(uv_error_to_io_error(e))
+            }
+        }
+    }
+
+    fn unix_connect<P: PathLike>(&mut self, path: &P) ->
+        Result<~RtioPipeObject, IoError>
+    {
+        let scheduler: ~Scheduler = Local::take();
+        let mut pipe = Pipe::new(self.uv_loop(), false);
+        let result_cell = Cell::new_empty();
+        let result_cell_ptr: *Cell<Result<~RtioPipeObject, IoError>> = &result_cell;
+
+        do scheduler.deschedule_running_task_and_then |_, task| {
+            let task_cell = Cell::new(task);
+            let cstr = do path.path_as_str |s| { s.to_c_str() };
+            do pipe.connect(&cstr) |stream, err| {
+                let res = match err {
+                    None => {
+                        let handle = stream.native_handle();
+                        let pipe = NativeHandle::from_native_handle(
+                                        handle as *uvll::uv_pipe_t);
+                        let home = get_handle_to_current_scheduler!();
+                        let pipe = UvUnboundPipe::new(pipe, home);
+                        Ok(~UvPipeStream::new(pipe))
+                    }
+                    Some(e) => { Err(uv_error_to_io_error(e)) }
+                };
+                unsafe { (*result_cell_ptr).put_back(res); }
+                let scheduler: ~Scheduler = Local::take();
+                scheduler.resume_blocked_task_immediately(task_cell.take());
+            }
+        }
+
+        assert!(!result_cell.is_empty());
+        let ret = result_cell.take();
+        if ret.is_err() {
+            let scheduler: ~Scheduler = Local::take();
+            do scheduler.deschedule_running_task_and_then |_, task| {
+                let task_cell = Cell::new(task);
+                do pipe.close {
+                    let scheduler: ~Scheduler = Local::take();
+                    scheduler.resume_blocked_task_immediately(task_cell.take());
+                }
+            }
+        }
+        return ret;
+    }
 }
 
 pub struct UvTcpListener {
@@ -843,9 +911,10 @@ impl RtioSocket for UvTcpListener {
 impl RtioTcpListener for UvTcpListener {
     fn listen(self) -> Result<~RtioTcpAcceptorObject, IoError> {
         do self.home_for_io_consume |self_| {
-            let mut acceptor = ~UvTcpAcceptor::new(self_);
+            let acceptor = ~UvTcpAcceptor::new(self_);
             let incoming = Cell::new(acceptor.incoming.clone());
-            do acceptor.listener.watcher.listen |mut server, status| {
+            let mut stream = acceptor.listener.watcher.as_stream();
+            let res = do stream.listen |mut server, status| {
                 do incoming.with_mut_ref |incoming| {
                     let inc = match status {
                         Some(_) => Err(standard_error(OtherIoError)),
@@ -860,7 +929,10 @@ impl RtioTcpListener for UvTcpListener {
                     incoming.send(inc);
                 }
             };
-            Ok(acceptor)
+            match res {
+                Ok(()) => Ok(acceptor),
+                Err(e) => Err(uv_error_to_io_error(e)),
+            }
         }
     }
 }
@@ -888,6 +960,17 @@ impl RtioSocket for UvTcpAcceptor {
     }
 }
 
+fn accept_simultaneously(stream: StreamWatcher, a: int) -> Result<(), IoError> {
+    let r = unsafe {
+        uvll::tcp_simultaneous_accepts(stream.native_handle(), a as c_int)
+    };
+
+    match status_to_maybe_uv_error(r) {
+        Some(err) => Err(uv_error_to_io_error(err)),
+        None => Ok(())
+    }
+}
+
 impl RtioTcpAcceptor for UvTcpAcceptor {
     fn accept(&mut self) -> Result<~RtioTcpStreamObject, IoError> {
         do self.home_for_io |self_| {
@@ -897,27 +980,13 @@ impl RtioTcpAcceptor for UvTcpAcceptor {
 
     fn accept_simultaneously(&mut self) -> Result<(), IoError> {
         do self.home_for_io |self_| {
-            let r = unsafe {
-                uvll::tcp_simultaneous_accepts(self_.listener.watcher.native_handle(), 1 as c_int)
-            };
-
-            match status_to_maybe_uv_error(r) {
-                Some(err) => Err(uv_error_to_io_error(err)),
-                None => Ok(())
-            }
+            accept_simultaneously(self_.listener.watcher.as_stream(), 1)
         }
     }
 
     fn dont_accept_simultaneously(&mut self) -> Result<(), IoError> {
         do self.home_for_io |self_| {
-            let r = unsafe {
-                uvll::tcp_simultaneous_accepts(self_.listener.watcher.native_handle(), 0 as c_int)
-            };
-
-            match status_to_maybe_uv_error(r) {
-                Some(err) => Err(uv_error_to_io_error(err)),
-                None => Ok(())
-            }
+            accept_simultaneously(self_.listener.watcher.as_stream(), 0)
         }
     }
 }
@@ -994,6 +1063,12 @@ pub struct UvUnboundPipe {
     priv home: SchedHandle,
 }
 
+impl UvUnboundPipe {
+    fn new(pipe: Pipe, home: SchedHandle) -> UvUnboundPipe {
+        UvUnboundPipe { pipe: pipe, home: home }
+    }
+}
+
 impl HomingIO for UvUnboundPipe {
     fn home<'r>(&'r mut self) -> &'r mut SchedHandle { &mut self.home }
 }
@@ -1013,18 +1088,12 @@ impl Drop for UvUnboundPipe {
     }
 }
 
-impl UvUnboundPipe {
-    pub unsafe fn bind(~self) -> UvPipeStream {
-        UvPipeStream { inner: self }
-    }
-}
-
 pub struct UvPipeStream {
-    priv inner: ~UvUnboundPipe,
+    priv inner: UvUnboundPipe,
 }
 
 impl UvPipeStream {
-    pub fn new(inner: ~UvUnboundPipe) -> UvPipeStream {
+    pub fn new(inner: UvUnboundPipe) -> UvPipeStream {
         UvPipeStream { inner: inner }
     }
 }
@@ -1612,6 +1681,84 @@ impl RtioProcess for UvProcess {
     }
 }
 
+pub struct UvUnixListener {
+    priv inner: UvUnboundPipe
+}
+
+impl HomingIO for UvUnixListener {
+    fn home<'r>(&'r mut self) -> &'r mut SchedHandle { self.inner.home() }
+}
+
+impl UvUnixListener {
+    fn new(pipe: UvUnboundPipe) -> UvUnixListener {
+        UvUnixListener { inner: pipe }
+    }
+}
+
+impl RtioUnixListener for UvUnixListener {
+    fn listen(self) -> Result<~RtioUnixAcceptorObject, IoError> {
+        do self.home_for_io_consume |self_| {
+            let acceptor = ~UvUnixAcceptor::new(self_);
+            let incoming = Cell::new(acceptor.incoming.clone());
+            let mut stream = acceptor.listener.inner.pipe.as_stream();
+            let res = do stream.listen |mut server, status| {
+                do incoming.with_mut_ref |incoming| {
+                    let inc = match status {
+                        Some(e) => Err(uv_error_to_io_error(e)),
+                        None => {
+                            let inc = Pipe::new(&server.event_loop(), false);
+                            server.accept(inc.as_stream());
+                            let home = get_handle_to_current_scheduler!();
+                            let pipe = UvUnboundPipe::new(inc, home);
+                            Ok(~UvPipeStream::new(pipe))
+                        }
+                    };
+                    incoming.send(inc);
+                }
+            };
+            match res {
+                Ok(()) => Ok(acceptor),
+                Err(e) => Err(uv_error_to_io_error(e)),
+            }
+        }
+    }
+}
+
+pub struct UvUnixAcceptor {
+    listener: UvUnixListener,
+    incoming: Tube<Result<~RtioPipeObject, IoError>>,
+}
+
+impl HomingIO for UvUnixAcceptor {
+    fn home<'r>(&'r mut self) -> &'r mut SchedHandle { self.listener.home() }
+}
+
+impl UvUnixAcceptor {
+    fn new(listener: UvUnixListener) -> UvUnixAcceptor {
+        UvUnixAcceptor { listener: listener, incoming: Tube::new() }
+    }
+}
+
+impl RtioUnixAcceptor for UvUnixAcceptor {
+    fn accept(&mut self) -> Result<~RtioPipeObject, IoError> {
+        do self.home_for_io |self_| {
+            self_.incoming.recv()
+        }
+    }
+
+    fn accept_simultaneously(&mut self) -> Result<(), IoError> {
+        do self.home_for_io |self_| {
+            accept_simultaneously(self_.listener.inner.pipe.as_stream(), 1)
+        }
+    }
+
+    fn dont_accept_simultaneously(&mut self) -> Result<(), IoError> {
+        do self.home_for_io |self_| {
+            accept_simultaneously(self_.listener.inner.pipe.as_stream(), 0)
+        }
+    }
+}
+
 #[test]
 fn test_simple_io_no_connect() {
     do run_in_mt_newsched_task {
diff --git a/src/libstd/rt/uv/uvll.rs b/src/libstd/rt/uv/uvll.rs
index 367585b0f0e..eb770d08070 100644
--- a/src/libstd/rt/uv/uvll.rs
+++ b/src/libstd/rt/uv/uvll.rs
@@ -1102,4 +1102,23 @@ extern {
     fn rust_set_stdio_container_stream(c: *uv_stdio_container_t,
                                        stream: *uv_stream_t);
     fn rust_uv_pipe_init(loop_ptr: *c_void, p: *uv_pipe_t, ipc: c_int) -> c_int;
+
+    pub fn uv_pipe_open(pipe: *uv_pipe_t, file: c_int) -> c_int;
+    pub fn uv_pipe_bind(pipe: *uv_pipe_t, name: *c_char) -> c_int;
+    pub fn uv_pipe_connect(req: *uv_connect_t, handle: *uv_pipe_t,
+                           name: *c_char, cb: uv_connect_cb);
+
+    // These should all really be constants...
+    #[rust_stack] pub fn rust_SOCK_STREAM() -> c_int;
+    #[rust_stack] pub fn rust_SOCK_DGRAM() -> c_int;
+    #[rust_stack] pub fn rust_SOCK_RAW() -> c_int;
+    #[rust_stack] pub fn rust_IPPROTO_UDP() -> c_int;
+    #[rust_stack] pub fn rust_IPPROTO_TCP() -> c_int;
+    #[rust_stack] pub fn rust_AI_ADDRCONFIG() -> c_int;
+    #[rust_stack] pub fn rust_AI_ALL() -> c_int;
+    #[rust_stack] pub fn rust_AI_CANONNAME() -> c_int;
+    #[rust_stack] pub fn rust_AI_NUMERICHOST() -> c_int;
+    #[rust_stack] pub fn rust_AI_NUMERICSERV() -> c_int;
+    #[rust_stack] pub fn rust_AI_PASSIVE() -> c_int;
+    #[rust_stack] pub fn rust_AI_V4MAPPED() -> c_int;
 }