diff --git a/examples/chat.rs b/examples/chat.rs index d141f62be..76e689b95 100644 --- a/examples/chat.rs +++ b/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(); diff --git a/examples/compress.rs b/examples/compress.rs index f0c2e23f4..3098abf74 100644 --- a/examples/compress.rs +++ b/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), diff --git a/examples/echo-threads.rs b/examples/echo-threads.rs index 8ac42d0f9..6ce8b1563 100644 --- a/examples/echo-threads.rs +++ b/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(()) diff --git a/examples/echo.rs b/examples/echo.rs index f72191134..558f3a68a 100644 --- a/examples/echo.rs +++ b/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(()) diff --git a/examples/hello.rs b/examples/hello.rs index 46b36b5d0..5ceb431b1 100644 --- a/examples/hello.rs +++ b/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)| { diff --git a/examples/proxy.rs b/examples/proxy.rs index c64ead6f2..98b86e9fe 100644 --- a/examples/proxy.rs +++ b/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); diff --git a/examples/sink.rs b/examples/sink.rs index 82bde294e..21456adae 100644 --- a/examples/sink.rs +++ b/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(()) }); diff --git a/examples/tinydb.rs b/examples/tinydb.rs index 19a7396d7..0a68a3146 100644 --- a/examples/tinydb.rs +++ b/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. diff --git a/src/lib.rs b/src/lib.rs index b0360c540..e7d308201 100644 --- a/src/lib.rs +++ b/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(); diff --git a/src/net/tcp.rs b/src/net/tcp.rs index 352e9db90..82dabdb73 100644 --- a/src/net/tcp.rs +++ b/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, io::Error> { - Ok(Async::Ready(Some(try_nb!(self.inner.accept())))) + let (socket, _) = try_nb!(self.inner.accept()); + Ok(Async::Ready(Some(socket))) } } diff --git a/tests/buffered.rs b/tests/buffered.rs index 67f109f99..2ba16b042 100644 --- a/tests/buffered.rs +++ b/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()); diff --git a/tests/chain.rs b/tests/chain.rs index c161b294d..b9ac48186 100644 --- a/tests/chain.rs +++ b/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(); diff --git a/tests/echo.rs b/tests/echo.rs index c96947ccf..d5bdae811 100644 --- a/tests/echo.rs +++ b/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()); diff --git a/tests/global.rs b/tests/global.rs index 1c15365fe..bf5682fa0 100644 --- a/tests/global.rs +++ b/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()); diff --git a/tests/limit.rs b/tests/limit.rs index 7e95c71f7..7055ce9b6 100644 --- a/tests/limit.rs +++ b/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(); diff --git a/tests/line-frames.rs b/tests/line-frames.rs index 2a8dc081e..3785dfeff 100644 --- a/tests/line-frames.rs +++ b/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(()) diff --git a/tests/stream-buffered.rs b/tests/stream-buffered.rs index 9b26b6ace..78fe1c372 100644 --- a/tests/stream-buffered.rs +++ b/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) diff --git a/tests/tcp.rs b/tests/tcp.rs index 2056dba6d..83cc425c1 100644 --- a/tests/tcp.rs +++ b/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());