Async/await polish (tokio-rs#1058)

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

Author: carllerche

.cirrus.yml

@@ -31,13 +31,13 @@ task:
     folder: $HOME/.cargo/registry
   test_script:
     - . $HOME/.cargo/env
-    - cargo test --all --no-fail-fast
+    - cargo test --all
     - (cd tokio-trace/test-log-support && cargo test)
     - (cd tokio-trace/test_static_max_level_features && cargo test)
     - cargo doc --all
   i686_test_script:
     - . $HOME/.cargo/env
     - |
-      cargo test --all --exclude tokio-tls --no-fail-fast --target i686-unknown-freebsd
+      cargo test --all --exclude tokio-tls --exclude tokio-macros --target i686-unknown-freebsd
   before_cache_script:
     - rm -rf $HOME/.cargo/registry/index

Cargo.toml

@@ -2,13 +2,14 @@
 
 members = [
   "tokio",
-  "tokio-async-await",
   "tokio-buf",
   "tokio-codec",
   "tokio-current-thread",
   "tokio-executor",
   "tokio-fs",
+  "tokio-futures",
   "tokio-io",
+  "tokio-macros",
   "tokio-reactor",
   "tokio-signal",
   "tokio-sync",

README.md

@@ -126,19 +126,21 @@ have greater guarantees of stability.
 
 The crates included as part of Tokio are:
 
-* [`tokio-async-await`]: Experimental `async` / `await` support.
-
-* [`tokio-codec`]: Utilities for encoding and decoding protocol frames.
-
 * [`tokio-current-thread`]: Schedule the execution of futures on the current
   thread.
 
 * [`tokio-executor`]: Task execution related traits and utilities.
 
 * [`tokio-fs`]: Filesystem (and standard in / out) APIs.
 
+* [`tokio-futures`]: Experimental `std::future::Future` and `async` / `await` support.
+
+* [`tokio-codec`]: Utilities for encoding and decoding protocol frames.
+
 * [`tokio-io`]: Asynchronous I/O related traits and utilities.
 
+* [`tokio-macros`]: Macros for usage with Tokio.
+
 * [`tokio-reactor`]: Event loop that drives I/O resources (like TCP and UDP
   sockets).
 
@@ -154,12 +156,13 @@ The crates included as part of Tokio are:
 * [`tokio-uds`]: Unix Domain Socket bindings for use with `tokio-io` and
   `tokio-reactor`.
 
-[`tokio-async-await`]: tokio-async-await
 [`tokio-codec`]: tokio-codec
 [`tokio-current-thread`]: tokio-current-thread
 [`tokio-executor`]: tokio-executor
 [`tokio-fs`]: tokio-fs
+[`tokio-futures`]: tokio-futures
 [`tokio-io`]: tokio-io
+[`tokio-macros`]: tokio-macros
 [`tokio-reactor`]: tokio-reactor
 [`tokio-tcp`]: tokio-tcp
 [`tokio-threadpool`]: tokio-threadpool

async-await/.cargo/config

@@ -0,0 +1,2 @@
+[build]
+target-dir = "../target"

async-await/Cargo.toml

@@ -0,0 +1,49 @@
+[package]
+name = "examples"
+edition = "2018"
+version = "0.1.0"
+authors = ["Carl Lerche <[email protected]>"]
+license = "MIT"
+
+# Break out of the parent workspace
+[workspace]
+
+[[bin]]
+name = "chat"
+path = "src/chat.rs"
+
+[[bin]]
+name = "echo_client"
+path = "src/echo_client.rs"
+
+[[bin]]
+name = "echo_server"
+path = "src/echo_server.rs"
+
+[[bin]]
+name = "hyper"
+path = "src/hyper.rs"
+
+[dependencies]
+tokio = { version = "0.1.18", features = ["async-await-preview"] }
+futures = "0.1.23"
+bytes = "0.4.9"
+hyper = "0.12.8"
+
+# Avoid using crates.io for Tokio dependencies
+[patch.crates-io]
+tokio = { path = "../tokio" }
+tokio-codec = { path = "../tokio-codec" }
+tokio-current-thread = { path = "../tokio-current-thread" }
+tokio-executor = { path = "../tokio-executor" }
+tokio-fs = { path = "../tokio-fs" }
+# tokio-futures  = { path = "../" }
+tokio-io = { path = "../tokio-io" }
+tokio-reactor = { path = "../tokio-reactor" }
+tokio-signal = { path = "../tokio-signal" }
+tokio-tcp = { path = "../tokio-tcp" }
+tokio-threadpool = { path = "../tokio-threadpool" }
+tokio-timer = { path = "../tokio-timer" }
+tokio-tls = { path = "../tokio-tls" }
+tokio-udp = { path = "../tokio-udp" }
+tokio-uds = { path = "../tokio-uds" }

tokio-async-await/examples/README.md async-await/README.md

@@ -1,9 +1,6 @@
-#![feature(await_macro, async_await, futures_api)]
-
-#[macro_use]
-extern crate tokio;
-extern crate futures; // v0.1
+#![feature(await_macro, async_await)]
 
+use tokio::await;
 use tokio::codec::{LinesCodec, Decoder};
 use tokio::net::{TcpListener, TcpStream};
 use tokio::prelude::*;
@@ -95,7 +92,8 @@ async fn process(stream: TcpStream, state: Arc<Mutex<Shared>>) -> io::Result<()>
     Ok(())
 }
 
-fn main() {
+#[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
@@ -113,23 +111,21 @@ fn main() {
     println!("server running on localhost:6142");
 
     // Start the Tokio runtime.
-    tokio::run_async(async move {
-        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");
-                }
-            });
-        }
-    });
+    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");
+            }
+        });
+    }
 }
 

