WIP generic

Author: LMG

josh-proxy/src/bin/josh-proxy.rs

@@ -4,12 +4,17 @@ extern crate clap;
 
 use bytes::Bytes;
 use clap::Parser;
+use http_body_util::combinators::BoxBody;
 use http_body_util::Full;
 use hyper::body::Incoming;
 use hyper::server::conn::http1;
 use hyper_util::rt::{tokio::TokioIo, tokio::TokioTimer};
 use josh_proxy::cli;
-use josh_proxy::{run_git_with_auth, FetchError, MetaConfig, RemoteAuth, RepoConfig, RepoUpdate};
+use josh_proxy::juniper_hyper::graphql;
+use josh_proxy::{
+    body::{empty, full},
+    run_git_with_auth, FetchError, MetaConfig, RemoteAuth, RepoConfig, RepoUpdate,
+};
 use opentelemetry::global;
 use opentelemetry::sdk::propagation::TraceContextPropagator;
 use tokio::pin;
@@ -297,7 +302,7 @@ async fn fetch_upstream(
 async fn static_paths(
     service: &JoshProxyService,
     path: &str,
-) -> josh::JoshResult<Option<Response<Full<Bytes>>>> {
+) -> josh::JoshResult<Option<Response<BoxBody<Bytes, hyper::Error>>>> {
     tracing::debug!("static_path {:?}", path);
     if path == "/version" {
         return Ok(Some(make_response(
@@ -356,7 +361,7 @@ async fn static_paths(
 async fn repo_update_fn(
     serv: Arc<JoshProxyService>,
     req: Request<Incoming>,
-) -> josh::JoshResult<Response<Full<Bytes>>> {
+) -> josh::JoshResult<Response<BoxBody<Bytes, hyper::Error>>> {
     let body = req.into_body().collect().await?.to_bytes();
 
     let s = tracing::span!(tracing::Level::TRACE, "repo update worker");
@@ -378,10 +383,10 @@ async fn repo_update_fn(
     Ok(match result {
         Ok(stderr) => Response::builder()
             .status(hyper::StatusCode::OK)
-            .body(Full::new(Bytes::from(stderr))),
+            .body(full(stderr)),
         Err(josh::JoshError(stderr)) => Response::builder()
             .status(hyper::StatusCode::INTERNAL_SERVER_ERROR)
-            .body(Full::new(Bytes::from(stderr))),
+            .body(full(stderr)),
     }?)
 }
 
@@ -509,19 +514,19 @@ async fn do_filter(
     Ok(())
 }
 
-fn make_response(body: &str, code: hyper::StatusCode) -> Response<Full<Bytes>> {
+fn make_response(body: &str, code: hyper::StatusCode) -> Response<BoxBody<Bytes, hyper::Error>> {
     let owned_body = body.to_owned();
     Response::builder()
         .status(code)
         .header(hyper::header::CONTENT_TYPE, "text/plain")
-        .body(Full::new(Bytes::from(owned_body)))
+        .body(full(owned_body))
         .expect("Can't build response")
 }
 
 async fn handle_ui_request(
     req: Request<Incoming>,
     resource_path: &str,
-) -> josh::JoshResult<Response<Full<Bytes>>> {
+) -> josh::JoshResult<Response<BoxBody<Bytes, hyper::Error>>> {
     /*
     // Proxy: can be used for UI development or to serve a different UI
     if let Some(proxy) = &ARGS.static_resource_proxy_target {
@@ -530,7 +535,7 @@ async fn handle_ui_request(
             Ok(response) => Ok(response),
             Err(error) => Ok(Response::builder()
                 .status(StatusCode::INTERNAL_SERVER_ERROR)
-                .body(Full::new(Bytes::from(format!("Proxy error: {:?}", error))))
+                .body(full(format!("Proxy error: {:?}", error)))
                 .unwrap()),
         };
     }*/
@@ -553,7 +558,7 @@ async fn handle_ui_request(
     let resolver = hyper_staticfile::Resolver::new("josh/static");
     let request = hyper::http::Request::get(resolve_path).body(()).unwrap();
     let result = resolver.resolve_request(&request).await?;
-    let response = hyper::Response::new(Full::new(
+    let response = hyper::Response::new(full(
         hyper_staticfile::ResponseBuilder::new()
             .request(&req)
             .build(result)?
@@ -932,7 +937,7 @@ fn head_ref_or_default(head_ref: &str) -> HeadRef {
 async fn handle_serve_namespace_request(
     serv: Arc<JoshProxyService>,
     req: Request<Incoming>,
-) -> josh::JoshResult<Response<Full<Bytes>>> {
+) -> josh::JoshResult<Response<BoxBody<Bytes, hyper::Error>>> {
     let error_response = |status: StatusCode| Ok(make_response("", status));
 
     if req.method() != hyper::Method::POST {
@@ -1126,7 +1131,7 @@ async fn handle_serve_namespace_request(
 async fn call_service(
     serv: Arc<JoshProxyService>,
     req_auth: (josh_proxy::auth::Handle, Request<Incoming>),
-) -> josh::JoshResult<Response<Full<Bytes>>> {
+) -> josh::JoshResult<Response<BoxBody<Bytes, hyper::Error>>> {
     let (auth, req) = req_auth;
 
     let path = {
@@ -1194,7 +1199,7 @@ async fn call_service(
             return Ok(Response::builder()
                 .status(hyper::StatusCode::FOUND)
                 .header("Location", redirect_path)
-                .body(Full::new(Bytes::new()))?);
+                .body(empty())?);
         }
     };
 
@@ -1232,7 +1237,7 @@ async fn call_service(
         return Ok(Response::builder()
             .status(hyper::StatusCode::TEMPORARY_REDIRECT)
             .header("Location", format!("{}{}", remote_url, parsed_url.pathinfo))
-            .body(Full::new(Bytes::new()))?);
+            .body(empty())?);
     }
 
     if is_repo_blocked(&meta) {
@@ -1260,7 +1265,7 @@ async fn call_service(
                 "Basic realm=User Visible Realm",
             )
             .status(hyper::StatusCode::UNAUTHORIZED);
-        return Ok(builder.body(Full::new(Bytes::new()))?);
+        return Ok(builder.body(empty())?);
     }
 
     if parsed_url.api == "/~/graphql" {
@@ -1269,11 +1274,13 @@ async fn call_service(
 
     if parsed_url.api == "/~/graphiql" {
         let addr = format!("/~/graphql{}", meta.config.repo);
-        return Ok(tokio::task::spawn_blocking(move || {
-            josh_proxy::juniper_hyper::graphiql(&addr, None)
-        })
-        .in_current_span()
-        .await??);
+        let resp =
+            tokio::task::spawn(
+                async move { josh_proxy::juniper_hyper::graphiql(&addr, None).await },
+            )
+            .in_current_span()
+            .await?;
+        return Ok(into_body(resp));
     }
 
     let headref = head_ref_or_default(&parsed_url.headref);
@@ -1297,11 +1304,11 @@ async fn call_service(
                     "Basic realm=User Visible Realm",
                 )
                 .status(hyper::StatusCode::UNAUTHORIZED);
-            return Ok(builder.body(Full::new(Bytes::new()))?);
+            return Ok(builder.body(empty())?);
         }
         Err(FetchError::Other(e)) => {
             let builder = Response::builder().status(hyper::StatusCode::INTERNAL_SERVER_ERROR);
-            return Ok(builder.body(Full::new(Bytes::from(e.0)))?);
+            return Ok(builder.body(full(e.0))?);
         }
     }
 
@@ -1382,7 +1389,9 @@ async fn call_service(
     // it is executed in all cases.
     std::mem::drop(temp_ns);
 
-    Ok(cgires.0)
+    Ok(cgires
+        .0
+        .map(|body| body.map_err(|never| match never {}).boxed()))
 }
 
 async fn serve_query(
@@ -1391,7 +1400,7 @@ async fn serve_query(
     upstream_repo: String,
     filter: josh::filter::Filter,
     head_ref: &str,
-) -> josh::JoshResult<Response<Full<Bytes>>> {
+) -> josh::JoshResult<Response<BoxBody<Bytes, hyper::Error>>> {
     let tracing_span = tracing::span!(tracing::Level::TRACE, "render worker");
     let head_ref = head_ref.to_string();
     let res = tokio::task::spawn_blocking(move || -> josh::JoshResult<_> {
@@ -1434,15 +1443,15 @@ async fn serve_query(
                     .get("content-type")
                     .unwrap_or(&"text/plain".to_string()),
             )
-            .body(Full::new(Bytes::from(res)))?,
+            .body(full(res))?,
 
         Ok(None) => Response::builder()
             .status(hyper::StatusCode::NOT_FOUND)
-            .body(Full::new(Bytes::from("File not found".to_string())))?,
+            .body(full("File not found".to_string()))?,
 
         Err(res) => Response::builder()
             .status(hyper::StatusCode::UNPROCESSABLE_ENTITY)
-            .body(Full::new(Bytes::from(res.to_string())))?,
+            .body(full(res.to_string()))?,
     })
 }
 
@@ -1756,16 +1765,8 @@ async fn serve_graphql(
     upstream_repo: String,
     upstream: String,
     auth: josh_proxy::auth::Handle,
-) -> josh::JoshResult<Response<Full<Bytes>>> {
+) -> josh::JoshResult<Response<BoxBody<Bytes, hyper::Error>>> {
     let remote_url = upstream.clone() + upstream_repo.as_str();
-    let parsed = match josh_proxy::juniper_hyper::parse_req(req).await {
-        Ok(r) => r,
-        Err(resp) => {
-            return Ok(hyper::Response::new(Full::new(Bytes::from(
-                resp.collect().await?.to_bytes(),
-            ))))
-        }
-    };
 
     let transaction_mirror = josh::cache::Transaction::open(
         &serv.repo_path.join("mirror"),
@@ -1796,12 +1797,16 @@ async fn serve_graphql(
         // First attempt to serve GraphQL query. If we can serve it
         // that means all requested revisions were specified by SHA and we could find
         // all of them locally, so no need to fetch.
-        let res = parsed.execute(&root_node, &context).await;
+        let res = graphql(root_node, context, req).await;
+
+        if !res.status().is_success() {
+            return Ok(res.map(|body| body.map_err(|never| match never {}).boxed()));
+        }
 
         // The "allow_refs" flag will be set by the query handler if we need to do a fetch
         // to complete the query.
         if !*context.allow_refs.lock().unwrap() {
-            res
+            Ok(res)
         } else {
             match fetch_upstream(
                 serv.clone(),
@@ -1823,16 +1828,16 @@ async fn serve_graphql(
                             "Basic realm=User Visible Realm",
                         )
                         .status(hyper::StatusCode::UNAUTHORIZED);
-                    return Ok(builder.body(Full::new(Bytes::new()))?);
+                    return Ok(builder.body(empty())?);
                 }
                 Err(FetchError::Other(e)) => {
                     let builder =
                         Response::builder().status(hyper::StatusCode::INTERNAL_SERVER_ERROR);
-                    return Ok(builder.body(Full::new(Bytes::from(e.0)))?);
+                    return Ok(builder.body(full(e.0))?);
                 }
             };
 
-            parsed.execute(&root_node, &context).await
+            Ok(graphql(root_node, context, req).await)
         }
     };
 
@@ -1842,8 +1847,8 @@ async fn serve_graphql(
         hyper::StatusCode::BAD_REQUEST
     };
 
-    let body = Full::new(Bytes::from(serde_json::to_string_pretty(&res).unwrap()));
-    let mut resp = Response::new(Full::new(Bytes::new()));
+    let body = full(serde_json::to_string_pretty(&res).unwrap());
+    let mut resp = Response::new(empty());
     *resp.status_mut() = code;
     resp.headers_mut().insert(
         hyper::header::CONTENT_TYPE,

josh-proxy/src/body.rs

@@ -0,0 +1,28 @@
+use http_body_util::{combinators::BoxBody, BodyExt, Empty, Full};
+use hyper::{
+    body::{Body, Bytes},
+    Response,
+};
+
+pub fn empty() -> BoxBody<Bytes, hyper::Error> {
+    Empty::<Bytes>::new()
+        .map_err(|never| match never {})
+        .boxed()
+}
+pub fn full<T: Into<Bytes>>(chunk: T) -> BoxBody<Bytes, hyper::Error> {
+    Full::new(chunk.into())
+        .map_err(|never| match never {})
+        .boxed()
+}
+
+pub fn into_box_body<T>(resp: &Response<T>) -> Response<BoxBody<Bytes, hyper::Error>>
+where
+    T: Body,
+    Bytes: From<T>,
+{
+    resp.map(|body| {
+        <Bytes as Into<Bytes>>::into(body)
+            .map_err(|e| e.into())
+            .boxed()
+    })
+}