Split out starting a listener from accepting incoming connections.

The Listener trait takes two type parameters, the type of connection and the type of Acceptor,
and specifies only one method, listen, which consumes the listener and produces an Acceptor.

The Acceptor trait takes one type parameter, the type of connection, and defines two methods.
The accept() method waits for an incoming connection attempt and returns the result.
The incoming() method creates an iterator over incoming connections and is a default method.

Example:

let listener = TcpListener.bind(addr); // Bind to a socket
let acceptor = listener.listen(); // Start the listener
for stream in acceptor.incoming() {
    // Process incoming connections forever (or until you break out of the loop)
}

1 files changed, 65 insertions, 51 deletions

diff --git a/src/libstd/rt/uv/uvio.rs b/src/libstd/rt/uv/uvio.rs
index ad9d2a59794..e37dfba0cc1 100644
--- a/src/libstd/rt/uv/uvio.rs
+++ b/src/libstd/rt/uv/uvio.rs
@@ -599,9 +599,7 @@ impl IoFactory for UvIoFactory {
 }
 
 pub struct UvTcpListener {
-    watcher: TcpWatcher,
-    listening: bool,
-    incoming_streams: Tube<Result<~RtioTcpStreamObject, IoError>>,
+    watcher : TcpWatcher,
     home: SchedHandle,
 }
 
