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

♻️ refactor: moving things around to support third-party ptth_server modules

main
_ 2021-05-09 17:54:29 +00:00
parent 93bf38ec03
commit c663d126aa
2 changed files with 20 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
crates/ptth_server/Cargo.toml
View File

@ -6,9 +6,10 @@ authors = ["Trish"]
edition = "2018" edition = "2018"
license = "AGPL-3.0" license = "AGPL-3.0"
description = "The server for PTTH"
repository = "https://github.com/ReactorScram/ptth"
categories = ["command-line-utilities", "web-programming::http-server"] categories = ["command-line-utilities", "web-programming::http-server"]
description = "The server for PTTH"
documentation = "https://docs.rs/ptth_server/"
repository = "https://github.com/ReactorScram/ptth"
[dependencies] [dependencies]

33
crates/ptth_server/src/lib.rs
View File

@ -81,20 +81,11 @@ struct State {
// When file_server responds, wrap it back up and stream it to the relay. // When file_server responds, wrap it back up and stream it to the relay.
async fn handle_one_req ( async fn handle_one_req (
state: &Arc <State>, state: &State,
wrapped_req: http_serde::WrappedRequest req_id: String,
response: http_serde::Response,
) -> Result <(), ServerError> ) -> Result <(), ServerError>
{ {
let (req_id, parts) = (wrapped_req.id, wrapped_req.req);
debug! ("Handling request {}", req_id);
let response = state.file_server.serve_all (
parts.method,
&parts.uri,
&parts.headers,
).await?;
let mut resp_req = state.client let mut resp_req = state.client
.post (&format! ("{}/http_response/{}", state.config.relay_url, req_id)) .post (&format! ("{}/http_response/{}", state.config.relay_url, req_id))
.header (ptth_core::PTTH_MAGIC_HEADER, base64::encode (rmp_serde::to_vec (&response.parts).map_err (ServerError::MessagePackEncodeResponse)?)); .header (ptth_core::PTTH_MAGIC_HEADER, base64::encode (rmp_serde::to_vec (&response.parts).map_err (ServerError::MessagePackEncodeResponse)?));
@ -153,7 +144,7 @@ async fn handle_one_req (
Ok::<(), ServerError> (()) Ok::<(), ServerError> (())
} }
async fn handle_req_resp ( async fn handle_requests (
state: &Arc <State>, state: &Arc <State>,
req_resp: reqwest::Response req_resp: reqwest::Response
) -> Result <(), ServerError> ) -> Result <(), ServerError>
@ -173,12 +164,22 @@ async fn handle_req_resp (
debug! ("Unwrapped {} requests", wrapped_reqs.len ()); debug! ("Unwrapped {} requests", wrapped_reqs.len ());
for wrapped_req in wrapped_reqs { for wrapped_req in wrapped_reqs {
let state = state.clone (); let state = Arc::clone (&state);
// These have to detach, so we won't be able to catch the join errors. // These have to detach, so we won't be able to catch the join errors.
tokio::spawn (async move { tokio::spawn (async move {
handle_one_req (&state, wrapped_req).await let (req_id, parts) = (wrapped_req.id, wrapped_req.req);
debug! ("Handling request {}", req_id);
let response = state.file_server.serve_all (
parts.method,
&parts.uri,
&parts.headers,
).await?;
handle_one_req (&state, req_id, response).await
}); });
} }
@ -372,7 +373,7 @@ async fn run_server_loop (
// Unpack the requests, spawn them into new tasks, then loop back // Unpack the requests, spawn them into new tasks, then loop back
// around. // around.
if handle_req_resp (&state, req_resp).await.is_err () { if handle_requests (&state, req_resp).await.is_err () {
backoff_delay = err_backoff_delay; backoff_delay = err_backoff_delay;
continue; continue;
} }