diff --git a/src/bin/h2s-server.rs b/src/bin/h2s-server.rs index 478aa98a..42c9cc19 100644 --- a/src/bin/h2s-server.rs +++ b/src/bin/h2s-server.rs @@ -1,26 +1,17 @@ +use std::sync::Arc; + use failure::*; use futures::*; - -// Simple H2 server to test H2 speed with h2s-client.rs - use hyper::{Request, Response, Body}; -use tokio::net::TcpListener; +use openssl::ssl::{SslMethod, SslAcceptor, SslFiletype}; +use tokio::net::{TcpListener, TcpStream}; use proxmox_backup::configdir; -use openssl::ssl::{SslMethod, SslAcceptor, SslFiletype}; -use std::sync::Arc; -use tokio_openssl::SslAcceptorExt; - -pub fn main() -> Result<(), Error> { - - start_h2_server()?; - - Ok(()) -} - -pub fn start_h2_server() -> Result<(), Error> { +// Simple H2 server to test H2 speed with h2s-client.rs +#[tokio::main] +async fn main() -> Result<(), Error> { let key_path = configdir!("/proxy.key"); let cert_path = configdir!("/proxy.pem"); @@ -37,70 +28,53 @@ pub fn start_h2_server() -> Result<(), Error> { println!("listening on {:?}", listener.local_addr()); - let server = listener - .incoming() - .map_err(Error::from) - .and_then(move |sock| { - sock.set_nodelay(true).unwrap(); - sock.set_send_buffer_size(1024*1024).unwrap(); - sock.set_recv_buffer_size(1024*1024).unwrap(); - acceptor.accept_async(sock).map_err(|e| e.into()) - }) - .then(|r| match r { - // accept()s can fail here with an Err() when eg. the client rejects - // the cert and closes the connection, so we follow up with mapping - // it to an option and then filtering None with filter_map - Ok(c) => Ok::<_, Error>(Some(c)), - Err(e) => { - if let Some(_io) = e.downcast_ref::() { - // "real" IO errors should not simply be ignored - bail!("shutting down..."); - } else { - // handshake errors just get filtered by filter_map() below: - Ok(None) + let mut incoming = listener.incoming(); + while let Some(socket) = incoming.try_next().await? { + tokio::spawn(handle_connection(socket, Arc::clone(&acceptor)) + .map(|res| { + if let Err(err) = res { + eprintln!("Error: {}", err); } - } - }) - .filter_map(|r| { - // Filter out the Nones - r - }) - .for_each(move |socket| { - - let mut http = hyper::server::conn::Http::new(); - http.http2_only(true); - // increase window size: todo - find optiomal size - let max_window_size = (1 << 31) - 2; - http.http2_initial_stream_window_size(max_window_size); - http.http2_initial_connection_window_size(max_window_size); - - let service = hyper::service::service_fn(|_req: Request| { - println!("Got request"); - let buffer = vec![65u8; 1024*1024]; // nonsense [A,A,A,A...] - let body = Body::from(buffer); - - let response = Response::builder() - .status(http::StatusCode::OK) - .header(http::header::CONTENT_TYPE, "application/octet-stream") - .body(body) - .unwrap(); - Ok::<_, Error>(response) - }); - http.serve_connection(socket, service) - .map_err(Error::from) - }) - .and_then(|_| { - println!("H2 connection CLOSE !"); - Ok(()) - }) - .then(|res| { - if let Err(e) = res { - println!(" -> err={:?}", e); - } - Ok(()) - }); - - tokio::run(server); + })); + } Ok(()) } + +async fn handle_connection( + socket: TcpStream, + acceptor: Arc, +) -> Result<(), Error> { + socket.set_nodelay(true).unwrap(); + socket.set_send_buffer_size(1024*1024).unwrap(); + socket.set_recv_buffer_size(1024*1024).unwrap(); + + let socket = tokio_openssl::accept(acceptor.as_ref(), socket).await?; + + let mut http = hyper::server::conn::Http::new(); + http.http2_only(true); + // increase window size: todo - find optiomal size + let max_window_size = (1 << 31) - 2; + http.http2_initial_stream_window_size(max_window_size); + http.http2_initial_connection_window_size(max_window_size); + + let service = hyper::service::service_fn(|_req: Request| { + println!("Got request"); + let buffer = vec![65u8; 1024*1024]; // nonsense [A,A,A,A...] + let body = Body::from(buffer); + + let response = Response::builder() + .status(http::StatusCode::OK) + .header(http::header::CONTENT_TYPE, "application/octet-stream") + .body(body) + .unwrap(); + future::ok::<_, Error>(response) + }); + + http.serve_connection(socket, service) + .map_err(Error::from) + .await?; + + println!("H2 connection CLOSE !"); + Ok(()) +}