aboutsummaryrefslogtreecommitdiff
path: root/src/web/web_server.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/web/web_server.rs')
-rw-r--r--src/web/web_server.rs110
1 files changed, 44 insertions, 66 deletions
diff --git a/src/web/web_server.rs b/src/web/web_server.rs
index 73780efb..766e3829 100644
--- a/src/web/web_server.rs
+++ b/src/web/web_server.rs
@@ -5,15 +5,12 @@ use std::{convert::Infallible, sync::Arc};
use futures::future::Future;
use hyper::{
+ body::Incoming as IncomingBody,
header::{HeaderValue, HOST},
- server::conn::AddrStream,
- service::{make_service_fn, service_fn},
- Body, Method, Request, Response, Server, StatusCode,
+ Method, Request, Response, StatusCode,
};
-use hyperlocal::UnixServerExt;
-
-use tokio::net::UnixStream;
+use tokio::net::{TcpListener, UnixListener};
use opentelemetry::{
global,
@@ -24,7 +21,8 @@ use opentelemetry::{
use crate::error::*;
-use garage_api::helpers::{authority_to_host, host_to_bucket};
+use garage_api::generic_server::{server_loop, UnixListenerOn};
+use garage_api::helpers::*;
use garage_api::s3::cors::{add_cors_headers, find_matching_cors_rule, handle_options_for_bucket};
use garage_api::s3::error::{
CommonErrorDerivative, Error as ApiError, OkOrBadRequest, OkOrInternalError,
@@ -74,78 +72,52 @@ pub struct WebServer {
impl WebServer {
/// Run a web server
- pub async fn run(
- garage: Arc<Garage>,
- addr: UnixOrTCPSocketAddress,
- root_domain: String,
- shutdown_signal: impl Future<Output = ()>,
- ) -> Result<(), GarageError> {
+ pub fn new(garage: Arc<Garage>, root_domain: String) -> Arc<Self> {
let metrics = Arc::new(WebMetrics::new());
- let web_server = Arc::new(WebServer {
+ Arc::new(WebServer {
garage,
metrics,
root_domain,
- });
-
- let tcp_service = make_service_fn(|conn: &AddrStream| {
- let web_server = web_server.clone();
-
- let client_addr = conn.remote_addr();
- async move {
- Ok::<_, Error>(service_fn(move |req: Request<Body>| {
- let web_server = web_server.clone();
-
- web_server.handle_request(req, client_addr.to_string())
- }))
- }
- });
-
- let unix_service = make_service_fn(|_: &UnixStream| {
- let web_server = web_server.clone();
-
- let path = addr.to_string();
- async move {
- Ok::<_, Error>(service_fn(move |req: Request<Body>| {
- let web_server = web_server.clone();
-
- web_server.handle_request(req, path.clone())
- }))
- }
- });
+ })
+ }
- info!("Web server listening on {}", addr);
+ pub async fn run(
+ self: Arc<Self>,
+ bind_addr: UnixOrTCPSocketAddress,
+ shutdown_signal: impl Future<Output = ()>,
+ ) -> Result<(), GarageError> {
+ info!("Web server listening on {}", bind_addr);
- match addr {
+ match bind_addr {
UnixOrTCPSocketAddress::TCPSocket(addr) => {
- Server::bind(&addr)
- .serve(tcp_service)
- .with_graceful_shutdown(shutdown_signal)
- .await?
+ let listener = TcpListener::bind(addr).await?;
+
+ let handler =
+ move |stream, socketaddr| self.clone().handle_request(stream, socketaddr);
+ server_loop(listener, handler, shutdown_signal).await
}
UnixOrTCPSocketAddress::UnixSocket(ref path) => {
if path.exists() {
fs::remove_file(path)?
}
- let bound = Server::bind_unix(path)?;
+ let listener = UnixListener::bind(path)?;
+ let listener = UnixListenerOn(listener, path.display().to_string());
fs::set_permissions(path, Permissions::from_mode(0o222))?;
- bound
- .serve(unix_service)
- .with_graceful_shutdown(shutdown_signal)
- .await?;
+ let handler =
+ move |stream, socketaddr| self.clone().handle_request(stream, socketaddr);
+ server_loop(listener, handler, shutdown_signal).await
}
- };
-
- Ok(())
+ }
}
async fn handle_request(
self: Arc<Self>,
- req: Request<Body>,
+ req: Request<IncomingBody>,
addr: String,
- ) -> Result<Response<Body>, Infallible> {
+ ) -> Result<Response<BoxBody<Error>>, http::Error> {
if let Ok(forwarded_for_ip_addr) =
forwarded_headers::handle_forwarded_for_headers(req.headers())
{
@@ -187,7 +159,8 @@ impl WebServer {
match res {
Ok(res) => {
debug!("{} {} {}", req.method(), res.status(), req.uri());
- Ok(res)
+ Ok(res
+ .map(|body| BoxBody::new(http_body_util::BodyExt::map_err(body, Error::from))))
}
Err(error) => {
info!(
@@ -220,7 +193,10 @@ impl WebServer {
Ok(exists)
}
- async fn serve_file(self: &Arc<Self>, req: &Request<Body>) -> Result<Response<Body>, Error> {
+ async fn serve_file(
+ self: &Arc<Self>,
+ req: &Request<IncomingBody>,
+ ) -> Result<Response<BoxBody<ApiError>>, Error> {
// Get http authority string (eg. [::1]:3902 or garage.tld:80)
let authority = req
.headers()
@@ -267,9 +243,11 @@ impl WebServer {
);
let ret_doc = match *req.method() {
- Method::OPTIONS => handle_options_for_bucket(req, &bucket),
- Method::HEAD => handle_head(self.garage.clone(), req, bucket_id, &key, None).await,
- Method::GET => handle_get(self.garage.clone(), req, bucket_id, &key, None).await,
+ Method::OPTIONS => handle_options_for_bucket(req, &bucket)
+ .map_err(ApiError::from)
+ .map(|res| res.map(|_empty_body: EmptyBody| empty_body())),
+ Method::HEAD => handle_head(self.garage.clone(), &req, bucket_id, &key, None).await,
+ Method::GET => handle_get(self.garage.clone(), &req, bucket_id, &key, None).await,
_ => Err(ApiError::bad_request("HTTP method not supported")),
};
@@ -281,7 +259,7 @@ impl WebServer {
Ok(Response::builder()
.status(StatusCode::FOUND)
.header("Location", url)
- .body(Body::empty())
+ .body(empty_body())
.unwrap())
}
_ => ret_doc,
@@ -310,7 +288,7 @@ impl WebServer {
// Create a fake HTTP request with path = the error document
let req2 = Request::builder()
.uri(format!("http://{}/{}", host, &error_document))
- .body(Body::empty())
+ .body(empty_body::<Infallible>())
.unwrap();
match handle_get(self.garage.clone(), &req2, bucket_id, &error_document, None).await
@@ -358,7 +336,7 @@ impl WebServer {
}
}
-fn error_to_res(e: Error) -> Response<Body> {
+fn error_to_res(e: Error) -> Response<BoxBody<Error>> {
// If we are here, it is either that:
// - there was an error before trying to get the requested URL
// from the bucket (e.g. bucket not found)
@@ -366,7 +344,7 @@ fn error_to_res(e: Error) -> Response<Body> {
// was a HEAD request or we couldn't get the error document)
// We do NOT enter this code path when returning the bucket's
// error document (this is handled in serve_file)
- let body = Body::from(format!("{}\n", e));
+ let body = string_body(format!("{}\n", e));
let mut http_error = Response::new(body);
*http_error.status_mut() = e.http_status_code();
e.add_headers(http_error.headers_mut());
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-02 00:25:54 +0000