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

♻️ Extract crate ptth_relay

main
_ 2020-11-27 00:20:18 +00:00
parent 84bb326f37
commit 4c9595ee2e
10 changed files with 49 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
Cargo.toml
View File

@ -14,7 +14,6 @@ aho-corasick = "0.7.14"
base64 = "0.12.3" base64 = "0.12.3"
blake3 = "0.3.7" blake3 = "0.3.7"
chrono = "0.4.19" chrono = "0.4.19"
dashmap = "3.11.10"
futures = "0.3.7" futures = "0.3.7"
handlebars = "3.5.1" handlebars = "3.5.1"
http = "0.2.1" http = "0.2.1"
@ -30,7 +29,6 @@ reqwest = { version = "0.10.8", features = ["stream"] }
rmp-serde = "0.14.4" rmp-serde = "0.14.4"
serde = {version = "1.0.117", features = ["derive"]} serde = {version = "1.0.117", features = ["derive"]}
structopt = "0.3.20" structopt = "0.3.20"
thiserror = "1.0.22"
tokio = { version = "0.2.22", features = ["full"] } tokio = { version = "0.2.22", features = ["full"] }
tracing = "0.1.21" tracing = "0.1.21"
tracing-futures = "0.2.4" tracing-futures = "0.2.4"
@ -41,6 +39,7 @@ url = "2.2.0"
always_equal = { path = "crates/always_equal" } always_equal = { path = "crates/always_equal" }
ptth_core = { path = "crates/ptth_core" } ptth_core = { path = "crates/ptth_core" }
ptth_relay = { path = "crates/ptth_relay" }
[workspace] [workspace]

7
crates/ptth_core/src/lib.rs
View File

@ -2,6 +2,13 @@ pub mod graceful_shutdown;
pub mod http_serde; pub mod http_serde;
pub mod prelude; pub mod prelude;
// It's easier if the server can stream its response body
// back to the relay un-changed inside its request body
// So we wrap the server's actual response head
// (status code, headers, etc.) in this one header field.
pub const PTTH_MAGIC_HEADER: &str = "X-PTTH-2LJYXWC4";
// The arguments are in order so they are in order overall: // The arguments are in order so they are in order overall:
// e.g. prefix_match ("/prefix", "/prefix/middle/suffix") -> "/middle/suffix" // e.g. prefix_match ("/prefix", "/prefix/middle/suffix") -> "/middle/suffix"

30
crates/ptth_relay/Cargo.toml Normal file
View File

@ -0,0 +1,30 @@
[package]
name = "ptth_relay"
version = "0.1.0"
authors = ["Trish"]
edition = "2018"
license = "AGPL-3.0"
[dependencies]
base64 = "0.12.3"
blake3 = "0.3.7"
chrono = "0.4.19"
dashmap = "3.11.10"
futures = "0.3.7"
handlebars = "3.5.1"
http = "0.2.1"
hyper = "0.13.8"
itertools = "0.9.0"
rmp-serde = "0.14.4"
serde = {version = "1.0.117", features = ["derive"]}
thiserror = "1.0.22"
tokio = { version = "0.2.22", features = ["full"] }
toml = "0.5.7"
tracing = "0.1.21"
tracing-futures = "0.2.4"
tracing-subscriber = "0.2.15"
ulid = "0.4.1"
ptth_core = { path = "../ptth_core" }

0
src/relay/config.rs → crates/ptth_relay/src/config.rs
View File

0
src/relay/git_version.rs → crates/ptth_relay/src/git_version.rs
View File

2
src/relay/mod.rs → crates/ptth_relay/src/lib.rs
View File

