tokio: rewrite examples with async. (#1228)

tokio/examples/echo-udp.rs

@@ -10,14 +10,13 @@
 //!
 //! Each line you type in to the `nc` terminal should be echo'd back to you!
 
+#![feature(async_await)]
 #![deny(warnings, rust_2018_idioms)]
 
-use futures::try_ready;
 use std::net::SocketAddr;
 use std::{env, io};
 use tokio;
 use tokio::net::UdpSocket;
-use tokio::prelude::*;
 
 struct Server {
     socket: UdpSocket,
@@ -25,29 +24,33 @@ struct Server {
     to_send: Option<(usize, SocketAddr)>,
 }
 
-impl Future for Server {
-    type Item = ();
-    type Error = io::Error;
+impl Server {
+    async fn run(self) -> Result<(), io::Error> {
+        let Server {
+            mut socket,
+            mut buf,
+            mut to_send,
+        } = self;
 
-    fn poll(&mut self) -> Poll<(), io::Error> {
         loop {
             // First we check to see if there's a message we need to echo back.
             // If so then we try to send it back to the original source, waiting
             // until it's writable and we're able to do so.
-            if let Some((size, peer)) = self.to_send {
-                let amt = try_ready!(self.socket.poll_send_to(&self.buf[..size], &peer));
+            if let Some((size, peer)) = to_send {
+                let amt = socket.send_to(&buf[..size], &peer).await?;
+
                 println!("Echoed {}/{} bytes to {}", amt, size, peer);
-                self.to_send = None;
             }
 
             // If we're here then `to_send` is `None`, so we take a look for the
             // next message we're going to echo back.
-            self.to_send = Some(try_ready!(self.socket.poll_recv_from(&mut self.buf)));
+            to_send = Some(socket.recv_from(&mut buf).await?);
         }
     }
 }
 
-fn main() -> Result<(), Box<dyn std::error::Error>> {
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string());
     let addr = addr.parse::<SocketAddr>()?;
 
@@ -61,11 +64,6 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     };
 
     // This starts the server task.
-    //
-    // `map_err` handles the error by logging it and maps the future to a type
-    // that can be spawned.
-    //
-    // `tokio::run` spawns the task on the Tokio runtime and starts running.
-    tokio::run(server.map_err(|e| println!("server error = {:?}", e)));
+    server.run().await?;
     Ok(())
 }

tokio/examples/hello_world.rs

@@ -0,0 +1,34 @@
+//! Hello world server.
+//!
+//! A simple client that opens a TCP stream, writes "hello world\n", and closes
+//! the connection.
+//!
+//! You can test this out by running:
+//!
+//!     ncat -l 6142
+//!
+//! And then in another terminal run:
+//!
+//!     cargo run --example hello_world
+
+#![deny(warnings, rust_2018_idioms)]
+#![feature(async_await)]
+
+use tokio;
+use tokio::io::AsyncWriteExt;
+use tokio::net::TcpStream;
+
+#[tokio::main]
+pub async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let addr = "127.0.0.1:6142".parse()?;
+
+    // Open a TCP stream to the socket address.
+    //
+    // Note that this is the Tokio TcpStream, which is fully async.
+    let mut stream = TcpStream::connect(&addr).await?;
+    println!("created stream");
+    let result = stream.write(b"hello world\n").await;
+    println!("wrote to stream; success={:?}", result.is_ok());
+
+    Ok(())
+}

tokio/examples/udp-client.rs

@@ -26,13 +26,13 @@
 //! Please mind that since the UDP protocol doesn't have any capabilities to detect a broken
 //! connection the server needs to be run first, otherwise the client will block forever.
 
+#![feature(async_await)]
 #![deny(warnings, rust_2018_idioms)]
 
 use std::env;
-use std::io::stdin;
+use std::io::{stdin, Read};
 use std::net::SocketAddr;
 use tokio::net::UdpSocket;
-use tokio::prelude::*;
 
 fn get_stdin_data() -> Result<Vec<u8>, Box<dyn std::error::Error>> {
     let mut buf = Vec::new();
@@ -40,7 +40,8 @@ fn get_stdin_data() -> Result<Vec<u8>, Box<dyn std::error::Error>> {
     Ok(buf)
 }
 
-fn main() -> Result<(), Box<dyn std::error::Error>> {
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let remote_addr: SocketAddr = env::args()
         .nth(1)
         .unwrap_or("127.0.0.1:8080".into())
@@ -52,18 +53,17 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         "[::]:0"
     }
     .parse()?;
-    let socket = UdpSocket::bind(&local_addr)?;
+    let mut socket = UdpSocket::bind(&local_addr)?;
     const MAX_DATAGRAM_SIZE: usize = 65_507;
-    socket
-        .send_dgram(get_stdin_data()?, &remote_addr)
-        .and_then(|(socket, _)| socket.recv_dgram(vec![0u8; MAX_DATAGRAM_SIZE]))
-        .map(|(_, data, len, _)| {
-            println!(
-                "Received {} bytes:\n{}",
-                len,
-                String::from_utf8_lossy(&data[..len])
-            )
-        })
-        .wait()?;
+    socket.connect(&remote_addr)?;
+    let data = get_stdin_data()?;
+    socket.send(&data).await?;
+    let mut data = vec![0u8; MAX_DATAGRAM_SIZE];
+    let len = socket.recv(&mut data).await?;
+    println!(
+        "Received {} bytes:\n{}",
+        len,
+        String::from_utf8_lossy(&data[..len])
+    );
     Ok(())
 }

tokio/examples_old/blocking.rs

@@ -9,8 +9,10 @@
 //! Note how non-blocking threads are executed before blocking threads finish
 //! their task.
 
+#![feature(async_await)]
 #![deny(warnings, rust_2018_idioms)]
 
+use std::pin::Pin;
 use std::thread;
 use std::time::Duration;
 use tokio;
@@ -24,18 +26,19 @@ struct BlockingFuture {
 }
 
 impl Future for BlockingFuture {
-    type Item = ();
-    type Error = ();
+    type Output = ();
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+    fn poll(self: Pin<&mut Self>, _ctx: &mut task::Context<'_>) -> Poll<Self::Output> {
         println!("Blocking begin: {}!", self.value);
         // Try replacing this part with commnted code
         blocking(|| {
             println!("Blocking part annotated: {}!", self.value);
             thread::sleep(Duration::from_millis(1000));
             println!("Blocking done annotated: {}!", self.value);
+        }).map(|result| match result {
+            Ok(result) => result,
+            Err(err) => panic!("Error in blocing block: {:?}", err),
         })
-        .map_err(|err| panic!("Error in blocing block: {:?}", err))
         // println!("Blocking part annotated: {}!", self.value);
         // thread::sleep(Duration::from_millis(1000));
         // println!("Blocking done annotated: {}!", self.value);
@@ -49,12 +52,11 @@ struct NonBlockingFuture {
 }
 
 impl Future for NonBlockingFuture {
-    type Item = ();
-    type Error = ();
+    type Output = ();
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+    fn poll(self: Pin<&mut Self>, _ctx: &mut task::Context<'_>) -> Poll<Self::Output> {
         println!("Non-blocking done: {}!", self.value);
-        Ok(Async::Ready(()))
+        Poll::Ready(())
     }
 }
 
@@ -62,10 +64,9 @@ impl Future for NonBlockingFuture {
 struct SpawningFuture;
 
 impl Future for SpawningFuture {
-    type Item = ();
-    type Error = ();
+    type Output = ();
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+    fn poll(self: Pin<&mut Self>, _ctx: &mut task::Context<'_>) -> Poll<Self::Output> {
         for i in 0..8 {
             let blocking_future = BlockingFuture { value: i };
 
@@ -75,13 +76,15 @@ impl Future for SpawningFuture {
             let non_blocking_future = NonBlockingFuture { value: i };
             tokio::spawn(non_blocking_future);
         }
-        Ok(Async::Ready(()))
+        Poll::Ready(())
     }
 }
 
 fn main() {
     let spawning_future = SpawningFuture;
 
-    let runtime = Builder::new().core_threads(4).build().unwrap();
-    runtime.block_on_all(spawning_future).unwrap();
+    let mut runtime = Builder::new()
+        .core_threads(4)
+        .build().unwrap();
+    runtime.block_on_all(spawning_future);
 }

tokio/examples_old/hello_world.rs

@@ -1,57 +0,0 @@
-//! Hello world server.
-//!
-//! A simple client that opens a TCP stream, writes "hello world\n", and closes
-//! the connection.
-//!
-//! You can test this out by running:
-//!
-//!     ncat -l 6142
-//!
-//! And then in another terminal run:
-//!
-//!     cargo run --example hello_world
-
-#![deny(warnings, rust_2018_idioms)]
-
-use tokio;
-use tokio::io;
-use tokio::net::TcpStream;
-use tokio::prelude::*;
-
-pub fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let addr = "127.0.0.1:6142".parse()?;
-
-    // Open a TCP stream to the socket address.
-    //
-    // Note that this is the Tokio TcpStream, which is fully async.
-    let client = TcpStream::connect(&addr)
-        .and_then(|stream| {
-            println!("created stream");
-            io::write_all(stream, "hello world\n").then(|result| {
-                println!("wrote to stream; success={:?}", result.is_ok());
-                Ok(())
-            })
-        })
-        .map_err(|err| {
-            // All tasks must have an `Error` type of `()`. This forces error
-            // handling and helps avoid silencing failures.
-            //
-            // In our example, we are only going to log the error to STDOUT.
-            println!("connection error = {:?}", err);
-        });
-
-    // Start the Tokio runtime.
-    //
-    // The Tokio is a pre-configured "out of the box" runtime for building
-    // asynchronous applications. It includes both a reactor and a task
-    // scheduler. This means applications are multithreaded by default.
-    //
-    // This function blocks until the runtime reaches an idle state. Idle is
-    // defined as all spawned tasks have completed and all I/O resources (TCP
-    // sockets in our case) have been dropped.
-    println!("About to create the stream and write to it...");
-    tokio::run(client);
-    println!("Stream has been created and written to.");
-
-    Ok(())
-}