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

♻️ refactor: extract routing module

main
_ 2021-04-10 09:38:53 -05:00
parent 6a84d2dbd5
commit 6b71b32cf5
3 changed files with 67 additions and 96 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -97,7 +97,7 @@ async fn handle_http_request (
req: http::request::Parts, req: http::request::Parts,
uri: String, uri: String,
state: Arc <RelayState>, state: Arc <RelayState>,
server_name: String server_name: &str
) )
-> Result <Response <Body>, http::Error> -> Result <Response <Body>, http::Error>
{ {
@ -108,7 +108,7 @@ async fn handle_http_request (
{ {
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 error_reply (StatusCode::NOT_FOUND, "Unknown server");
} }
} }
@ -124,7 +124,7 @@ async fn handle_http_request (
state.audit_log.push (AuditEvent::new (AuditData::WebClientGet { state.audit_log.push (AuditEvent::new (AuditData::WebClientGet {
req_id: req_id.clone (), req_id: req_id.clone (),
server_name: server_name.clone (), server_name: server_name.to_string (),
})).await; })).await;
trace! ("Created request {}", req_id); trace! ("Created request {}", req_id);
@ -143,7 +143,7 @@ async fn handle_http_request (
req, req,
}; };
let new_rendezvous = match request_rendezvous.remove (&server_name) { let new_rendezvous = match request_rendezvous.remove (server_name) {
Some (ParkedClients (mut v)) => { Some (ParkedClients (mut v)) => {
debug! ("Parking request {} ({} already queued)", req_id, v.len ()); debug! ("Parking request {} ({} already queued)", req_id, v.len ());
v.push (wrapped); v.push (wrapped);
@ -179,7 +179,7 @@ async fn handle_http_request (
}, },
}; };
request_rendezvous.insert (server_name, new_rendezvous); request_rendezvous.insert (server_name.to_string (), new_rendezvous);
} }
let timeout = tokio::time::sleep (std::time::Duration::from_secs (30)); let timeout = tokio::time::sleep (std::time::Duration::from_secs (30));
@ -470,7 +470,7 @@ async fn handle_endless_source (gib: usize, throttle: Option <usize>)
.body (Body::wrap_stream (ReceiverStream::new (rx))) .body (Body::wrap_stream (ReceiverStream::new (rx)))
} }
async fn handle_gen_scraper_key (state: Arc <RelayState>) async fn handle_gen_scraper_key (_state: Arc <RelayState>)
-> Result <Response <Body>, http::Error> -> Result <Response <Body>, http::Error>
{ {
let key = ptth_core::gen_key (); let key = ptth_core::gen_key ();
@ -491,97 +491,68 @@ async fn handle_all (
) )
-> Result <Response <Body>, RequestError> -> Result <Response <Body>, RequestError>
{ {
let path = req.uri ().path ().to_string (); use routing::Route::*;
//println! ("{}", path);
// The path is cloned here, so it's okay to consume the request
// later.
let path = req.uri ().path ().to_string ();
trace! ("Request path: {}", path); trace! ("Request path: {}", path);
if req.method () == Method::POST { let route = routing::route_url (req.method (), &path);
// This is stuff the server can use. Clients can't let response = match route {
// POST right now ClientAuditLog => handle_audit_log (state, handlebars).await?,
ClientRelayIsUp => error_reply (StatusCode::OK, "Relay is up")?,
return if let Some (request_code) = prefix_match ("/7ZSFUKGV/http_response/", &path) { ClientServerGet {
let request_code = request_code.into (); listen_code,
Ok (server_endpoint::handle_response (req, state, request_code).await?) path,
} } => {
else if path == "/frontend/debug/endless_sink" {
Ok (handle_endless_sink (req).await?)
}
else {
error! ("Can't POST {}", path);
Ok (error_reply (StatusCode::BAD_REQUEST, "Can't POST this")?)
};
}
if let Some (listen_code) = prefix_match ("/7ZSFUKGV/http_listen/", &path) {
let api_key = req.headers ().get ("X-ApiKey");
let api_key = match api_key {
None => return Ok (error_reply (StatusCode::FORBIDDEN, "Can't run server without an API key")?),
Some (x) => x,
};
server_endpoint::handle_listen (state, listen_code.into (), api_key.as_bytes ()).await
}
else if let Some (rest) = prefix_match ("/frontend/servers/", &path) {
// DRY T4H76LB3
if rest.is_empty () {
Ok (handle_server_list (state, handlebars).await?)
}
else if let Some (idx) = rest.find ('/') {
let listen_code = String::from (&rest [0..idx]);
let path = String::from (&rest [idx..]);
let (parts, _) = req.into_parts (); let (parts, _) = req.into_parts ();
Ok (handle_http_request (parts, path, state, listen_code).await?) handle_http_request (parts, path.to_string (), state, listen_code).await?
} },
else { ClientServerList => handle_server_list (state, handlebars).await?,
Ok (error_reply (StatusCode::BAD_REQUEST, "Bad URI format")?) ClientUnregisteredServers => handle_unregistered_servers (state, handlebars).await?,
} Debug => {
}
else if path == "/frontend/unregistered_servers" {
Ok (handle_unregistered_servers (state, handlebars).await?)
}
else if path == "/frontend/audit_log" {
Ok (handle_audit_log (state, handlebars).await?)
}
else if let Some (rest) = prefix_match ("/frontend/debug/", &path) {
if rest.is_empty () {
let s = handlebars.render ("debug", &())?; let s = handlebars.render ("debug", &())?;
Ok (ok_reply (s)?) ok_reply (s)?
} },
else if rest == "endless_source" { DebugEndlessSink => handle_endless_sink (req).await?,
Ok (handle_endless_source (1, None).await?) DebugEndlessSource (throttle) => handle_endless_source (1, throttle).await?,
} DebugGenKey => handle_gen_scraper_key (state).await?,
else if rest == "endless_source_throttled" { DebugMysteriousError => return Err (RequestError::Mysterious),
Ok (handle_endless_source (1, Some (1024 / 64)).await?) ErrorBadUriFormat => error_reply (StatusCode::BAD_REQUEST, "Bad URI format")?,
} ErrorCantPost => {
else if rest == "endless_sink" { error! ("Can't POST {}", path);
Ok (error_reply (StatusCode::METHOD_NOT_ALLOWED, "Don't GET this URL, POST to it.")?) error_reply (StatusCode::BAD_REQUEST, "Can't POST this")?
} },
else if rest == "gen_key" { ErrorMethodNotAllowed => error_reply (StatusCode::METHOD_NOT_ALLOWED, "Method not allowed. Are you POST-ing to a GET-only url, or vice versa?")?,
Ok (handle_gen_scraper_key (state).await?) ErrorRoutingFailed => error_reply (StatusCode::OK, "URL routing failed")?,
} Root => {
else { let s = handlebars.render ("root", &())?;
Ok (error_reply (StatusCode::NOT_FOUND, "Can't route URL")?) ok_reply (s)?
} },
} Scraper {
else if path == "/" { rest,
let s = handlebars.render ("root", &())?; } => scraper_api::handle (req, state, rest).await?,
Ok (ok_reply (s)?) ServerHttpListen {
} listen_code,
else if path == "/frontend/relay_up_check" { } => {
Ok (error_reply (StatusCode::OK, "Relay is up")?) let api_key = req.headers ().get ("X-ApiKey");
}
else if path == "/frontend/test_mysterious_error" { let api_key = match api_key {
Err (RequestError::Mysterious) None => return Ok (error_reply (StatusCode::FORBIDDEN, "Can't run server without an API key")?),
} Some (x) => x,
else if let Some (rest) = prefix_match ("/scraper/", &path) { };
scraper_api::handle (req, state, rest).await server_endpoint::handle_listen (state, listen_code.into (), api_key.as_bytes ()).await?
} },
else { ServerHttpResponse {
Ok (error_reply (StatusCode::OK, "Can't route URL")?) request_code,
} } => {
let request_code = request_code.into ();
server_endpoint::handle_response (req, state, request_code).await?
},
};
Ok (response)
} }
pub fn load_templates (asset_root: &Path) pub fn load_templates (asset_root: &Path)

