diff --git a/Cargo.toml b/Cargo.toml index 808f158..c2b2d27 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -38,6 +38,10 @@ harness = false name = "deviceatlas" harness = false +[[example]] +name = "maxmind" +harness = false + [dependencies] hyper = { version = "1.2", features = ["full"] } hyper-tls = { version = "0.6.0" } @@ -59,6 +63,7 @@ serde-inline-default = "0.2.3" enum_dispatch = "0.3.13" foyer = "0.17" postcard = "1.1.3" +reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "http2"] } [dev-dependencies] tonic = "0.11" diff --git a/examples/maxmind/config.yaml b/examples/maxmind/config.yaml new file mode 100644 index 0000000..44e4802 --- /dev/null +++ b/examples/maxmind/config.yaml @@ -0,0 +1,31 @@ +listening_port: 8080 +prometheus_port: 8081 +healthcheck_port: 8082 +minimum_log_level: info + +middlewares: + # Just emit a log line that you can find in the standard output. + - log: + level: info + value: 'Cache status: $cache_status, Forward target: $forward_target' + # This will try to serve the request from cache 'files'. + - use_cache: + cache_name: files + when: + status_code_is: 200 + # This will forward the request to the origin server if there is a cache miss. + # The url will be :/// + - forward_request: + scheme: https + target_host: download.maxmind.com + +caches: + # The hybrid cache saves cached responses both in memory and on disk. + # Saving to disk is useful to persist cache across pod restarts, provided you have a persistent volume mounted. + - hybrid: + name: files + path: /tmp/cacheus/maxmind + ttl_seconds: 86400 + hash_path: false + hash_query: true + hash_body: false diff --git a/examples/maxmind/main.rs b/examples/maxmind/main.rs new file mode 100644 index 0000000..a876357 --- /dev/null +++ b/examples/maxmind/main.rs @@ -0,0 +1,17 @@ +use cacheus::{self, CacheusServer}; +use futures::join; +use log::info; +use tokio::signal; + +#[tokio::main] +async fn main() +{ + let cacheus_fut = CacheusServer::start_from_config_file("examples/maxmind/config.yaml"); + + join!(cacheus_fut); + + // Wait for SIGTERM or SIGINT + signal::ctrl_c().await.expect("failed to listen for ctrl_c"); + + info!("Shutting down Cacheus server..."); +} diff --git a/src/buffered_body.rs b/src/buffered_body.rs index 162d905..96ca8c1 100644 --- a/src/buffered_body.rs +++ b/src/buffered_body.rs @@ -57,6 +57,12 @@ pub struct BufferedBody } impl BufferedBody { + /// Returns a read-only view of the raw buffered body bytes. + /// This is different from `to_bytes`, which serializes with metadata for persistence. + pub fn body_bytes(&self) -> &[u8] { + self.bufs.as_ref() + } + pub fn to_bytes(&self) -> Vec { let mut v = Vec::new(); diff --git a/src/config/middleware.rs b/src/config/middleware.rs index 78856ea..6f75481 100644 --- a/src/config/middleware.rs +++ b/src/config/middleware.rs @@ -1,11 +1,10 @@ use std::{str::FromStr, sync::Arc}; use hyper::{header::{HeaderName, HeaderValue}, Request, Response, Uri}; -use hyper_tls::HttpsConnector; -use hyper_util::client::legacy::{Client, connect::HttpConnector}; use serde::{Deserialize, Serialize}; use serde_inline_default::serde_inline_default; -use crate::{CallContext, buffered_body::BufferedBody, config::{CacheConfig, cache::CachedResponse, conditions::Condition}, executor::TokioExecutor}; +use crate::{CallContext, buffered_body::BufferedBody, config::{CacheConfig, cache::CachedResponse, conditions::Condition}}; +use reqwest as rw; trait Middleware { async fn init(&mut self) {} @@ -174,42 +173,62 @@ impl Middleware for CacheMiddleware { } #[derive(Debug, Serialize, Deserialize, Clone)] +#[serde(default)] pub struct Forward { pub target_host: String, - #[serde(default)] pub enforce_http2: bool, - #[serde(default)] pub enforce_http: bool, - #[serde(default)] + /// Nagle algorithm can cause delays in sending small packets. + /// Default is disabled. pub use_nagle: bool, - #[serde(default)] + /// The maximum number of redirects to follow. Set 0 to disable redirects. + /// Default is 3. + pub max_redirects: u8, pub scheme: Option, - #[serde(default)] pub when: Option, #[serde(skip)] - pub client: Option, BufferedBody>>>, + pub client: Option>, +} + +impl Default for Forward { + fn default() -> Self { + Self { + target_host: "".to_string(), + enforce_http2: false, + enforce_http: false, + use_nagle: false, + max_redirects: 3, + scheme: None, + when: None, + client: None, + } + } } impl Middleware for Forward { async fn init(&mut self) { + let mut builder = rw::Client::builder(); - let mut http = HttpConnector::new(); - http.set_nodelay(!self.use_nagle); - http.enforce_http(self.enforce_http); - let connector = HttpsConnector::new_with_connector(http); - - let client = Client::builder(TokioExecutor) - .http2_only(self.enforce_http2) - // .pool_max_idle_per_host(configuration.max_idle_connections_per_host as usize) - // .http2_max_send_buf_size(128_000_000) - // .timer(hyper_util::rt::TokioTimer::new()) - // .pool_timer(hyper_util::rt::TokioTimer::new()) - // .pool_idle_timeout(std::time::Duration::from_secs(90)) - // .http2_keep_alive_interval(Some(Duration::from_secs(300))) - // .retry_canceled_requests(false) - .set_host(false) - .build(connector); + if self.max_redirects > 0 { + builder = builder.redirect(rw::redirect::Policy::limited(3)); + } else { + builder = builder.redirect(rw::redirect::Policy::none()); + } + + builder = builder.tcp_nodelay(!self.use_nagle); + + // Try to honor HTTP/2 vs HTTP/1 preferences when specified. + if self.enforce_http2 { + builder = builder.http2_prior_knowledge(); + } + if self.enforce_http { + builder = builder.http1_only(); + } + + let client = builder + .build() + .expect("Failed to build reqwest client"); self.client = Some(Arc::new(client)); } @@ -236,21 +255,51 @@ impl Middleware for Forward { context.variables.insert("$forward_target".to_string(), target_uri.to_string()); - let mut forwarded_req = Request::builder() - .method(request.method()) - .uri(target_uri) - .version(request.version()) - .body(request.body().clone()).unwrap(); - - let headers = forwarded_req.headers_mut(); - headers.extend(request.headers().iter().map(|(k, v)| (k.clone(), v.clone()))); - headers.insert("host", target_host.parse().unwrap()); - headers.remove("accept-encoding"); - - let res = self.client.as_ref().unwrap().request(forwarded_req).await.expect("Failed to send request"); - let (parts, body) = res.into_parts(); - let buffered_body = BufferedBody::collect_buffered(body).await.unwrap(); - return Some(Response::from_parts(parts, buffered_body)); + // Build reqwest request + let url = target_uri.to_string(); + let client = self.client.as_ref().expect("Client not initialized"); + let mut req_builder = client.request( + reqwest::Method::from_bytes(request.method().as_str().as_bytes()).unwrap(), + &url, + ); + + // Copy headers, skipping hop-by-hop or ones managed by reqwest + // - Skip "host"; reqwest sets it from URL + // - Skip "accept-encoding" to avoid auto-compression issues + // - Content-Length will be set by reqwest based on body + for (name, value) in request.headers().iter() { + let name_str = name.as_str(); + if name_str.eq_ignore_ascii_case("host") || name_str.eq_ignore_ascii_case("accept-encoding") || name_str.eq_ignore_ascii_case("content-length") { + continue; + } + req_builder = req_builder.header(name, value); + } + + // Attach body + let body_bytes = request.body().body_bytes(); + if !body_bytes.is_empty() { + req_builder = req_builder.body(body_bytes.to_vec()); + } + + let res = req_builder.send().await.expect("Failed to send request"); + + // Map back to hyper::Response + let status = res.status(); + let mut builder = Response::builder().status(status); + + // Headers + { + let headers_mut = builder.headers_mut().unwrap(); + for (k, v) in res.headers().iter() { + // reqwest's HeaderName/Value are re-exported from http, compatible with hyper + headers_mut.insert(k.clone(), v.clone()); + } + } + + let bytes = res.bytes().await.expect("Failed to read response body"); + let body = BufferedBody::from_body(&bytes); + let response = builder.body(body).expect("Failed to build response"); + return Some(response); } }