Update examples to std-future

examples/akamai.rs

@@ -1,3 +1,10 @@
+fn main() {
+    // Enable the below code once tokio_rustls moves to std::future
+}
+
+/*
+#![feature(async_await)]
+
 use h2::client;
 
 use futures::*;
@@ -10,10 +17,12 @@ use tokio_rustls::ClientConfigExt;
 use webpki::DNSNameRef;
 
 use std::net::ToSocketAddrs;
+use std::error::Error;
 
 const ALPN_H2: &str = "h2";
 
-pub fn main() {
+#[tokio::main]
+pub async fn main() -> Result<(), Box<dyn Error>> {
     let _ = env_logger::try_init();
 
     let tls_client_config = std::sync::Arc::new({
@@ -33,49 +42,30 @@ pub fn main() {
 
     println!("ADDR: {:?}", addr);
 
-    let tcp = TcpStream::connect(&addr);
+    let tcp = TcpStream::connect(&addr).await?;
     let dns_name = DNSNameRef::try_from_ascii_str("http2.akamai.com").unwrap();
+    let res = tls_client_config.connect_async(dns_name, tcp).await;
+    let tls = res.unwrap();
+    {
+        let (_, session) = tls.get_ref();
+        let negotiated_protocol = session.get_alpn_protocol();
+        assert_eq!(Some(ALPN_H2), negotiated_protocol.as_ref().map(|x| &**x));
+    }
 
-    let tcp = tcp.then(move |res| {
+    println!("Starting client handshake");
-        let tcp = res.unwrap();
+    let (mut client, h2) = client::handshake(tls).await?;
-        tls_client_config
-            .connect_async(dns_name, tcp)
-            .then(|res| {
-                let tls = res.unwrap();
-                {
-                    let (_, session) = tls.get_ref();
-                    let negotiated_protocol = session.get_alpn_protocol();
-                    assert_eq!(Some(ALPN_H2), negotiated_protocol.as_ref().map(|x| &**x));
-                }
 
-                println!("Starting client handshake");
+    let request = Request::builder()
-                client::handshake(tls)
-            })
-            .then(|res| {
-                let (mut client, h2) = res.unwrap();
-
-                let request = Request::builder()
                     .method(Method::GET)
                     .uri("https://http2.akamai.com/")
                     .body(())
                     .unwrap();
 
-                let (response, _) = client.send_request(request, true).unwrap();
+    let (response, _) = client.send_request(request, true).unwrap();
-
+    let (_, mut body) = response.await?.into_parts();
-                let stream = response.and_then(|response| {
+    while let Some(chunk) = body.next().await {
-                    let (_, body) = response.into_parts();
+        println!("RX: {:?}", chunk?);
-
+    }
-                    body.for_each(|chunk| {
+    Ok(())
-                        println!("RX: {:?}", chunk);
-                        Ok(())
-                    })
-                });
-
-                h2.join(stream)
-            })
-    })
-        .map_err(|e| eprintln!("ERROR: {:?}", e))
-        .map(|((), ())| ());
-
-    tokio::run(tcp);
 }
+*/

examples/client.rs

@@ -1,8 +1,14 @@
+#![feature(async_await)]
+
+use futures::{ready, Stream};
 use h2::client;
 use h2::RecvStream;
+use http::{HeaderMap, Request};
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
 
-use futures::*;
+use std::error::Error;
-use http::*;
 
 use tokio::net::TcpStream;
 
@@ -12,76 +18,70 @@ struct Process {
 }
 
 impl Future for Process {
-    type Item = ();
+    type Output = Result<(), h2::Error>;
-    type Error = h2::Error;
 
-    fn poll(&mut self) -> Poll<(), h2::Error> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         loop {
             if self.trailers {
-                let trailers = try_ready!(self.body.poll_trailers());
+                let trailers = ready!(self.body.poll_trailers(cx));
 
                 println!("GOT TRAILERS: {:?}", trailers);
 
-                return Ok(().into());
+                return Poll::Ready(Ok(()));
             } else {
-                match try_ready!(self.body.poll()) {
+                match ready!(Pin::new(&mut self.body).poll_next(cx)) {
-                    Some(chunk) => {
+                    Some(Ok(chunk)) => {
                         println!("GOT CHUNK = {:?}", chunk);
-                    },
+                    }
+                    Some(Err(e)) => return Poll::Ready(Err(e)),
                     None => {
                         self.trailers = true;
-                    },
+                    }
                 }
             }
         }
     }
 }
 
