src/bin/h2client.rs

   1 use failure::*;
   2 use futures::*;
   3
   4 // Simple H2 client to test H2 download speed using h2server.rs
   5
   6 use tokio::net::TcpStream;
   7
   8 struct Process {
   9     body: h2::RecvStream,
  10     trailers: bool,
  11     bytes: usize,
  12 }
  13
  14 impl Future for Process {
  15     type Item = usize;
  16     type Error = Error;
  17
  18     fn poll(&mut self) -> Poll<usize, Error> {
  19         loop {
  20             if self.trailers {
  21                 let trailers = try_ready!(self.body.poll_trailers());
  22                 if let Some(trailers) = trailers {
  23                     println!("trailers: {:?}", trailers);
  24                 }
  25                 println!("Received {} bytes", self.bytes);
  26
  27                 return Ok(Async::Ready(self.bytes));
  28             } else {
  29                 match try_ready!(self.body.poll()) {
  30                     Some(chunk) => {
  31                         self.body.release_capacity().release_capacity(chunk.len())?;
  32                         self.bytes += chunk.len();
  33                         // println!("GOT FRAME {}", chunk.len());
  34                     },
  35                     None => {
  36                         self.trailers = true;
  37                     },
  38                 }
  39             }
  40         }
  41     }
  42 }
  43
  44 fn send_request(mut client: h2::client::SendRequest<bytes::Bytes>) -> impl Future<Item=usize, Error=Error> {
  45
  46     println!("sending request");
  47
  48     let request = http::Request::builder()
  49         .uri("http://localhost/")
  50         .body(())
  51         .unwrap();
  52
  53     let (response, _stream) = client.send_request(request, true).unwrap();
  54
  55     response
  56         .map_err(Error::from)
  57         .and_then(|response| {
  58             Process { body: response.into_body(), trailers: false, bytes: 0 }
  59         })
  60 }
  61
  62 pub fn main() -> Result<(), Error> {
  63
  64     let tcp_stream = TcpStream::connect(&"127.0.0.1:8008".parse().unwrap());
  65
  66     let start = std::time::SystemTime::now();
  67
  68     let tcp = tcp_stream
  69         .map_err(Error::from)
  70         .and_then(|c| {
  71             h2::client::handshake(c)
  72                 .map_err(Error::from)
  73         })
  74         .and_then(|(client, h2)| {
  75
  76             // Spawn a task to run the conn...
  77             tokio::spawn(h2.map_err(|e| println!("GOT ERR={:?}", e)));
  78
  79             futures::stream::repeat(())
  80                 .take(10)
  81                 .and_then(move |_| send_request(client.clone()))
  82                 .fold(0, move |mut acc, size| {
  83                     acc += size;
  84                     Ok::<_, Error>(acc)
  85                 })
  86         })
  87         .then(move |result| {
  88             match result {
  89                 Err(err) => {
  90                     println!("ERROR {}", err);
  91                 }
  92                 Ok(bytes) => {
  93                     let elapsed = start.elapsed().unwrap();
  94                     let elapsed = (elapsed.as_secs() as f64) +
  95                         (elapsed.subsec_millis() as f64)/1000.0;
  96
  97                     println!("Downloaded {} bytes, {} MB/s", bytes, (bytes as f64)/(elapsed*1024.0*1024.0));
  98                 }
  99             }
 100             Ok(())
 101         });
 102
 103     tokio::run(tcp);
 104
 105     Ok(())
 106 }