Change net::Incoming signature to match std. (#89)

std's `Incoming` iterator yields `TcpStream` instances. This patch
updates the `Incoming` future to match this signature.

This changes the yielded value from `(TcpStream, SocketAddr)` ->
`TcpStream`.

examples/chat.rs

@@ -49,7 +49,9 @@ fn main() {
     // Once the same thread executor lands, transition to single threaded.
     let connections = Arc::new(Mutex::new(HashMap::new()));
 
-    let srv = socket.incoming().for_each(move |(stream, addr)| {
+    let srv = socket.incoming().for_each(move |stream| {
+        let addr = stream.peer_addr().unwrap();
+
         println!("New Connection: {}", addr);
         let (reader, writer) = stream.split();
 

examples/compress.rs

@@ -50,7 +50,8 @@ fn main() {
 
     // The compress logic will happen in the function below, but everything's
     // still a future! Each client is spawned to concurrently get processed.
-    let server = socket.incoming().for_each(move |(socket, addr)| {
+    let server = socket.incoming().for_each(move |socket| {
+        let addr = socket.peer_addr().unwrap();
         pool.execute(compress(socket, &pool).then(move |result| {
             match result {
                 Ok((r, w)) => println!("{}: compressed {} bytes to {}", addr, r, w),

examples/echo-threads.rs

@@ -56,7 +56,7 @@ fn main() {
     // shipped round-robin to a particular thread which will associate the
     // socket with the corresponding event loop and process the connection.
     let mut next = 0;
-    let srv = listener.incoming().for_each(|(socket, _)| {
+    let srv = listener.incoming().for_each(|socket| {
         channels[next].unbounded_send(socket).expect("worker thread died");
         next = (next + 1) % channels.len();
         Ok(())

examples/echo.rs

@@ -60,12 +60,11 @@ fn main() {
     // connections made to the server).  The return value of the `for_each`
     // method is itself a future representing processing the entire stream of
     // connections, and ends up being our server.
-    let done = socket.incoming().for_each(move |(socket, addr)| {
+    let done = socket.incoming().for_each(move |socket| {
 
         // Once we're inside this closure this represents an accepted client
-        // from our server. The `socket` is the client connection and `addr` is
-        // the remote address of the client (similar to how the standard library
-        // operates).
+        // from our server. The `socket` is the client connection (similar to
+        // how the standard library operates).
         //
         // We just want to copy all data read from the socket back onto the
         // socket itself (e.g. "echo"). We can use the standard `io::copy`
@@ -88,8 +87,8 @@ fn main() {
         // information.
         let msg = amt.then(move |result| {
             match result {
-                Ok((amt, _, _)) => println!("wrote {} bytes to {}", amt, addr),
-                Err(e) => println!("error on {}: {}", addr, e),
+                Ok((amt, _, _)) => println!("wrote {} bytes", amt),
+                Err(e) => println!("error: {}", e),
             }
 
             Ok(())

examples/hello.rs

@@ -33,7 +33,7 @@ fn main() {
     println!("Listening for connections on {}", addr);
 
     let clients = listener.incoming();
-    let welcomes = clients.and_then(|(socket, _peer_addr)| {
+    let welcomes = clients.and_then(|socket| {
         tokio_io::io::write_all(socket, b"Hello!\n")
     });
     let server = welcomes.for_each(|(_socket, _welcome)| {

examples/proxy.rs

@@ -48,7 +48,7 @@ fn main() {
     println!("Listening on: {}", listen_addr);
     println!("Proxying to: {}", server_addr);
 
-    let done = socket.incoming().for_each(move |(client, client_addr)| {
+    let done = socket.incoming().for_each(move |client| {
         let server = TcpStream::connect(&server_addr);
         let amounts = server.and_then(move |server| {
             // Create separate read/write handles for the TCP clients that we're
@@ -82,8 +82,8 @@ fn main() {
         });
 
         let msg = amounts.map(move |(from_client, from_server)| {
-            println!("client at {} wrote {} bytes and received {} bytes",
-                     client_addr, from_client, from_server);
+            println!("client wrote {} bytes and received {} bytes",
+                     from_client, from_server);
         }).map_err(|e| {
             // Don't panic. Maybe the client just disconnected too soon.
             println!("error: {}", e);

examples/sink.rs

@@ -41,8 +41,8 @@ fn main() {
 
     let socket = TcpListener::bind(&addr).unwrap();
     println!("Listening on: {}", addr);
-    let server = socket.incoming().for_each(|(socket, addr)| {
-        println!("got a socket: {}", addr);
+    let server = socket.incoming().for_each(|socket| {
+        println!("got a socket: {}", socket.peer_addr().unwrap());
         pool.execute(write(socket).or_else(|_| Ok(()))).unwrap();
         Ok(())
     });

examples/tinydb.rs

@@ -100,7 +100,7 @@ fn main() {
         map: Mutex::new(initial_db),
     });
 
-    let done = listener.incoming().for_each(move |(socket, _addr)| {
+    let done = listener.incoming().for_each(move |socket| {
         // As with many other small examples, the first thing we'll do is
         // *split* this TCP stream into two separately owned halves. This'll
         // allow us to work with the read and write halves independently.

src/lib.rs

@@ -59,7 +59,7 @@
 //!         .expect("unable to bind TCP listener");
 //!
 //!     // Pull out a stream of sockets for incoming connections
-//!     let server = listener.incoming().for_each(|(sock, _)| {
+//!     let server = listener.incoming().for_each(|sock| {
 //!         // Split up the reading and writing parts of the
 //!         // socket.
 //!         let (reader, writer) = sock.split();

src/net/tcp.rs

@@ -182,11 +182,12 @@ impl fmt::Debug for TcpListener {
 }
 
 impl Stream for Incoming {
-    type Item = (TcpStream, SocketAddr);
+    type Item = TcpStream;
     type Error = io::Error;
 
     fn poll(&mut self) -> Poll<Option<Self::Item>, io::Error> {
-        Ok(Async::Ready(Some(try_nb!(self.inner.accept()))))
+        let (socket, _) = try_nb!(self.inner.accept());
+        Ok(Async::Ready(Some(socket)))
     }
 }
 

tests/buffered.rs

@@ -46,7 +46,7 @@ fn echo_server() {
         (expected, t2)
     });
 
-    let clients = srv.incoming().take(2).map(|e| e.0).collect();
+    let clients = srv.incoming().take(2).collect();
     let copied = clients.and_then(|clients| {
         let mut clients = clients.into_iter();
         let a = BufReader::new(clients.next().unwrap());

tests/chain.rs

@@ -32,7 +32,7 @@ fn chain_clients() {
         s3.write_all(b"baz").unwrap();
     });
 
-    let clients = srv.incoming().map(|e| e.0).take(3);
+    let clients = srv.incoming().take(3);
     let copied = clients.collect().and_then(|clients| {
         let mut clients = clients.into_iter();
         let a = clients.next().unwrap();

tests/echo.rs

@@ -41,7 +41,7 @@ fn echo_server() {
 
     let clients = srv.incoming();
     let client = clients.into_future().map(|e| e.0.unwrap()).map_err(|e| e.0);
-    let halves = client.map(|s| s.0.split());
+    let halves = client.map(|s| s.split());
     let copied = halves.and_then(|(a, b)| copy(a, b));
 
     let (amt, _, _) = t!(copied.wait());

tests/global.rs

@@ -21,7 +21,7 @@ fn hammer() {
             let addr = t!(srv.local_addr());
             let mine = TcpStream::connect(&addr);
             let theirs = srv.incoming().into_future()
-                .map(|(s, _)| s.unwrap().0)
+                .map(|(s, _)| s.unwrap())
                 .map_err(|(s, _)| s);
             let (mine, theirs) = t!(mine.join(theirs).wait());
 

tests/limit.rs

@@ -28,7 +28,7 @@ fn limit() {
         s1.write_all(b"foo bar baz").unwrap();
     });
 
-    let clients = srv.incoming().map(|e| e.0).take(1);
+    let clients = srv.incoming().take(1);
     let copied = clients.collect().and_then(|clients| {
         let mut clients = clients.into_iter();
         let a = clients.next().unwrap();

tests/line-frames.rs

@@ -59,7 +59,7 @@ fn echo() {
     let listener = TcpListener::bind(&"127.0.0.1:0".parse().unwrap()).unwrap();
     let addr = listener.local_addr().unwrap();
     let pool_inner = pool.clone();
-    let srv = listener.incoming().for_each(move |(socket, _)| {
+    let srv = listener.incoming().for_each(move |socket| {
         let (sink, stream) = socket.framed(LineCodec).split();
         pool_inner.execute(sink.send_all(stream).map(|_| ()).map_err(|_| ())).unwrap();
         Ok(())

tests/stream-buffered.rs

@@ -42,7 +42,7 @@ fn echo_server() {
     });
 
     let future = srv.incoming()
-                    .map(|s| s.0.split())
+                    .map(|s| s.split())
                     .map(|(a, b)| copy(a, b).map(|_| ()))
                     .buffered(10)
                     .take(2)

tests/tcp.rs

@@ -43,7 +43,7 @@ fn accept() {
     let (tx, rx) = channel();
     let client = srv.incoming().map(move |t| {
         tx.send(()).unwrap();
-        t.0
+        t
     }).into_future().map_err(|e| e.0);
     assert!(rx.try_recv().is_err());
     let t = thread::spawn(move || {
@@ -71,7 +71,7 @@ fn accept2() {
     let (tx, rx) = channel();
     let client = srv.incoming().map(move |t| {
         tx.send(()).unwrap();
-        t.0
+        t
     }).into_future().map_err(|e| e.0);
     assert!(rx.try_recv().is_err());