8
crates/ptth_relay/src/routing.rs
View File

@ -12,7 +12,7 @@ pub enum Route <'a> {
ClientUnregisteredServers, ClientUnregisteredServers,
Debug, Debug,
DebugEndlessSink, DebugEndlessSink,
DebugEndlessSource (Option <u64>), DebugEndlessSource (Option <usize>),
DebugGenKey, DebugGenKey,
DebugMysteriousError, DebugMysteriousError,
ErrorBadUriFormat, ErrorBadUriFormat,
@ -31,7 +31,7 @@ pub enum Route <'a> {
}, },
} }
pub fn route_url (method: Method, path: &str) -> Route { pub fn route_url <'a> (method: &Method, path: &'a str) -> Route <'a> {
if method == Method::POST { if method == Method::POST {
return if let Some (request_code) = path.strip_prefix ("/7ZSFUKGV/http_response/") { return if let Some (request_code) = path.strip_prefix ("/7ZSFUKGV/http_response/") {
Route::ServerHttpResponse { Route::ServerHttpResponse {
@ -124,14 +124,14 @@ mod tests {
for (input, expected) in vec! [ for (input, expected) in vec! [
("/", Route::Root), ("/", Route::Root),
].into_iter () { ].into_iter () {
let actual = route_url (Method::GET, input); let actual = route_url (&Method::GET, input);
assert_eq! (actual, expected); assert_eq! (actual, expected);
} }
for (input, expected) in vec! [ for (input, expected) in vec! [
("/", Route::ErrorCantPost), ("/", Route::ErrorCantPost),
].into_iter () { ].into_iter () {
let actual = route_url (Method::POST, input); let actual = route_url (&Method::POST, input);
assert_eq! (actual, expected); assert_eq! (actual, expected);
} }
} }

2
crates/ptth_relay/src/scraper_api.rs
View File

@ -171,7 +171,7 @@ async fn api_v1 (
// This is ugly. I don't like having scraper_api know about the // This is ugly. I don't like having scraper_api know about the
// crate root. // crate root.
Ok (crate::handle_http_request (parts, path, state, listen_code).await?) Ok (crate::handle_http_request (parts, path, state, &listen_code).await?)
} }
else { else {
Ok (error_reply (StatusCode::BAD_REQUEST, "Bad URI format")?) Ok (error_reply (StatusCode::BAD_REQUEST, "Bad URI format")?)