Merge 556bc59 into 6688e5e

gabotechs · Dec 11, 2023 · 0583417 · 0583417
2 parents 6688e5e + 556bc59
commit 0583417
Show file tree

Hide file tree

Showing 13 changed files with 899 additions and 536 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/server/Cargo.toml b/server/Cargo.toml
@@ -9,9 +9,11 @@ edition = "2021"
 
 [dependencies]
 openssl = { version = "0.10", features = ["vendored"] } # NOTE: neeeded for cross compilations
-hyper = { version = "^0.14.26", features = ["full"] }
-hyper-tls = "^0.5.0"
-tokio = { version = "^1.28.1", features = ["full"] }
+hyper = { version = "^1.0.1", features = ["full"] }
+http-body-util = "0.1.0"
+hyper-util = { version = "0.1.1", features = ["full"]}
+hyper-tls = "0.6.0"
+tokio = { version = "^1.34.0", features = ["full"] }
 url = "^2.3.1"
 anyhow = "^1.0.71"
 serde = { version = "^1.0.163", features = ["derive"] }
@@ -23,7 +25,9 @@ time = { version = "^0.3.6", features = ["formatting", "macros", "parsing"] }
 percent-encoding = "^2.2.0"
 tracing = "^0.1.37"
 async-trait = "^0.1.68"
+futures-util = "0.3.29"
+bytes = "1.5.0"
 
 [dev-dependencies]
 lazy_static = "^1.4.0"
