examples/h2client.rs

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