reactor
/
ptth
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

♻️ refactor: error handling

main
_ 2021-04-27 21:39:42 -05:00
parent 33d07c45a8
commit 3240ad72b2
2 changed files with 62 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
crates/ptth_relay/src/errors.rs
View File

@ -55,7 +55,7 @@ pub enum HandleHttpResponseError {
#[derive (Debug, Error)] #[derive (Debug, Error)]
pub enum RequestError { pub enum RequestError {
#[error ("HTTP error")] #[error (transparent)]
Http (#[from] http::Error), Http (#[from] http::Error),
#[error ("MessagePack encode error")] #[error ("MessagePack encode error")]
@ -67,6 +67,21 @@ pub enum RequestError {
#[error ("Error handling HTTP response")] #[error ("Error handling HTTP response")]
HandleHttpResponse (#[from] HandleHttpResponseError), HandleHttpResponse (#[from] HandleHttpResponseError),
#[error ("Unknown server")]
UnknownServer,
#[error ("Bad request")]
BadRequest,
#[error ("Server never responded")]
ServerNeverResponded,
#[error ("Server timed out")]
ServerTimedOut,
#[error ("Relay shutting down")]
RelayShuttingDown,
#[error ("Error is mysterious!")] #[error ("Error is mysterious!")]
Mysterious, Mysterious,
} }

77
crates/ptth_relay/src/lib.rs
View File

@ -106,25 +106,26 @@ async fn handle_http_request (
state: Arc <Relay>, state: Arc <Relay>,
server_name: &str server_name: &str
) )
-> Result <Response <Body>, http::Error> -> Result <Response <Body>, RequestError>
{ {
use crate::relay_state::{ use crate::relay_state::{
AuditData, AuditData,
AuditEvent, AuditEvent,
}; };
use RequestError::*;
let req_method = req.method.clone (); let req_method = req.method.clone ();
{ {
let config = state.config.read ().await; let config = state.config.read ().await;
if ! config.servers.contains_key (server_name) { if ! config.servers.contains_key (server_name) {
return error_reply (StatusCode::NOT_FOUND, "Unknown server"); return Err (UnknownServer);
} }
} }
let req = match http_serde::RequestParts::from_hyper (req.method, uri, req.headers) { let req = match http_serde::RequestParts::from_hyper (req.method, uri, req.headers) {
Ok (x) => x, Ok (x) => x,
Err (_) => return error_reply (StatusCode::BAD_REQUEST, "Bad request"), Err (_) => return Err (BadRequest),
}; };
let (tx, rx) = oneshot::channel (); let (tx, rx) = oneshot::channel ();
@ -144,44 +145,48 @@ async fn handle_http_request (
state.park_client (server_name, req, &req_id).await; state.park_client (server_name, req, &req_id).await;
// UKAUFFY4 (Receive half)
let received = match tokio::time::timeout (Duration::from_secs (30), rx).await let received = match tokio::time::timeout (Duration::from_secs (30), rx).await
{ {
Ok (x) => x,
Err (_) => { Err (_) => {
debug! ("Timed out request {}", req_id); debug! ("Timed out request {}", req_id);
return error_reply (StatusCode::GATEWAY_TIMEOUT, "Remote server never responded"); return Err (ServerNeverResponded);
} }
Ok (x) => x,
}; };
// UKAUFFY4 (Receive half) let received = match received {
match received {
Ok (Ok ((parts, body))) => {
let mut resp = Response::builder ()
.status (hyper::StatusCode::from (parts.status_code));
if
req_method == hyper::Method::GET &&
parts.headers.get ("accept-ranges").is_some ()
{
trace! ("Stream restart code could go here");
}
for (k, v) in parts.headers {
resp = resp.header (&k, v);
}
debug! ("Unparked request {}", req_id);
resp.body (body)
},
Ok (Err (ShuttingDownError::ShuttingDown)) => {
error_reply (StatusCode::GATEWAY_TIMEOUT, "Relay shutting down")
},
Err (_) => { Err (_) => {
debug! ("Responder sender dropped for request {}", req_id); debug! ("Responder sender dropped for request {}", req_id);
error_reply (StatusCode::GATEWAY_TIMEOUT, "Remote server timed out") return Err (ServerTimedOut);
}, },
Ok (x) => x,
};
let (parts, body) = match received {
Err (ShuttingDownError::ShuttingDown) => {
return Err (RelayShuttingDown);
},
Ok (x) => x,
};
let mut resp = Response::builder ()
.status (hyper::StatusCode::from (parts.status_code));
if
req_method == hyper::Method::GET &&
parts.headers.get ("accept-ranges").is_some ()
{
trace! ("Stream restart code could go here");
} }
for (k, v) in parts.headers {
resp = resp.header (&k, v);
}
debug! ("Unparked request {}", req_id);
Ok (resp.body (body)?)
} }
#[derive (Debug, PartialEq)] #[derive (Debug, PartialEq)]
@ -637,7 +642,17 @@ pub async fn run_relay (
async { async {
Ok::<_, Infallible> (handle_all (req, state, handlebars).await.unwrap_or_else (|e| { Ok::<_, Infallible> (handle_all (req, state, handlebars).await.unwrap_or_else (|e| {
error! ("{}", e); error! ("{}", e);
error_reply (StatusCode::INTERNAL_SERVER_ERROR, "Error in relay").unwrap ()
let status_code = match &e {
UnknownServer => StatusCode::NOT_FOUND,
BadRequest => StatusCode::BAD_REQUEST,
ServerNeverResponded => StatusCode::GATEWAY_TIMEOUT,
ServerTimedOut => StatusCode::GATEWAY_TIMEOUT,
_ => StatusCode::INTERNAL_SERVER_ERROR,
};
error_reply (status_code, "Error in relay").unwrap ()
})) }))
} }
})) }))