-reqwest = { version = "^0.11.18", features = ["json"] }
+reqwest = { version = "^0.11.22", features = ["json"] }
diff --git a/server/src/_test_tools.rs b/server/src/_test_tools.rs
@@ -1,24 +1,23 @@
-use hyper::{Body, Request};
-
 #[cfg(test)]
 pub(crate) mod tests {
+    use hyper::Request;
     use std::collections::HashMap;
-    use std::convert::Infallible;
+    use std::error::Error;
     use std::str::FromStr;
 
     use anyhow::{anyhow, Context};
     use async_trait::async_trait;
+    use http_body_util::Full;
     use hyper::header::HeaderName;
     use hyper::{HeaderMap, Response, StatusCode, Uri};
     use serde::de::DeserializeOwned;
     use time::{OffsetDateTime, PrimitiveDateTime};
     use url::Url;
 
-    use crate::signing::{ElementsToSign, UrlSigner};
+    use crate::signing::{ElementsToSign, SignedBody, UrlSigner};
+    use crate::sw_body::{empty, sw_body_from_string, SwBody};
     use crate::{GetSecretResponse, SecretGetter, SecretGetterResult};
 
-    use super::*;
-
     #[derive(Clone, Debug)]
     pub(crate) struct ReqBuilder {
         method: String,
@@ -75,10 +74,10 @@ pub(crate) mod tests {
             }
         }
 
-        pub(crate) fn build(&self) -> anyhow::Result<Request<Body>> {
-            let body = match &self.body {
-                Some(b) => Body::from(b.to_string()),
-                None => Body::empty(),
+        pub(crate) fn build(self) -> anyhow::Result<Request<SwBody>> {
+            let body = match self.body.clone() {
+                Some(b) => sw_body_from_string(b),
+                None => empty(),
             };
 
             let mut builder = Request::builder();
@@ -101,7 +100,10 @@ pub(crate) mod tests {
                 datetime: PrimitiveDateTime::new(now.date(), now.time()),
                 method: self.method.clone(),
                 headers: self.build_headers()?,
-                body: self.body.clone(),
+                body: match self.body.clone() {
+                    None => SignedBody::None,
+                    Some(value) => SignedBody::Some(value),
+                },
             };
 
             let signer = UrlSigner::new(id, secret);
@@ -156,18 +158,16 @@ pub(crate) mod tests {
 
     #[async_trait]
     impl SecretGetter for InMemorySecretGetter {
-        type Error = Infallible;
-
-        async fn get_secret(&self, id: &str) -> Result<GetSecretResponse, Self::Error> {
+        async fn get_secret(&self, id: &str) -> Result<GetSecretResponse, Box<dyn Error>> {
             let secret = match self.0.get(id).cloned() {
                 Some(a) => a,
                 None => {
                     return Ok(GetSecretResponse::EarlyResponse(
                         Response::builder()
                             .status(StatusCode::UNAUTHORIZED)
-                            .body(Body::empty())
+                            .body(Full::default())
                             .unwrap(),
-                    ))
+                    ));
                 }
             };
 

diff --git a/server/src/body.rs b/server/src/body.rs
diff --git a/server/src/gateway_callbacks.rs b/server/src/gateway_callbacks.rs
@@ -1,21 +1,25 @@
-use async_trait::async_trait;
-use hyper::{Body, Request, Response};
 use std::fmt::{Display, Formatter};
+
+use async_trait::async_trait;
+use bytes::Bytes;
+use http_body_util::Full;
+use hyper::http::{request, response};
+use hyper::Response;
 use url::Url;
 
 pub enum CallbackResult {
-    EarlyResponse(Response<Body>),
+    EarlyResponse(Response<Full<Bytes>>),
     Empty,
 }
 
 #[async_trait]
 pub trait OnRequest: Sync + Send {
-    async fn call(&self, id: &str, req: &Request<Body>) -> CallbackResult;
+    async fn call(&self, id: &str, req: &request::Parts) -> CallbackResult;
 }
 
 #[async_trait]
 pub trait OnSuccess: Sync + Send {
-    async fn call(&self, id: &str, res: &Response<Body>) -> CallbackResult;
+    async fn call(&self, id: &str, res: &response::Parts) -> CallbackResult;
 }
 
 #[derive(Debug, Clone)]
@@ -44,27 +48,49 @@ pub struct BytesTransferredInfo {
     pub kind: BytesTransferredKind,
 }
 
+impl BytesTransferredInfo {
+    pub fn in_kind(id: &str, url: &Url) -> Self {
+        BytesTransferredInfo {
+            id: id.to_string(),
+            proxy_url: url.clone(),
+            kind: BytesTransferredKind::In,
+        }
+    }
+
+    pub fn out_kind(id: &str, url: &Url) -> Self {
+        BytesTransferredInfo {
+            id: id.to_string(),
+            proxy_url: url.clone(),
+            kind: BytesTransferredKind::Out,
+        }
+    }
+}
+
 #[async_trait]
 pub trait OnBytesTransferred: Sync + Send {
     async fn call(&self, bytes: usize, info: BytesTransferredInfo);
 }
 
 #[cfg(test)]
 mod tests {
+    use std::collections::HashMap;
+    use std::str::FromStr;
+    use std::sync::atomic::AtomicU64;
+    use std::sync::atomic::Ordering::SeqCst;
+    use std::time::Duration;
+
+    use async_trait::async_trait;
+    use hyper::http::{request, response};
+    use hyper::{Request, StatusCode};
+
     use crate::_test_tools::tests::{InMemorySecretGetter, ReqBuilder};
-    use crate::body::body_to_string;
     use crate::gateway_callbacks::{CallbackResult, OnRequest, OnSuccess};
+    use crate::sw_body::{sw_body_to_string, SwBody};
     use crate::{
         BytesTransferredInfo, HeaderMap, OnBytesTransferred, SecretGetterResult, SignwayServer,
     };
-    use async_trait::async_trait;
-    use hyper::body::HttpBody;
-    use hyper::{Body, Request, Response, StatusCode};
-    use std::collections::HashMap;
-    use std::sync::atomic::AtomicU64;
-    use std::sync::atomic::Ordering::SeqCst;
 
-    fn server() -> SignwayServer<InMemorySecretGetter> {
+    fn server() -> SignwayServer {
         SignwayServer::from_env(InMemorySecretGetter(HashMap::from([(
             "foo".to_string(),
             SecretGetterResult {
@@ -74,7 +100,7 @@ mod tests {
         )])))
     }
 
-    fn req() -> Request<Body> {
+    fn req() -> Request<SwBody> {
         ReqBuilder::default()
             .query("page", "1")
             .header("Content-Length", "3")
@@ -90,16 +116,18 @@ mod tests {
 
     #[async_trait]
     impl<'a> OnRequest for SizeCollector<'a> {
-        async fn call(&self, _id: &str, req: &Request<Body>) -> CallbackResult {
-            self.0.fetch_add(req.size_hint().exact().unwrap(), SeqCst);
+        async fn call(&self, _id: &str, req: &request::Parts) -> CallbackResult {
+            let size: &str = req.headers.get("content-length").unwrap().to_str().unwrap();
+            self.0.fetch_add(u64::from_str(size).unwrap(), SeqCst);
             CallbackResult::Empty
         }
     }
 
     #[async_trait]
     impl<'a> OnSuccess for SizeCollector<'a> {
-        async fn call(&self, _id: &str, res: &Response<Body>) -> CallbackResult {
-            self.0.fetch_add(res.size_hint().exact().unwrap(), SeqCst);
+        async fn call(&self, _id: &str, res: &response::Parts) -> CallbackResult {
+            let size: &str = res.headers.get("content-length").unwrap().to_str().unwrap();
+            self.0.fetch_add(u64::from_str(size).unwrap(), SeqCst);
             CallbackResult::Empty
         }
     }
@@ -118,7 +146,7 @@ mod tests {
 
         let response = server()
             .on_request(size_collector)
-            .route_gateway(req())
+            .handler(req())
             .await
             .unwrap();
 
@@ -133,7 +161,7 @@ mod tests {
 
         let response = server()
             .on_success(size_collector)
-            .route_gateway(req())
+            .handler(req())
             .await
             .unwrap();
 
@@ -148,13 +176,15 @@ mod tests {
 
         let response = server()
             .on_bytes_transferred(size_collector)
-            .route_gateway(req())
+            .handler(req())
             .await
             .unwrap();
 
+        tokio::time::sleep(Duration::from_millis(1)).await;
         assert_eq!(response.status(), StatusCode::OK);
         assert_eq!(COUNTER.load(SeqCst), 3);
-        body_to_string(response.into_body(), 396).await.unwrap();
+        sw_body_to_string(response.into_body(), 396).await.unwrap();
+        tokio::time::sleep(Duration::from_millis(1)).await;
         assert_eq!(COUNTER.load(SeqCst), 399);
     }
 }
diff --git a/server/src/lib.rs b/server/src/lib.rs
@@ -1,4 +1,6 @@
+pub use http_body_util;
 pub use hyper;
+pub use hyper_util;
 
 pub use gateway_callbacks::*;
 pub use secret_getter::*;
@@ -7,9 +9,9 @@ pub use server::*;
 #[cfg(test)]
 mod _test_tools;
 
-mod body;
 mod gateway_callbacks;
 mod route_gateway;
 mod secret_getter;
 mod server;
 mod signing;
+mod sw_body;