tokio-async-await/examples/src/chat.rs async-await/src/chat.rs

@@ -1,8 +1,6 @@
-#![feature(await_macro, async_await, futures_api)]
-
-#[macro_use]
-extern crate tokio;
+#![feature(await_macro, async_await)]
 
+use tokio::await;
 use tokio::net::TcpStream;
 use tokio::prelude::*;
 
@@ -36,18 +34,17 @@ async fn run_client(addr: &SocketAddr) -> io::Result<()> {
     Ok(())
 }
 
-fn main() {
+#[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
 
-    tokio::run_async(async move {
-        match await!(run_client(&addr)) {
-            Ok(_) => println!("done."),
-            Err(e) => eprintln!("echo client failed; error = {:?}", e),
-        }
-    });
+    match await!(run_client(&addr)) {
+        Ok(_) => println!("done."),
+        Err(e) => eprintln!("echo client failed; error = {:?}", e),
+    }
 }

tokio-async-await/examples/src/echo_client.rs async-await/src/echo_client.rs

@@ -1,8 +1,6 @@
-#![feature(await_macro, async_await, futures_api)]
-
-#[macro_use]
-extern crate tokio;
+#![feature(await_macro, async_await)]
 
+use tokio::await;
 use tokio::net::{TcpListener, TcpStream};
 use tokio::prelude::*;
 
@@ -24,7 +22,8 @@ fn handle(mut stream: TcpStream) {
     });
 }
 
-fn main() {
+#[tokio::main]
+async fn main() {
     use std::env;
 
     let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string());
@@ -34,12 +33,10 @@ fn main() {
     let listener = TcpListener::bind(&addr).unwrap();
     println!("Listening on: {}", addr);
 
-    tokio::run_async(async {
-        let mut incoming = listener.incoming();
+    let mut incoming = listener.incoming();
 
-        while let Some(stream) = await!(incoming.next()) {
-            let stream = stream.unwrap();
-            handle(stream);
-        }
-    });
+    while let Some(stream) = await!(incoming.next()) {
+        let stream = stream.unwrap();
+        handle(stream);
+    }
 }

tokio-async-await/examples/src/echo_server.rs async-await/src/echo_server.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());
+    }
+}

async-await/src/hyper.rs

@@ -76,7 +76,7 @@ jobs:
   parameters:
     name: async_await
     displayName: Async / Await
-    rust: nightly-2019-04-22
+    rust: nightly-2019-04-25
     noDefaultFeatures: ''
     benches: true
     crates:

azure-pipelines.yml

@@ -2,12 +2,12 @@
 # repository.
 [patch.crates-io]
 tokio = { path = "tokio" }
-tokio-async-await = { path = "tokio-async-await" }
 tokio-buf = { path = "tokio-buf" }
 tokio-codec = { path = "tokio-codec" }
 tokio-current-thread = { path = "tokio-current-thread" }
 tokio-executor = { path = "tokio-executor" }
 tokio-fs = { path = "tokio-fs" }
+tokio-futures = { path = "tokio-futures" }
 tokio-io = { path = "tokio-io" }
 tokio-reactor = { path = "tokio-reactor" }
 tokio-signal = { path = "tokio-signal" }