From fea3ae6ca916712c2e32c7fa31be8289a19582cd Mon Sep 17 00:00:00 2001
From: Anthony Ramine <123095+nox@users.noreply.github.com>
Date: Thu, 6 May 2021 20:42:29 +0200
Subject: [PATCH] Read body in the example server (#544)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Co-authored-by: Kornel Lesiński <kornel@cloudflare.com>
---
 examples/server.rs | 45 +++++++++++++++++++++++++++++++++------------
 src/server.rs      |  8 +++++---
 2 files changed, 38 insertions(+), 15 deletions(-)

diff --git a/examples/server.rs b/examples/server.rs
index 777f4ea..6d6490d 100644
--- a/examples/server.rs
+++ b/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,7 +17,7 @@ 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);
                 }
             });
@@ -23,22 +25,41 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     }
 }
 
-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 mut send = respond.send_response(response, false)?;
-
-        println!(">>>> sending data");
-        send.send_data(Bytes::from_static(b"hello world"), true)?;
+        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 !!!!!! ~~~~~~~~~~~");
+    println!("~~~~~~~~~~~ H2 connection CLOSE !!!!!! ~~~~~~~~~~~");
+    Ok(())
+}
+
+async fn handle_request(
+    mut request: Request<RecvStream>,
+    mut respond: SendResponse<Bytes>,
+) -> Result<(), Box<dyn Error + Send + Sync>> {
+    println!("GOT request: {:?}", request);
+
+    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());
+    }
+
+    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(())
 }
diff --git a/src/server.rs b/src/server.rs
index 6ad010b..f713153 100644
--- a/src/server.rs
+++ b/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`.
+///     });
 /// }
 /// # }
 /// #