Async/await polish (#1058)

A general refresh of Tokio's experimental async / await support.

async-await/src/chat.rs

@@ -0,0 +1,131 @@
+#![feature(await_macro, async_await)]
+
+use tokio::await;
+use tokio::codec::{LinesCodec, Decoder};
+use tokio::net::{TcpListener, TcpStream};
+use tokio::prelude::*;
+
+use futures::sync::mpsc;
+
+use std::collections::HashMap;
+use std::io;
+use std::net::SocketAddr;
+use std::sync::{Arc, Mutex};
+
+/// Shorthand for the transmit half of the message channel.
+type Tx = mpsc::UnboundedSender<String>;
+
+struct Shared {
+    peers: HashMap<SocketAddr, Tx>,
+}
+
+impl Shared {
+    /// Create a new, empty, instance of `Shared`.
+    fn new() -> Self {
+        Shared {
+            peers: HashMap::new(),
+        }
+    }
+}
+
+async fn process(stream: TcpStream, state: Arc<Mutex<Shared>>) -> io::Result<()> {
+    let addr = stream.peer_addr().unwrap();
+    let mut lines = LinesCodec::new().framed(stream);
+
+    // Extract the peer's name
+    let name = match await!(lines.next()) {
+        Some(name) => name?,
+        None => {
+            // Disconnected early
+            return Ok(());
+        }
+    };
+
+    println!("`{}` is joining the chat", name);
+
+    let (tx, mut rx) = mpsc::unbounded();
+
+    // Register the socket
+    state.lock().unwrap()
+        .peers.insert(addr, tx);
+
+    // Split the `lines` handle into send and recv handles. This allows spawning
+    // separate tasks.
+    let (mut lines_tx, mut lines_rx) = lines.split();
+
+    // Spawn a task that receives all lines broadcasted to us from other peers
+    // and writes it to the client.
+    tokio::spawn_async(async move {
+        while let Some(line) = await!(rx.next()) {
+            let line = line.unwrap();
+            await!(lines_tx.send_async(line)).unwrap();
+        }
+    });
+
+    // Use the current task to read lines from the socket and broadcast them to
+    // other peers.
+    while let Some(message) = await!(lines_rx.next()) {
+        // TODO: Error handling
+        let message = message.unwrap();
+
+        let mut line = name.clone();
+        line.push_str(": ");
+        line.push_str(&message);
+        line.push_str("\r\n");
+
+        let state = state.lock().unwrap();
+
+        for (peer_addr, tx) in &state.peers {
+            if *peer_addr != addr {
+                // TODO: Error handling
+                tx.unbounded_send(line.clone()).unwrap();
+            }
+        }
+    }
+
+    // Remove the client from the shared state. Doing so will also result in the
+    // tx task to terminate.
+    state.lock().unwrap()
+        .peers.remove(&addr)
+        .expect("bug");
+
+    Ok(())
+}
+
+#[tokio::main]
+async fn main() {
+    // Create the shared state. This is how all the peers communicate.
+    //
+    // The server task will hold a handle to this. For every new client, the
+    // `state` handle is cloned and passed into the task that processes the
+    // client connection.
+    let state = Arc::new(Mutex::new(Shared::new()));
+
+    let addr = "127.0.0.1:6142".parse().unwrap();
+
+    // Bind a TCP listener to the socket address.
+    //
+    // Note that this is the Tokio TcpListener, which is fully async.
+    let listener = TcpListener::bind(&addr).unwrap();
+
+    println!("server running on localhost:6142");
+
+    // Start the Tokio runtime.
+    let mut incoming = listener.incoming();
+
+    while let Some(stream) = await!(incoming.next()) {
+        let stream = match stream {
+            Ok(stream) => stream,
+            Err(_) => continue,
+        };
+
+        let state = state.clone();
+
+        tokio::spawn_async(async move {
+            if let Err(_) = await!(process(stream, state)) {
+                eprintln!("failed to process connection");
+            }
+        });
+    }
+}
+

async-await/src/echo_client.rs

@@ -0,0 +1,50 @@
+#![feature(await_macro, async_await)]
+
+use tokio::await;
+use tokio::net::TcpStream;
+use tokio::prelude::*;
+
+use std::io;
+use std::net::SocketAddr;
+
+const MESSAGES: &[&str] = &[
+    "hello",
+    "world",
+    "one two three",
+];
+
+async fn run_client(addr: &SocketAddr) -> io::Result<()> {
+    let mut stream = await!(TcpStream::connect(addr))?;
+
+    // Buffer to read into
+    let mut buf = [0; 128];
+
+    for msg in MESSAGES {
+        println!(" > write = {:?}", msg);
+
+        // Write the message to the server
+        await!(stream.write_all_async(msg.as_bytes()))?;
+
+        // Read the message back from the server
+        await!(stream.read_exact_async(&mut buf[..msg.len()]))?;
+
+        assert_eq!(&buf[..msg.len()], msg.as_bytes());
+    }
+
+    Ok(())
+}
+
+#[tokio::main]
+async fn main() {
+    use std::env;
+
+    let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string());
+    let addr = addr.parse::<SocketAddr>().unwrap();
+
+    // Connect to the echo serveer
+
+    match await!(run_client(&addr)) {
+        Ok(_) => println!("done."),
+        Err(e) => eprintln!("echo client failed; error = {:?}", e),
+    }
+}

async-await/src/echo_server.rs

@@ -0,0 +1,42 @@
+#![feature(await_macro, async_await)]
+
+use tokio::await;
+use tokio::net::{TcpListener, TcpStream};
+use tokio::prelude::*;
+
+use std::net::SocketAddr;
+
+fn handle(mut stream: TcpStream) {
+    tokio::spawn_async(async move {
+        let mut buf = [0; 1024];
+
+        loop {
+            match await!(stream.read_async(&mut buf)).unwrap() {
+                0 => break, // Socket closed
+                n => {
+                    // Send the data back
+                    await!(stream.write_all_async(&buf[0..n])).unwrap();
+                }
+            }
+        }
+    });
+}
+
+#[tokio::main]
+async fn main() {
+    use std::env;
+
+    let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string());
+    let addr = addr.parse::<SocketAddr>().unwrap();
+
+    // Bind the TCP listener
+    let listener = TcpListener::bind(&addr).unwrap();
+    println!("Listening on: {}", addr);
+
+    let mut incoming = listener.incoming();
+
+    while let Some(stream) = await!(incoming.next()) {
+        let stream = stream.unwrap();
+        handle(stream);
+    }
+}

async-await/src/hyper.rs

@@ -0,0 +1,29 @@
+#![feature(await_macro, async_await)]
+
+use tokio::await;
+use tokio::prelude::*;
+use hyper::Client;
+
+use std::time::Duration;
+use std::str;
+
+#[tokio::main]
+async fn main() {
+    let client = Client::new();
+
+    let uri = "http://httpbin.org/ip".parse().unwrap();
+
+    let response = await!({
+        client.get(uri)
+            .timeout(Duration::from_secs(10))
+    }).unwrap();
+
+    println!("Response: {}", response.status());
+
+    let mut body = response.into_body();
+
+    while let Some(chunk) = await!(body.next()) {
+        let chunk = chunk.unwrap();
+        println!("chunk = {}", str::from_utf8(&chunk[..]).unwrap());
+    }
+}