@@ -611,15 +609,8 @@ impl HomingIO for UvTcpListener {
 
 impl UvTcpListener {
     fn new(watcher: TcpWatcher, home: SchedHandle) -> UvTcpListener {
-        UvTcpListener {
-            watcher: watcher,
-            listening: false,
-            incoming_streams: Tube::new(),
-            home: home,
-        }
+        UvTcpListener { watcher: watcher, home: home }
     }
-
-    fn watcher(&self) -> TcpWatcher { self.watcher }
 }
 
 impl Drop for UvTcpListener {
@@ -628,10 +619,10 @@ impl Drop for UvTcpListener {
         let self_ = unsafe { transmute::<&UvTcpListener, &mut UvTcpListener>(self) };
         do self_.home_for_io_with_sched |self_, scheduler| {
             do scheduler.deschedule_running_task_and_then |_, task| {
-                let task_cell = Cell::new(task);
-                do self_.watcher().as_stream().close {
+                let task = Cell::new(task);
+                do self_.watcher.as_stream().close {
                     let scheduler: ~Scheduler = Local::take();
-                    scheduler.resume_blocked_task_immediately(task_cell.take());
+                    scheduler.resume_blocked_task_immediately(task.take());
                 }
             }
         }
@@ -641,50 +632,71 @@ impl Drop for UvTcpListener {
 impl RtioSocket for UvTcpListener {
     fn socket_name(&mut self) -> Result<SocketAddr, IoError> {
         do self.home_for_io |self_| {
-          socket_name(Tcp, self_.watcher)
+            socket_name(Tcp, self_.watcher)
         }
     }
 }
 
 impl RtioTcpListener for UvTcpListener {
-
-    fn accept(&mut self) -> Result<~RtioTcpStreamObject, IoError> {
-        do self.home_for_io |self_| {
-
-            if !self_.listening {
-                self_.listening = true;
-
-                let incoming_streams_cell = Cell::new(self_.incoming_streams.clone());
-
-                do self_.watcher().listen |mut server, status| {
-                    let stream = match status {
+    fn listen(self) -> Result<~RtioTcpAcceptorObject, IoError> {
+        do self.home_for_io_consume |self_| {
+            let mut acceptor = ~UvTcpAcceptor::new(self_);
+            let incoming = Cell::new(acceptor.incoming.clone());
+            do acceptor.listener.watcher.listen |mut server, status| {
+                do incoming.with_mut_ref |incoming| {
+                    let inc = match status {
                         Some(_) => Err(standard_error(OtherIoError)),
                         None => {
-                            let client = TcpWatcher::new(&server.event_loop());
-                            // XXX: needs to be surfaced in interface
-                            server.accept(client.as_stream());
+                            let inc = TcpWatcher::new(&server.event_loop());
+                            // first accept call in the callback guarenteed to succeed
+                            server.accept(inc.as_stream());
                             let home = get_handle_to_current_scheduler!();
-                            Ok(~UvTcpStream { watcher: client, home: home })
+                            Ok(~UvTcpStream { watcher: inc, home: home })
                         }
                     };
-
-                    let mut incoming_streams = incoming_streams_cell.take();
-                    incoming_streams.send(stream);
-                    incoming_streams_cell.put_back(incoming_streams);
+                    incoming.send(inc);
                 }
+            };
+            Ok(acceptor)
+        }
+    }
+}
 
-            }
-            self_.incoming_streams.recv()
+pub struct UvTcpAcceptor {
+    listener: UvTcpListener,
+    incoming: Tube<Result<~RtioTcpStreamObject, IoError>>,
+}
+
+impl HomingIO for UvTcpAcceptor {
+    fn home<'r>(&'r mut self) -> &'r mut SchedHandle { self.listener.home() }
+}
+
+impl UvTcpAcceptor {
+    fn new(listener: UvTcpListener) -> UvTcpAcceptor {
+        UvTcpAcceptor { listener: listener, incoming: Tube::new() }
+    }
+}
+
+impl RtioSocket for UvTcpAcceptor {
+    fn socket_name(&mut self) -> Result<SocketAddr, IoError> {
+        do self.home_for_io |self_| {
+            socket_name(Tcp, self_.listener.watcher)
         }
     }
+}
+
+impl RtioTcpAcceptor for UvTcpAcceptor {
+    fn accept(&mut self) -> Result<~RtioTcpStreamObject, IoError> {
+        self.incoming.recv()
+    }
 
     fn accept_simultaneously(&mut self) -> Result<(), IoError> {
         do self.home_for_io |self_| {
             let r = unsafe {
-                uvll::tcp_simultaneous_accepts(self_.watcher().native_handle(), 1 as c_int)
+                uvll::tcp_simultaneous_accepts(self_.listener.watcher.native_handle(), 1 as c_int)
             };
 
-            match status_to_maybe_uv_error(self_.watcher(), r) {
+            match status_to_maybe_uv_error(self_.listener.watcher, r) {
                 Some(err) => Err(uv_error_to_io_error(err)),
                 None => Ok(())
             }
@@ -694,10 +706,10 @@ impl RtioTcpListener for UvTcpListener {
     fn dont_accept_simultaneously(&mut self) -> Result<(), IoError> {
         do self.home_for_io |self_| {
             let r = unsafe {
-                uvll::tcp_simultaneous_accepts(self_.watcher().native_handle(), 0 as c_int)
+                uvll::tcp_simultaneous_accepts(self_.listener.watcher.native_handle(), 0 as c_int)
             };
 
-            match status_to_maybe_uv_error(self_.watcher(), r) {
+            match status_to_maybe_uv_error(self_.listener.watcher, r) {
                 Some(err) => Err(uv_error_to_io_error(err)),
                 None => Ok(())
             }
@@ -1440,8 +1452,9 @@ fn test_simple_tcp_server_and_client() {
         do spawntask {
             unsafe {
                 let io: *mut IoFactoryObject = Local::unsafe_borrow();
-                let mut listener = (*io).tcp_bind(addr).unwrap();
-                let mut stream = listener.accept().unwrap();
+                let listener = (*io).tcp_bind(addr).unwrap();
+                let mut acceptor = listener.listen().unwrap();
+                let mut stream = acceptor.accept().unwrap();
                 let mut buf = [0, .. 2048];
                 let nread = stream.read(buf).unwrap();
                 assert_eq!(nread, 8);
@@ -1498,11 +1511,10 @@ fn test_simple_tcp_server_and_client_on_diff_threads() {
         };
 
         let server_fn: ~fn() = || {
-            let io: *mut IoFactoryObject = unsafe {
-                Local::unsafe_borrow()
-            };
-            let mut listener = unsafe { (*io).tcp_bind(server_addr).unwrap() };
-            let mut stream = listener.accept().unwrap();
+            let io: *mut IoFactoryObject = unsafe { Local::unsafe_borrow() };
+            let listener = unsafe { (*io).tcp_bind(server_addr).unwrap() };
+            let mut acceptor = listener.listen().unwrap();
+            let mut stream = acceptor.accept().unwrap();
             let mut buf = [0, .. 2048];
             let nread = stream.read(buf).unwrap();
             assert_eq!(nread, 8);
@@ -1583,8 +1595,9 @@ fn test_read_and_block() {
 
         do spawntask {
             let io: *mut IoFactoryObject = unsafe { Local::unsafe_borrow() };
-            let mut listener = unsafe { (*io).tcp_bind(addr).unwrap() };
-            let mut stream = listener.accept().unwrap();
+            let listener = unsafe { (*io).tcp_bind(addr).unwrap() };
+            let mut acceptor = listener.listen().unwrap();
+            let mut stream = acceptor.accept().unwrap();
             let mut buf = [0, .. 2048];
 
             let expected = 32;
@@ -1639,8 +1652,9 @@ fn test_read_read_read() {
         do spawntask {
             unsafe {
                 let io: *mut IoFactoryObject = Local::unsafe_borrow();
-                let mut listener = (*io).tcp_bind(addr).unwrap();
-                let mut stream = listener.accept().unwrap();
+                let listener = (*io).tcp_bind(addr).unwrap();
+                let mut acceptor = listener.listen().unwrap();
+                let mut stream = acceptor.accept().unwrap();
                 let buf = [1, .. 2048];
                 let mut total_bytes_written = 0;
                 while total_bytes_written < MAX {