Read body in the example server (#544)

Co-authored-by: Kornel Lesiński <[email protected]>

Author: nox

examples/server.rs

@@ -1,7 +1,9 @@
 use std::error::Error;
 
 use bytes::Bytes;
-use h2::server;
+use h2::server::{self, SendResponse};
+use h2::RecvStream;
+use http::Request;
 use tokio::net::{TcpListener, TcpStream};
 
 #[tokio::main]
@@ -15,30 +17,49 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     loop {
         if let Ok((socket, _peer_addr)) = listener.accept().await {
             tokio::spawn(async move {
-                if let Err(e) = handle(socket).await {
+                if let Err(e) = serve(socket).await {
                     println!("  -> err={:?}", e);
                 }
             });
         }
     }
 }
 
-async fn handle(socket: TcpStream) -> Result<(), Box<dyn Error + Send + Sync>> {
+async fn serve(socket: TcpStream) -> Result<(), Box<dyn Error + Send + Sync>> {
     let mut connection = server::handshake(socket).await?;
     println!("H2 connection bound");
 
     while let Some(result) = connection.accept().await {
-        let (request, mut respond) = result?;
-        println!("GOT request: {:?}", request);
-        let response = http::Response::new(());
+        let (request, respond) = result?;
+        tokio::spawn(async move {
+            if let Err(e) = handle_request(request, respond).await {
+                println!("error while handling request: {}", e);
+            }
+        });
+    }
+
+    println!("~~~~~~~~~~~ H2 connection CLOSE !!!!!! ~~~~~~~~~~~");
+    Ok(())
+}
 
-        let mut send = respond.send_response(response, false)?;
+async fn handle_request(
+    mut request: Request<RecvStream>,
+    mut respond: SendResponse<Bytes>,
+) -> Result<(), Box<dyn Error + Send + Sync>> {
+    println!("GOT request: {:?}", request);
 
-        println!(">>>> sending data");
-        send.send_data(Bytes::from_static(b"hello world"), true)?;
+    let body = request.body_mut();
+    while let Some(data) = body.data().await {
+        let data = data?;
+        println!("<<<< recv {:?}", data);
+        let _ = body.flow_control().release_capacity(data.len());
     }
 
-    println!("~~~~~~~~~~~~~~~~~~~~~~~~~~~ H2 connection CLOSE !!!!!! ~~~~~~~~~~~");
+    let response = http::Response::new(());
+    let mut send = respond.send_response(response, false)?;
+    println!(">>>> send");
+    send.send_data(Bytes::from_static(b"hello "), false)?;
+    send.send_data(Bytes::from_static(b"world\n"), true)?;
 
     Ok(())
 }

src/server.rs

@@ -182,9 +182,11 @@ pub struct Handshake<T, B: Buf = Bytes> {
 /// # async fn doc<T: AsyncRead + AsyncWrite + Unpin>(my_io: T) {
 /// let mut server = server::handshake(my_io).await.unwrap();
 /// while let Some(request) = server.accept().await {
-///     let (request, respond) = request.unwrap();
-///     // Process the request and send the response back to the client
-///     // using `respond`.
+///     tokio::spawn(async move {
+///         let (request, respond) = request.unwrap();
+///         // Process the request and send the response back to the client
+///         // using `respond`.
+///     });
 /// }
 /// # }
 /// #