reverse proxy expirementation

Cargo.toml

@@ -7,6 +7,7 @@ members = [
     "io",
     "postgres",
     "postgres-codegen",
+    "reverse-proxy",
     "router",
     "server",
     "service",

client/src/body.rs

@@ -49,7 +49,7 @@ impl fmt::Debug for ResponseBody<'_> {
 }
 
 impl ResponseBody<'_> {
-    pub(crate) fn into_owned(self) -> ResponseBody<'static> {
+    pub fn into_owned(self) -> ResponseBody<'static> {
         match self {
             #[cfg(feature = "http1")]
             Self::H1(body) => ResponseBody::H1Owned(body.map_conn(Into::into)),
@@ -101,7 +101,7 @@ impl Stream for ResponseBody<'_> {
 }
 
 /// type erased stream body.
-pub struct BoxBody(Pin<Box<dyn Stream<Item = Result<Bytes, BodyError>> + Send + 'static>>);
+pub struct BoxBody(Pin<Box<dyn Stream<Item = Result<Bytes, BodyError>> + 'static>>);
 
 impl Default for BoxBody {
     fn default() -> Self {
@@ -113,7 +113,7 @@ impl BoxBody {
     #[inline]
     pub fn new<B, E>(body: B) -> Self
     where
-        B: Stream<Item = Result<Bytes, E>> + Send + 'static,
+        B: Stream<Item = Result<Bytes, E>> + 'static,
         E: Into<BodyError>,
     {
         Self(Box::pin(BoxStreamMapErr { body }))

client/src/client.rs

@@ -77,7 +77,7 @@ impl Client {
     #[inline]
     pub fn request<B, E>(&self, req: http::Request<B>) -> RequestBuilder<'_>
     where
-        B: Stream<Item = Result<Bytes, E>> + Send + 'static,
+        B: Stream<Item = Result<Bytes, E>> + 'static,
         BodyError: From<E>,
     {
         RequestBuilder::new(req, self)
@@ -391,8 +391,6 @@ mod test {
     #[tokio::test]
     async fn connect_google() {
         let res = Client::builder()
-            .middleware(crate::middleware::FollowRedirect::new)
-            .middleware(crate::middleware::Decompress::new)
             .openssl()
             .finish()
             .get("https://www.google.com/")

client/src/request.rs

@@ -81,7 +81,7 @@ impl<'a> RequestBuilder<'a, marker::Http> {
     #[inline]
     pub fn stream<B, E>(self, body: B) -> Self
     where
-        B: Stream<Item = Result<Bytes, E>> + Send + 'static,
+        B: Stream<Item = Result<Bytes, E>> + 'static,
         E: Into<BodyError>,
     {
         self.map_body(body)
@@ -96,7 +96,7 @@ impl<'a> RequestBuilder<'a, marker::Http> {
 impl<'a, M> RequestBuilder<'a, M> {
     pub(crate) fn new<B, E>(req: http::Request<B>, client: &'a Client) -> Self
     where
-        B: Stream<Item = Result<Bytes, E>> + Send + 'static,
+        B: Stream<Item = Result<Bytes, E>> + 'static,
         E: Into<BodyError>,
     {
         Self {
@@ -212,7 +212,7 @@ impl<'a, M> RequestBuilder<'a, M> {
 
     fn map_body<B, E>(mut self, b: B) -> RequestBuilder<'a, M>
     where
-        B: Stream<Item = Result<Bytes, E>> + Send + 'static,
+        B: Stream<Item = Result<Bytes, E>> + 'static,
         E: Into<BodyError>,
     {
         self.req = self.req.map(|_| BoxBody::new(b));

client/src/response.rs

@@ -9,7 +9,7 @@ use futures_core::stream::Stream;
 use tokio::time::{Instant, Sleep};
 use tracing::debug;
 use xitca_http::{bytes::BytesMut, http};
-
+use xitca_http::http::response::Parts;
 use crate::{
     body::ResponseBody,
     error::{Error, TimeoutError},
@@ -85,6 +85,11 @@ impl<'a, const PAYLOAD_LIMIT: usize> Response<'a, PAYLOAD_LIMIT> {
             timeout: dur,
         }
     }
+    /// Collect response body as String. Response is consumed.
+    #[inline]
+    pub fn into_parts(self) -> (Parts, ResponseBody<'a>) {
+        self.res.into_parts()
+    }
 
     /// Collect response body as String. Response is consumed.
     #[inline]

client/src/service.rs

@@ -11,14 +11,14 @@ use crate::{
     uri::Uri,
 };
 
-type BoxFuture<'f, T, E> = Pin<Box<dyn Future<Output = Result<T, E>> + Send + 'f>>;
+type BoxFuture<'f, T, E> = Pin<Box<dyn Future<Output = Result<T, E>> + 'f>>;
 
 /// trait for composable http services. Used for middleware,resolver and tls connector.
 pub trait Service<Req> {
     type Response;
     type Error;
 
-    fn call(&self, req: Req) -> impl Future<Output = Result<Self::Response, Self::Error>> + Send;
+    fn call(&self, req: Req) -> impl Future<Output = Result<Self::Response, Self::Error>>;
 }
 
 pub trait ServiceDyn<Req> {
@@ -48,8 +48,7 @@ where
 
 impl<I, Req> Service<Req> for Box<I>
 where
-    Req: Send,
-    I: ServiceDyn<Req> + ?Sized + Send + Sync,
+    I: ServiceDyn<Req> + ?Sized,
 {
     type Response = I::Response;
     type Error = I::Error;
@@ -72,7 +71,7 @@ pub struct ServiceRequest<'r, 'c> {
 
 /// type alias for object safe wrapper of type implement [Service] trait.
 pub type HttpService =
-    Box<dyn for<'r, 'c> ServiceDyn<ServiceRequest<'r, 'c>, Response = Response<'c>, Error = Error> + Send + Sync>;
+    Box<dyn for<'r, 'c> ServiceDyn<ServiceRequest<'r, 'c>, Response = Response<'c>, Error = Error>>;
 
 pub(crate) fn base_service() -> HttpService {
     struct HttpService;

reverse-proxy/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "redirectionio-xitca-proxy"
+version = "0.1.0"
+authors = ["Joel Wurtz <[email protected]>"]
+edition = "2021"
+description = "Xitca web reverse HTTP and Websocket proxy"
+
+[dependencies]
+bytes = "1.9.0"
+lazy_static = "1.5.0"
+xitca-client = { version = "0.1.0", features = ["dangerous", "openssl"] }
+xitca-http = "0.7.0"
+xitca-web = "0.7.0"
+
+[dev-dependencies]
+tokio = { version = "1.42.0", features = ["full"] }
+
+[[example]]
+name = "http_proxy"

reverse-proxy/examples/http_proxy.rs

@@ -0,0 +1,21 @@
+use redirectionio_xitca_proxy::{HttpPeer, Proxy};
+use std::net::ToSocketAddrs;
+use xitca_web::App;
+
+#[tokio::main]
+async fn main() -> std::io::Result<()> {
+    let address = "github.com:443"
+        .to_socket_addrs()
+        .expect("error getting addresses")
+        .next()
+        .expect("cannot get address");
+
+    App::new()
+        .at("", Proxy::new(HttpPeer::new(address, "github.com:443").tls(true)))
+        .serve()
+        .bind("127.0.0.1:8080")?
+        .run()
+        .await?;
+
+    Ok(())
+}

reverse-proxy/src/error.rs

@@ -0,0 +1,31 @@
+use crate::forwarder::ForwardError;
+use std::error::Error;
+use std::fmt;
+use xitca_web::error::Error as XitcaError;
+
+#[derive(Debug)]
+pub enum ProxyError {
+    CannotReadRequestBody(XitcaError),
+    ForwardError(ForwardError),
+    NoPeer,
+}
+
+impl fmt::Display for ProxyError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::CannotReadRequestBody(e) => write!(f, "error when reading request body: {}", e),
+            Self::ForwardError(e) => write!(f, "error when forwarding request: {}", e),
+            Self::NoPeer => f.write_str("no peer found"),
+        }
+    }
+}
+
+impl Error for ProxyError {
+    fn source(&self) -> Option<&(dyn Error + 'static)> {
+        match self {
+            Self::CannotReadRequestBody(err) => Some(err),
+            Self::ForwardError(err) => Some(err),
+            Self::NoPeer => None,
+        }
+    }
+}

reverse-proxy/src/forwarder/error.rs

@@ -0,0 +1,32 @@
+use std::{error::Error, fmt};
+
+/// Errors that can result from using a connector service.
+#[derive(Debug)]
+pub enum ForwardError {
+    /// Failed to build a request from origin
+    UriError(xitca_web::http::Error),
+}
+
+impl fmt::Display for ForwardError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::UriError(_) => f.write_str("could not build request from origin"),
+        }
+    }
+}
+
+impl Error for ForwardError {
+    fn source(&self) -> Option<&(dyn Error + 'static)> {
+        match self {
+            Self::UriError(err) => Some(err),
+        }
+    }
+}
+
+impl ForwardError {
+    pub fn into_error_status(self) -> xitca_web::error::ErrorStatus {
+        match self {
+            Self::UriError(_) => xitca_web::error::ErrorStatus::bad_request(),
+        }
+    }
+}