src/bin/h2s-client.rs: avoid hyper, use h2 directly

But performance is still bad.

src/bin/h2s-client.rs

@@ -3,54 +3,101 @@ use futures::*;
 
 // Simple H2 client to test H2 download speed using h2s-server.rs
 
-fn build_client() -> hyper::client::Client<hyper_tls::HttpsConnector<hyper::client::HttpConnector>> {
+use tokio::net::TcpStream;
-    let mut builder = native_tls::TlsConnector::builder();
+//use native_tls::TlsConnector;
-    builder.danger_accept_invalid_certs(true);
+use std::io::{Read, Write};
-    let tlsconnector = builder.build().unwrap();
+
-    let mut httpc = hyper::client::HttpConnector::new(1);
+struct Process {
-    httpc.set_nodelay(true); // important for h2 download performance!
+    body: h2::RecvStream,
-    httpc.enforce_http(false); // we want https...
+    trailers: bool,
-    let mut https = hyper_tls::HttpsConnector::from((httpc, tlsconnector));
+    bytes: usize,
-    https.https_only(true); // force it!
+}
-    hyper::client::Client::builder()
+
-        .http2_only(true)
+impl Future for Process {
-        .http2_initial_stream_window_size( (1 << 31) - 2)
+    type Item = usize;
-        .http2_initial_connection_window_size( (1 << 31) - 2)
+    type Error = Error;
-         // howto?? .http2_max_frame_size(4*1024*1024) ??
+
-        .build::<_, hyper::Body>(https)
+    fn poll(&mut self) -> Poll<usize, Error> {
+        loop {
+            if self.trailers {
+                let trailers = try_ready!(self.body.poll_trailers());
+                if let Some(trailers) = trailers {
+                    println!("trailers: {:?}", trailers);
+                }
+                println!("Received {} bytes", self.bytes);
+
+                return Ok(Async::Ready(self.bytes));
+            } else {
+                match try_ready!(self.body.poll()) {
+                    Some(chunk) => {
+                        self.body.release_capacity().release_capacity(chunk.len())?;
+                        self.bytes += chunk.len();
+                        // println!("GOT FRAME {}", chunk.len());
+                    },
+                    None => {
+                        self.trailers = true;
+                    },
+                }
+            }
+        }
+    }
+}
+
+fn send_request(mut client: h2::client::SendRequest<bytes::Bytes>) -> impl Future<Item=usize, Error=Error> {
+
+    println!("sending request");
+
+    let request = http::Request::builder()
+        .uri("http://localhost/")
+        .body(())
+        .unwrap();
+
+    let (response, _stream) = client.send_request(request, true).unwrap();
+
+    response
+        .map_err(Error::from)
+        .and_then(|response| {
+            Process { body: response.into_body(), trailers: false, bytes: 0 }
+        })
 }
 
 pub fn main() -> Result<(), Error> {
 
-    let client = build_client();
+    let tcp_stream = tokio::net::TcpStream::connect(&"127.0.0.1:8008".parse().unwrap());
 
     let start = std::time::SystemTime::now();
 
-    let task = futures::stream::repeat(())
+    let tcp = tcp_stream
-        .take(100)
+        .map_err(Error::from)
-        .and_then(move |_| {
+        .and_then(|c| {
-            let request = http::Request::builder()
+            let mut builder = native_tls::TlsConnector::builder();
-                .method("GET")
+            builder.danger_accept_invalid_certs(true);
-                .uri("https://localhost:8008/")
+            let connector = builder.build().unwrap();
-                .body(hyper::Body::empty())
+            let connector = tokio_tls::TlsConnector::from(connector);
-                .unwrap();
+            connector.connect("localhost", c)
-
-            client
-                .request(request)
                 .map_err(Error::from)
-                .and_then(|resp| {
-                    resp.into_body()
-                        .map_err(Error::from)
-                        .fold(0, move |mut acc, chunk| {
-                            println!("got frame {}", chunk.len());
-                            acc += chunk.len();
-                            Ok::<_, Error>(acc)
-                        })
-                })
         })
-        .fold(0, move |mut acc, size| {
+        .map_err(Error::from)
-            acc += size;
+        .and_then(|c| {
-            Ok::<_, Error>(acc)
+            h2::client::Builder::new()
+                .initial_connection_window_size(1024*1024*1024)
+                .initial_window_size(1024*1024*1024)
+                .max_frame_size(4*1024*1024)
+                .handshake(c)
+                .map_err(Error::from)
+        })
+        .and_then(|(client, h2)| {
+
+            // Spawn a task to run the conn...
+            tokio::spawn(h2.map_err(|e| println!("GOT ERR={:?}", e)));
+
+            futures::stream::repeat(())
+                .take(100)
+                .and_then(move |_| send_request(client.clone()))
+                .fold(0, move |mut acc, size| {
+                    acc += size;
+                    Ok::<_, Error>(acc)
+                })
         })
         .then(move |result| {
             match result {
@@ -68,7 +115,7 @@ pub fn main() -> Result<(), Error> {
             Ok(())
         });
 
-    tokio::run(task);
+    tokio::run(tcp);
 
     Ok(())
 }