@ -256,7 +256,7 @@ async fn handle_http_response (
-> Response <Body> -> Response <Body>
{ {
let (parts, mut body) = req.into_parts (); let (parts, mut body) = req.into_parts ();
let resp_parts: http_serde::ResponseParts = rmp_serde::from_read_ref (&base64::decode (parts.headers.get (crate::PTTH_MAGIC_HEADER).unwrap ()).unwrap ()).unwrap (); let resp_parts: http_serde::ResponseParts = rmp_serde::from_read_ref (&base64::decode (parts.headers.get (ptth_core::PTTH_MAGIC_HEADER).unwrap ()).unwrap ()).unwrap ();
// Intercept the body packets here so we can check when the stream // Intercept the body packets here so we can check when the stream
// ends or errors out // ends or errors out

9
src/bin/ptth_relay.rs → crates/ptth_relay/src/main.rs
View File

@ -11,10 +11,11 @@ use tracing_subscriber::{
EnvFilter, EnvFilter,
}; };
use ptth::relay; use ptth_relay::{
use ptth::relay::{
Config, Config,
git_version::GIT_VERSION,
RelayState, RelayState,
run_relay,
}; };
#[tokio::main] #[tokio::main]
@ -28,12 +29,12 @@ async fn main () -> Result <(), Box <dyn Error>> {
let config_path = PathBuf::from ("config/ptth_relay.toml"); let config_path = PathBuf::from ("config/ptth_relay.toml");
let config = Config::from_file (&config_path).await?; let config = Config::from_file (&config_path).await?;
info! ("ptth_relay Git version: {:?}", ptth::relay::git_version::GIT_VERSION); info! ("ptth_relay Git version: {:?}", GIT_VERSION);
let (shutdown_rx, forced_shutdown) = ptth_core::graceful_shutdown::init_with_force (); let (shutdown_rx, forced_shutdown) = ptth_core::graceful_shutdown::init_with_force ();
forced_shutdown.wrap_server ( forced_shutdown.wrap_server (
relay::run_relay ( run_relay (
Arc::new (RelayState::from (config)), Arc::new (RelayState::from (config)),
shutdown_rx, shutdown_rx,
Some (config_path) Some (config_path)

8
src/lib.rs
View File

@ -1,11 +1,3 @@
// It's easier if the server can stream its response body
// back to the relay un-changed inside its request body
// So we wrap the server's actual response head
// (status code, headers, etc.) in this one header field.
pub const PTTH_MAGIC_HEADER: &str = "X-PTTH-2LJYXWC4";
pub mod relay;
pub mod server; pub mod server;
#[cfg (test)] #[cfg (test)]

2
src/server/mod.rs
View File

@ -90,7 +90,7 @@ async fn handle_req_resp <'a> (
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 (crate::PTTH_MAGIC_HEADER, base64::encode (rmp_serde::to_vec (&response.parts).unwrap ())); .header (ptth_core::PTTH_MAGIC_HEADER, base64::encode (rmp_serde::to_vec (&response.parts).unwrap ()));
if let Some (length) = response.content_length { if let Some (length) = response.content_length {
resp_req = resp_req.header ("Content-Length", length.to_string ()); resp_req = resp_req.header ("Content-Length", length.to_string ());

9
src/tests.rs
View File

@ -14,7 +14,6 @@ use tokio::{
}; };
use super::{ use super::{
relay,
server, server,
}; };
@ -36,22 +35,22 @@ fn end_to_end () {
let api_key = "AnacondaHardcoverGrannyUnlatchLankinessMutate"; let api_key = "AnacondaHardcoverGrannyUnlatchLankinessMutate";
let tripcode = base64::encode (blake3::hash (api_key.as_bytes ()).as_bytes ()); let tripcode = base64::encode (blake3::hash (api_key.as_bytes ()).as_bytes ());
debug! ("Relay is expecting tripcode {}", tripcode); debug! ("Relay is expecting tripcode {}", tripcode);
let config_file = relay::config::file::Config { let config_file = ptth_relay::config::file::Config {
port: None, port: None,
servers: hashmap! { servers: hashmap! {
server_name.into () => relay::config::file::Server { server_name.into () => ptth_relay::config::file::Server {
tripcode, tripcode,
display_name: None, display_name: None,
}, },
}, },
}; };
let relay_state = Arc::new (relay::RelayState::from (relay::config::Config::try_from (config_file).unwrap ())); let relay_state = Arc::new (ptth_relay::RelayState::from (ptth_relay::config::Config::try_from (config_file).unwrap ()));
let relay_state_2 = relay_state.clone (); let relay_state_2 = relay_state.clone ();
let (stop_relay_tx, stop_relay_rx) = oneshot::channel (); let (stop_relay_tx, stop_relay_rx) = oneshot::channel ();
let task_relay = spawn (async move { let task_relay = spawn (async move {
relay::run_relay (relay_state_2, stop_relay_rx, None).await.unwrap (); ptth_relay::run_relay (relay_state_2, stop_relay_rx, None).await.unwrap ();
}); });
assert! (relay_state.list_servers ().await.is_empty ()); assert! (relay_state.list_servers ().await.is_empty ());