-pub fn main() {
+#[tokio::main]
+pub async fn main() -> Result<(), Box<dyn Error>> {
     let _ = env_logger::try_init();
 
-    let tcp = TcpStream::connect(&"127.0.0.1:5928".parse().unwrap());
+    let tcp = TcpStream::connect(&"127.0.0.1:5928".parse().unwrap()).await?;
+    let (mut client, h2) = client::handshake(tcp).await?;
 
-    let tcp = tcp.then(|res| {
+    println!("sending request");
-        let tcp = res.unwrap();
-        client::handshake(tcp)
-    }).then(|res| {
-            let (mut client, h2) = res.unwrap();
 
-            println!("sending request");
+    let request = Request::builder()
+        .uri("https://http2.akamai.com/")
+        .body(())
+        .unwrap();
 
-            let request = Request::builder()
+    let mut trailers = HeaderMap::new();
-                .uri("https://http2.akamai.com/")
+    trailers.insert("zomg", "hello".parse().unwrap());
-                .body(())
-                .unwrap();
 
-            let mut trailers = HeaderMap::new();
+    let (response, mut stream) = client.send_request(request, false).unwrap();
-            trailers.insert("zomg", "hello".parse().unwrap());
 
-            let (response, mut stream) = client.send_request(request, false).unwrap();
+    // send trailers
+    stream.send_trailers(trailers).unwrap();
 
-            // send trailers
+    // Spawn a task to run the conn...
-            stream.send_trailers(trailers).unwrap();
+    tokio::spawn(async move {
+        if let Err(e) = h2.await {
+            println!("GOT ERR={:?}", e);
+        }
+    });
 
-            // Spawn a task to run the conn...
+    let response = response.await?;
-            tokio::spawn(h2.map_err(|e| println!("GOT ERR={:?}", e)));
+    println!("GOT RESPONSE: {:?}", response);
 
-            response
+    // Get the body
-                .and_then(|response| {
+    let (_, body) = response.into_parts();
-                    println!("GOT RESPONSE: {:?}", response);
 
-                    // Get the body
+    Process {
-                    let (_, body) = response.into_parts();
+        body,
-
+        trailers: false,
-                    Process {
+    }
-                        body,
+    .await?;
-                        trailers: false,
+    Ok(())
-                    }
-                })
-                .map_err(|e| {
-                    println!("GOT ERR={:?}", e);
-                })
-        });
-
-    tokio::run(tcp);
 }

examples/server.rs

@@ -1,62 +1,50 @@
+#![feature(async_await)]
+
 use h2::server;
 
 use bytes::*;
 use futures::*;
-use http::*;
+use http::{Response, StatusCode};
 
-use tokio::net::TcpListener;
+use tokio::net::{TcpListener, TcpStream};
+use std::error::Error;
 
-pub fn main() {
+#[tokio::main]
+pub async fn main() -> Result<(), Box<dyn Error>> {
     let _ = env_logger::try_init();
 
     let listener = TcpListener::bind(&"127.0.0.1:5928".parse().unwrap()).unwrap();
 
     println!("listening on {:?}", listener.local_addr());
+    let mut incoming = listener.incoming();
 
-    let server = listener.incoming().for_each(move |socket| {
+    while let Some(socket) = incoming.next().await {
-        // let socket = io_dump::Dump::to_stdout(socket);
+        tokio::spawn(async move {
+            if let Err(e) = handle(socket).await {
+                println!("  -> err={:?}", e); 
+            }
+        });
+    }
 
-        let connection = server::handshake(socket)
+    Ok(())
-            .and_then(|conn| {
-                println!("H2 connection bound");
-
-                conn.for_each(|(request, mut respond)| {
-                    println!("GOT request: {:?}", request);
-
-                    let response = Response::builder().status(StatusCode::OK).body(()).unwrap();
-
-                    let mut send = match respond.send_response(response, false) {
-                        Ok(send) => send,
-                        Err(e) => {
-                            println!(" error respond; err={:?}", e);
-                            return Ok(());
-                        }
-                    };
-
-                    println!(">>>> sending data");
-                    if let Err(e) = send.send_data(Bytes::from_static(b"hello world"), true) {
-                        println!("  -> err={:?}", e);
-                    }
-
-                    Ok(())
-                })
-            })
-            .and_then(|_| {
-                println!("~~~~~~~~~~~~~~~~~~~~~~~~~~~ H2 connection CLOSE !!!!!! ~~~~~~~~~~~");
-                Ok(())
-            })
-            .then(|res| {
-                if let Err(e) = res {
-                    println!("  -> err={:?}", e);
-                }
-
-                Ok(())
-            });
-
-        tokio::spawn(Box::new(connection));
-        Ok(())
-    })
-    .map_err(|e| eprintln!("accept error: {}", e));
-
-    tokio::run(server);
 }
+
+async fn handle(socket: io::Result<TcpStream>) -> Result<(), Box<dyn Error>> {
+    let mut connection = server::handshake(socket?).await?;
+    println!("H2 connection bound");
+
+    while let Some(result) = connection.next().await {
+        let (request, mut respond) = result?;
+        println!("GOT request: {:?}", request);
+        let response = Response::builder().status(StatusCode::OK).body(()).unwrap();
+
+        let mut send = respond.send_response(response, false)?;
+
+        println!(">>>> sending data");
+        send.send_data(Bytes::from_static(b"hello world"), true)?;
+    }
+
+    println!("~~~~~~~~~~~~~~~~~~~~~~~~~~~ H2 connection CLOSE !!!!!! ~~~~~~~~~~~");
+    
+    Ok(())
+}