From 0205b855d0794d882959e9147890e53c4ca6d2fe Mon Sep 17 00:00:00 2001 From: oberien Date: Fri, 23 Sep 2016 02:14:03 +0200 Subject: [PATCH 1/4] Add Chat example --- examples/chat.rs | 96 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 96 insertions(+) create mode 100644 examples/chat.rs diff --git a/examples/chat.rs b/examples/chat.rs new file mode 100644 index 000000000..4d43c4f96 --- /dev/null +++ b/examples/chat.rs @@ -0,0 +1,96 @@ +//! A chat server that broadcasts a message to all connections. + +extern crate tokio_core; +extern crate futures; + +use std::collections::HashMap; +use std::rc::Rc; +use std::cell::RefCell; +use std::iter; +use std::env; +use std::io::BufReader; + +use tokio_core::net::TcpListener; +use tokio_core::reactor::Core; +use tokio_core::io::{self, Io}; + +use futures::stream::{self, Stream}; +use futures::Future; + +fn main() { + let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string()); + let addr = addr.parse().unwrap(); + // We are single-threaded, so we can just use Rc and RefCell. + let connections = Rc::new(RefCell::new(HashMap::new())); + + let mut core = Core::new().unwrap(); + let handle = core.handle(); + let socket = TcpListener::bind(&addr, &handle).unwrap(); + println!("Listening on: {}", addr); + + let connections = connections.clone(); + let future = socket.incoming().for_each(move |(stream, addr)| { + let connections = connections.clone(); + let handle_inner = handle.clone(); + // We create a new future in which we create all other futures. + // This makes `stream` be bound on the `lazy` future's task, allowing + // `ReadHalf` and `WriteHalf` to be shared between inner futures. + handle.spawn_fn(move || { + println!("New Connection: {}", addr); + let (reader, writer) = stream.split(); + // channel to send messages to this connection from other futures + let (tx, rx) = tokio_core::channel::channel(&handle_inner).unwrap(); + // add sender to hashmap of all current connections + connections.borrow_mut().insert(addr, tx); + + let connections_inner = connections.clone(); + // https://users.rust-lang.org/t/loop-futures-for-client-handling/6950/2 + // We have an endless loop reading from a client. + // In order to fuse the reading and writing futures in the end, we need to have the same + // output type. Therefore we use `(Option>>, + // Option>)`. + let reader = BufReader::new(reader); + let socket_reader = stream::iter::<_, _, std::io::Error>(iter::repeat(()).map(Ok)).fold((Some(reader),None), move |(reader, _), _| { + let reader = reader.unwrap(); + let connections = connections_inner.clone(); + // read and parse length prefix + io::read_until(reader, '\n' as u8, vec![]) + .and_then(|(reader, vec)| futures::lazy(|| { + // EOF was hit without reading a delimiter + if vec.len() == 0 { + futures::failed((std::io::Error::new(std::io::ErrorKind::BrokenPipe, "Broken Pipe"))).boxed() + } else { + futures::finished((reader, vec)).boxed() + } + })) + // convert bytes into string + .map(|(reader, vec)| (reader, String::from_utf8(vec).unwrap())) + .and_then(move |(reader, message)| { + println!("{}: {:?}", addr, message); + // For each open connection except the sender, send the string via the channel + for tx in connections.borrow_mut().iter().filter(|&(&k,_)| k != addr).map(|(_,v)| v) { + tx.send(message.clone()).unwrap(); + } + futures::finished((Some(reader),None)) + }) + }); + + // Whenever we receive a string on the Receiver, we write it to `WriteHalf`. + let socket_writer = rx.fold((None, Some(writer)), move |(_, writer), msg| { + let writer = writer.unwrap(); + io::write_all(writer, msg.into_bytes()).map(|(writer, _)| (None, Some(writer))).boxed() + }); + + socket_reader.select(socket_writer) + .then(move |_| { + connections.borrow_mut().remove(&addr); + println!("Connection {:?} closed.", addr); + Ok(()) + }) + }); + Ok(()) + }); + + // exectue server + core.run(future).unwrap(); +} From 6961efa8dde54a429cfebc157594b25e5d25283f Mon Sep 17 00:00:00 2001 From: oberien Date: Thu, 6 Oct 2016 15:38:20 +0200 Subject: [PATCH 2/4] fix(chat): Implement alexcrichton's suggestions * Remove unnecessary clone * Improve rightward drift * Remove unnecessary lazy future * Improve utf-8 handling * Refactor to make code more understandable --- examples/chat.rs | 91 ++++++++++++++++++++++++++++-------------------- 1 file changed, 53 insertions(+), 38 deletions(-) diff --git a/examples/chat.rs b/examples/chat.rs index 4d43c4f96..beb1d78e3 100644 --- a/examples/chat.rs +++ b/examples/chat.rs @@ -8,7 +8,7 @@ use std::rc::Rc; use std::cell::RefCell; use std::iter; use std::env; -use std::io::BufReader; +use std::io::{Error, ErrorKind, BufReader}; use tokio_core::net::TcpListener; use tokio_core::reactor::Core; @@ -28,12 +28,11 @@ fn main() { let socket = TcpListener::bind(&addr, &handle).unwrap(); println!("Listening on: {}", addr); - let connections = connections.clone(); let future = socket.incoming().for_each(move |(stream, addr)| { let connections = connections.clone(); let handle_inner = handle.clone(); // We create a new future in which we create all other futures. - // This makes `stream` be bound on the `lazy` future's task, allowing + // This makes `stream` be bound on the outer future's task, allowing // `ReadHalf` and `WriteHalf` to be shared between inner futures. handle.spawn_fn(move || { println!("New Connection: {}", addr); @@ -43,50 +42,65 @@ fn main() { // add sender to hashmap of all current connections connections.borrow_mut().insert(addr, tx); + let reader = BufReader::new(reader); let connections_inner = connections.clone(); // https://users.rust-lang.org/t/loop-futures-for-client-handling/6950/2 - // We have an endless loop reading from a client. - // In order to fuse the reading and writing futures in the end, we need to have the same - // output type. Therefore we use `(Option>>, - // Option>)`. - let reader = BufReader::new(reader); - let socket_reader = stream::iter::<_, _, std::io::Error>(iter::repeat(()).map(Ok)).fold((Some(reader),None), move |(reader, _), _| { - let reader = reader.unwrap(); + // First we need to get an infinite iterator + let iter = stream::iter::<_, _, std::io::Error>(iter::repeat(()).map(Ok)); + // Then we fold it as infinite loop + let socket_reader = iter.fold(reader, move |reader, _| { let connections = connections_inner.clone(); - // read and parse length prefix - io::read_until(reader, '\n' as u8, vec![]) - .and_then(|(reader, vec)| futures::lazy(|| { - // EOF was hit without reading a delimiter - if vec.len() == 0 { - futures::failed((std::io::Error::new(std::io::ErrorKind::BrokenPipe, "Broken Pipe"))).boxed() - } else { - futures::finished((reader, vec)).boxed() + // read line + let amt = io::read_until(reader, '\n' as u8, vec![]); + // check if we hit EOF and need to close the connection + let amt = amt.and_then(|(reader, vec)| { + // EOF was hit without reading a delimiter + if vec.len() == 0 { + let err = Error::new(ErrorKind::BrokenPipe, "Broken Pipe"); + futures::failed(err).boxed() + } else { + futures::finished((reader, vec)).boxed() + } + }); + // convert bytes into string + let amt = amt.map(|(reader, vec)| (reader, String::from_utf8(vec))); + amt.and_then(move |(reader, message)| { + println!("{}: {:?}", addr, message); + let conns = connections.borrow_mut(); + if let Ok(msg) = message { + // For each open connection except the sender, send the string + // via the channel + let iter = conns.iter().filter(|&(&k,_)| k != addr).map(|(_,v)| v); + for tx in iter { + tx.send(msg.clone()).unwrap(); } - })) - // convert bytes into string - .map(|(reader, vec)| (reader, String::from_utf8(vec).unwrap())) - .and_then(move |(reader, message)| { - println!("{}: {:?}", addr, message); - // For each open connection except the sender, send the string via the channel - for tx in connections.borrow_mut().iter().filter(|&(&k,_)| k != addr).map(|(_,v)| v) { - tx.send(message.clone()).unwrap(); - } - futures::finished((Some(reader),None)) - }) + } else { + let tx = conns.get(&addr).unwrap(); + tx.send("You didn't send valid UTF-8.".to_string()).unwrap(); + } + futures::finished(reader) + }) }); // Whenever we receive a string on the Receiver, we write it to `WriteHalf`. - let socket_writer = rx.fold((None, Some(writer)), move |(_, writer), msg| { - let writer = writer.unwrap(); - io::write_all(writer, msg.into_bytes()).map(|(writer, _)| (None, Some(writer))).boxed() + let socket_writer = rx.fold(writer, |writer, msg| { + let amt = io::write_all(writer, msg.into_bytes()); + let amt = amt.map(|(writer, _)| writer); + amt }); - socket_reader.select(socket_writer) - .then(move |_| { - connections.borrow_mut().remove(&addr); - println!("Connection {:?} closed.", addr); - Ok(()) - }) + // In order to fuse the reading and writing futures in the end, we need to have the + // same output type. Therefore we use `(Option>>, + // Option>)`. + let socket_reader = socket_reader.map(|reader| (Some(reader), None)); + let socket_writer = socket_writer.map(|writer| (None, Some(writer))); + + let amt = socket_reader.select(socket_writer); + amt.then(move |_| { + connections.borrow_mut().remove(&addr); + println!("Connection {:?} closed.", addr); + Ok(()) + }) }); Ok(()) }); @@ -94,3 +108,4 @@ fn main() { // exectue server core.run(future).unwrap(); } + From b227738bd7e5347ece625e256cb2e9a90aa3cae2 Mon Sep 17 00:00:00 2001 From: oberien Date: Fri, 7 Oct 2016 14:57:30 +0200 Subject: [PATCH 3/4] ref(chat): Make code more readable * Send source address of message in addition to the message to connected clients. * Move `spawn_fn` to the bottom. * Use `map` instead of `and_then` if there is no need for blocking. * `map` to unit where values are not needed anymore. --- examples/chat.rs | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/examples/chat.rs b/examples/chat.rs index beb1d78e3..91f8c47b0 100644 --- a/examples/chat.rs +++ b/examples/chat.rs @@ -34,7 +34,7 @@ fn main() { // We create a new future in which we create all other futures. // This makes `stream` be bound on the outer future's task, allowing // `ReadHalf` and `WriteHalf` to be shared between inner futures. - handle.spawn_fn(move || { + let main_fn = move || { println!("New Connection: {}", addr); let (reader, writer) = stream.split(); // channel to send messages to this connection from other futures @@ -64,7 +64,7 @@ fn main() { }); // convert bytes into string let amt = amt.map(|(reader, vec)| (reader, String::from_utf8(vec))); - amt.and_then(move |(reader, message)| { + amt.map(move |(reader, message)| { println!("{}: {:?}", addr, message); let conns = connections.borrow_mut(); if let Ok(msg) = message { @@ -72,13 +72,13 @@ fn main() { // via the channel let iter = conns.iter().filter(|&(&k,_)| k != addr).map(|(_,v)| v); for tx in iter { - tx.send(msg.clone()).unwrap(); + tx.send(format!("{}: {}", addr, msg)).unwrap(); } } else { let tx = conns.get(&addr).unwrap(); tx.send("You didn't send valid UTF-8.".to_string()).unwrap(); } - futures::finished(reader) + reader }) }); @@ -90,18 +90,19 @@ fn main() { }); // In order to fuse the reading and writing futures in the end, we need to have the - // same output type. Therefore we use `(Option>>, - // Option>)`. - let socket_reader = socket_reader.map(|reader| (Some(reader), None)); - let socket_writer = socket_writer.map(|writer| (None, Some(writer))); + // same output type. As we don't need the values anymore, we can just map them + // to `()`. + let socket_reader = socket_reader.map(|_| ()); + let socket_writer = socket_writer.map(|_| ()); let amt = socket_reader.select(socket_writer); amt.then(move |_| { connections.borrow_mut().remove(&addr); - println!("Connection {:?} closed.", addr); + println!("Connection {} closed.", addr); Ok(()) }) - }); + }; + handle.spawn_fn(main_fn); Ok(()) }); From 315f6018229f38b34107096924e6f09a52527cac Mon Sep 17 00:00:00 2001 From: oberien Date: Tue, 11 Oct 2016 20:08:13 +0200 Subject: [PATCH 4/4] ref(examples): Minor refactoring in chat example * Move connections-clone down a bit * Use `Ok` and `Err` as IntoFuture --- examples/chat.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/examples/chat.rs b/examples/chat.rs index 91f8c47b0..18a09d8a7 100644 --- a/examples/chat.rs +++ b/examples/chat.rs @@ -49,7 +49,6 @@ fn main() { let iter = stream::iter::<_, _, std::io::Error>(iter::repeat(()).map(Ok)); // Then we fold it as infinite loop let socket_reader = iter.fold(reader, move |reader, _| { - let connections = connections_inner.clone(); // read line let amt = io::read_until(reader, '\n' as u8, vec![]); // check if we hit EOF and need to close the connection @@ -57,13 +56,14 @@ fn main() { // EOF was hit without reading a delimiter if vec.len() == 0 { let err = Error::new(ErrorKind::BrokenPipe, "Broken Pipe"); - futures::failed(err).boxed() + Err(err) } else { - futures::finished((reader, vec)).boxed() + Ok((reader, vec)) } }); // convert bytes into string let amt = amt.map(|(reader, vec)| (reader, String::from_utf8(vec))); + let connections = connections_inner.clone(); amt.map(move |(reader, message)| { println!("{}: {:?}", addr, message); let conns = connections.borrow_mut();