Improved the way Body is read

Author: amigin

.gitignore

@@ -1,4 +1,5 @@
 **/target
 **/Cargo.lock
 .vscode
-.idea
+.idea
+.DS_Store

my-http-server-core/Cargo.toml

@@ -34,4 +34,5 @@ hyper-util = { version = "*", features = ["tokio"] }
 flate2 = "*"
 brotli = "*"
 http = "*"
+bytes = "*"
 my-hyper-utils = { tag = "0.1.0", git = "https://github.com/MyJetTools/my-hyper-utils.git" }

my-http-server-core/src/http_request/http_request.rs

@@ -2,7 +2,7 @@ use std::{collections::HashMap, net::SocketAddr, sync::Arc};
 
 use crate::{
     http_headers::*, CookiesReader, HttpFailResult, HttpPath, HttpPathReader, HttpRequestBody,
-    HttpRequestHeaders, RequestData, RequestIp, UrlEncodedData,
+    HttpRequestHeaders, MyHyperHttpRequest, RequestData, RequestIp, UrlEncodedData,
 };
 
 use hyper::{Method, Uri};
@@ -46,7 +46,7 @@ impl HttpRequest {
         let http_path = HttpPath::from_str(req.uri().path());
 
         Self {
-            data: RequestData::Incoming(Some(req)),
+            data: RequestData::new(req),
             addr,
             key_values: None,
             content_type_header: None,
@@ -79,16 +79,15 @@ impl HttpRequest {
     }
 
     pub async fn get_body(&mut self) -> Result<&HttpRequestBody, HttpFailResult> {
-        let result = self.data.convert_to_body_if_requires().await?;
-        Ok(result.unwrap())
+        self.data.get_body().await
     }
 
     pub async fn receive_body(&mut self) -> Result<HttpRequestBody, HttpFailResult> {
         self.data.receive_body().await
     }
 
-    pub fn take_incoming_body(&mut self) -> hyper::Request<hyper::body::Incoming> {
-        self.data.take_incoming_body()
+    pub fn take_my_hyper_http_request(&mut self) -> MyHyperHttpRequest {
+        self.data.take_my_hyper_http_request()
     }
 
     pub fn get_path<'s>(&'s self) -> HttpPathReader<'s> {

my-http-server-core/src/http_request/http_request_body_mode.rs

@@ -0,0 +1,51 @@
+use crate::{ContentEncoding, HttpFailResult, HttpRequestBody};
+
+pub enum HttpRequestBodyMode {
+    Incoming(Option<hyper::body::Incoming>),
+    Full(HttpRequestBody),
+}
+
+impl HttpRequestBodyMode {
+    pub async fn get_http_request_body(&mut self) -> Result<&HttpRequestBody, HttpFailResult> {
+        match self {
+            HttpRequestBodyMode::Incoming(incoming) => {
+                let take = incoming.take().unwrap();
+                let bytes = read_bytes(ContentEncoding::None, take).await?;
+                let body = HttpRequestBody::new(bytes, None)?;
+                *self = HttpRequestBodyMode::Full(body);
+            }
+            HttpRequestBodyMode::Full(http_request_body) => return Ok(http_request_body),
+        }
+
+        match self {
+            HttpRequestBodyMode::Incoming(_) => {
+                panic!("We should never be here")
+            }
+            HttpRequestBodyMode::Full(http_request_body) => Ok(http_request_body),
+        }
+    }
+
+    pub async fn into_http_request_body(self) -> Result<HttpRequestBody, HttpFailResult> {
+        match self {
+            HttpRequestBodyMode::Incoming(mut incoming) => {
+                let take = incoming.take().unwrap();
+                let bytes = read_bytes(ContentEncoding::None, take).await?;
+                let body = HttpRequestBody::new(bytes, None)?;
+                return Ok(body);
+            }
+            HttpRequestBodyMode::Full(http_request_body) => return Ok(http_request_body),
+        }
+    }
+}
+
+async fn read_bytes(
+    body_compression: ContentEncoding,
+    incoming: impl hyper::body::Body<Data = hyper::body::Bytes, Error = hyper::Error>,
+) -> Result<Vec<u8>, HttpFailResult> {
+    use http_body_util::BodyExt;
+
+    let collected = incoming.collect().await?;
+    let bytes = collected.to_bytes();
+
+    body_compression.decompress_if_needed(bytes)
+}

my-http-server-core/src/http_request/mod.rs

@@ -14,3 +14,7 @@ mod http_path_reader;
 pub use http_path_reader::*;
 mod content_encoding;
 pub use content_encoding::*;
+mod http_request_body_mode;
+pub use http_request_body_mode::*;
+mod my_hyper_http_request;
+pub use my_hyper_http_request::*;

my-http-server-core/src/http_request/my_hyper_http_request.rs

@@ -0,0 +1,35 @@
+use http::Uri;
+
+pub enum MyHyperHttpRequest {
+    Incoming(hyper::Request<hyper::body::Incoming>),
+    Full(hyper::Request<http_body_util::Full<bytes::Bytes>>),
+}
+
+impl MyHyperHttpRequest {
+    pub fn uri(&self) -> &Uri {
+        match self {
+            MyHyperHttpRequest::Incoming(req) => &req.uri(),
+            MyHyperHttpRequest::Full(req) => &req.uri(),
+        }
+    }
+    pub fn headers(&self) -> &http::HeaderMap<http::header::HeaderValue> {
+        match self {
+            MyHyperHttpRequest::Incoming(req) => req.headers(),
+            MyHyperHttpRequest::Full(req) => req.headers(),
+        }
+    }
+
+    pub fn version(&self) -> http::Version {
+        match self {
+            MyHyperHttpRequest::Incoming(req) => req.version(),
+            MyHyperHttpRequest::Full(req) => req.version(),
+        }
+    }
+
+    pub fn extensions(&self) -> &http::Extensions {
+        match self {
+            MyHyperHttpRequest::Incoming(req) => req.extensions(),
+            MyHyperHttpRequest::Full(req) => req.extensions(),
+        }
+    }
+}

my-http-server-core/src/http_request/request_data.rs

@@ -1,149 +1,71 @@
-use hyper::{HeaderMap, Uri};
+use http::{HeaderMap, HeaderValue};
 
-use crate::{HttpFailResult, HttpRequestBody, HttpRequestHeaders};
+use hyper::Uri;
 
-use super::ContentEncoding;
+use crate::{HttpFailResult, HttpRequestBody, HttpRequestBodyMode, MyHyperHttpRequest};
 
-pub enum RequestData {
-    Incoming(Option<hyper::Request<hyper::body::Incoming>>),
-    AsBody {
-        uri: Uri,
-        headers: HeaderMap,
-        body: Option<HttpRequestBody>,
-    },
-    Taken,
+pub struct RequestData {
+    parts: hyper::http::request::Parts,
+    body: Option<HttpRequestBodyMode>,
 }
 
 impl RequestData {
-    pub async fn convert_to_body_if_requires(
-        &mut self,
-    ) -> Result<Option<&HttpRequestBody>, HttpFailResult> {
-        let (uri, headers, bytes) = match self {
-            Self::Incoming(incoming) => {
-                let incoming = incoming.take().unwrap();
-                let (parts, incoming) = incoming.into_parts();
-
-                let headers = parts.headers;
-
-                let content_encoding = headers.get_content_encoding()?;
-
-                let body = read_bytes(content_encoding, incoming).await?;
-
-                (parts.uri, headers, body)
-            }
-            Self::AsBody { body, .. } => match body.as_ref() {
-                Some(itm) => {
-                    return Ok(Some(itm));
-                }
-                None => {
-                    panic!("You are trying to access body content after receiving ownership of it")
-                }
-            },
-            Self::Taken => {
-                panic!("Body is taken by some middleware before")
-            }
-        };
-
-        let content_type = headers.try_get_case_insensitive("content-type");
-
-        let content_type = match content_type {
-            Some(content_type) => Some(content_type.as_str()?.to_string()),
-            None => None,
-        };
-
-        let body = HttpRequestBody::new(bytes, content_type)?;
-        *self = Self::AsBody {
-            body: Some(body),
-            uri,
-            headers,
-        };
-        Ok(self.try_unwrap_as_body())
+    pub fn new(req: hyper::Request<hyper::body::Incoming>) -> Self {
+        let parts = req.into_parts();
+        Self {
+            parts: parts.0,
+            body: Some(HttpRequestBodyMode::Incoming(Some(parts.1))),
+        }
     }
 
-    pub async fn receive_body(&mut self) -> Result<HttpRequestBody, HttpFailResult> {
-        match self {
-            Self::Incoming(incoming) => {
-                let incoming = incoming.take().unwrap();
-
-                let content_encoding = incoming.headers().get_content_encoding()?;
-
-                let bytes = read_bytes(content_encoding, incoming).await?;
-
-                let body = HttpRequestBody::new(bytes, None)?;
-                return Ok(body);
-            }
-            Self::AsBody { body, .. } => match body.take() {
-                Some(itm) => {
-                    return Ok(itm);
-                }
-                None => {
-                    panic!("You are trying to receive body for a second time")
-                }
-            },
-            Self::Taken => {
-                panic!("Body is taken by some middleware before")
+    pub async fn get_body(&mut self) -> Result<&HttpRequestBody, HttpFailResult> {
+        match self.body.as_mut() {
+            Some(body) => body.get_http_request_body().await,
+            None => {
+                panic!("Body is removed and can not be accessed")
             }
-        };
+        }
     }
 
-    fn try_unwrap_as_body(&self) -> Option<&HttpRequestBody> {
-        match self {
-            Self::Incoming(_) => None,
-            Self::AsBody { body, .. } => {
-                let body = body.as_ref()?;
-                return Some(body);
-            }
-            Self::Taken => {
+    pub async fn receive_body(&mut self) -> Result<HttpRequestBody, HttpFailResult> {
+        match self.body.take() {
+            Some(body) => return body.into_http_request_body().await,
+            None => {
                 panic!("Body is taken by some middleware before")
             }
         }
     }
 
     pub fn uri(&self) -> &Uri {
-        match self {
-            Self::Incoming(incoming) => incoming.as_ref().unwrap().uri(),
-            Self::AsBody { uri, .. } => uri,
-            Self::Taken => {
-                panic!("Body is taken by some middleware before")
-            }
-        }
+        &self.parts.uri
     }
 
-    pub fn headers(&self) -> &impl HttpRequestHeaders {
-        match self {
-            Self::Incoming(incoming) => incoming.as_ref().unwrap().headers(),
-            Self::AsBody { headers, .. } => headers,
-            Self::Taken => {
-                panic!("Body is taken by some middleware before")
-            }
-        }
+    pub fn headers(&self) -> &HeaderMap<HeaderValue> {
+        &self.parts.headers
     }
 
-    pub fn take_incoming_body(&mut self) -> hyper::Request<hyper::body::Incoming> {
-        match self {
-            Self::Incoming(incoming) => {
-                let result = incoming.take().unwrap();
-                *self = Self::Taken;
-                result
-            }
-            Self::AsBody { .. } => {
-                panic!("Body is taken by some middleware before")
-            }
-            Self::Taken => {
+    pub fn take_my_hyper_http_request(&mut self) -> MyHyperHttpRequest {
+        match self.body.take() {
+            Some(body) => match body {
+                HttpRequestBodyMode::Incoming(mut incoming) => {
+                    let result =
+                        hyper::Request::from_parts(self.parts.clone(), incoming.take().unwrap());
+
+                    return MyHyperHttpRequest::Incoming(result);
+                }
+                HttpRequestBodyMode::Full(body) => {
+                    let body = body.as_slice().to_vec();
+
+                    let body = http_body_util::Full::new(bytes::Bytes::from(body));
+
+                    let req = hyper::Request::from_parts(self.parts.clone(), body);
+
+                    MyHyperHttpRequest::Full(req)
+                }
+            },
+            None => {
                 panic!("Body is taken by some middleware before")
             }
         }
     }
 }
-
-async fn read_bytes(
-    body_compression: ContentEncoding,
-    incoming: impl hyper::body::Body<Data = hyper::body::Bytes, Error = hyper::Error>,
-) -> Result<Vec<u8>, HttpFailResult> {
-    use http_body_util::BodyExt;
-
-    let collected = incoming.collect().await?;
-    let bytes = collected.to_bytes();
-
-    body_compression.decompress_if_needed(bytes)
-}

my-http-server-web-sockets/src/helpers.rs

@@ -23,7 +23,7 @@ pub async fn handle_web_socket_upgrade<
 
     let addr = req.addr.clone();
 
-    let req = req.take_incoming_body();
+    let req = req.take_my_hyper_http_request();
 
     let upgrade_result = crate::web_sockets_upgrade::upgrade(
         id,

my-http-server-web-sockets/src/my_web_socket_http_request.rs

@@ -1,5 +1,6 @@
 use hyper::{header::*, *};
 use hyper_tungstenite::tungstenite::http::Extensions;
+use my_http_server_core::MyHyperHttpRequest;
 
 pub struct MyWebSocketHttpRequest {
     uri: Uri,
@@ -9,7 +10,7 @@ pub struct MyWebSocketHttpRequest {
 }
 
 impl<'s> MyWebSocketHttpRequest {
-    pub fn new(req: &Request<hyper::body::Incoming>) -> Self {
+    pub fn new(req: &MyHyperHttpRequest) -> Self {
         Self {
             uri: req.uri().clone(),
             headers: req.headers().clone(),