micropelago/hyper-reverse-proxy
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Make examples/simple.rs compatible with hyper v1.

Browse Source
This commit is contained in:
Stefan Sundin 2024-05-16 20:48:41 -07:00
parent 7adb97ceaa
commit 88e08c98f1
2 changed files with 86 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
Cargo.toml
View File

@ -31,23 +31,19 @@ tokio = { version = "1.17.0", features = ["io-util", "rt"] }
tracing = "0.1.34" tracing = "0.1.34"
[dev-dependencies] [dev-dependencies]
hyper = { version = "1.2.0", features = ["client", "http1"] } hyper = { version = "1.2.0", features = ["client", "http1", "server"] }
futures = "0.3.21" futures = "0.3.21"
async-trait = "0.1.53" async-trait = "0.1.53"
async-tungstenite = { version = "0.17", features = ["tokio-runtime"] } async-tungstenite = { version = "0.17", features = ["tokio-runtime"] }
tokio-test = "0.4.2" tokio-test = "0.4.2"
test-context = "0.1.3" test-context = "0.1.3"
tokiotest-httpserver = "0.2.1" tokiotest-httpserver = "0.2.1"
hyper-trust-dns = { version = "0.4.2", features = [
"rustls-http2",
"dnssec-ring",
"dns-over-https-rustls",
"rustls-webpki"
] }
rand = "0.8.5" rand = "0.8.5"
tungstenite = "0.17" tungstenite = "0.17"
url = "2.2" url = "2.2"
criterion = "0.3.5" criterion = "0.3.5"
hyper-rustls = "0.27.1"
rustls = "0.23.6"
[features] [features]

113
examples/simple.rs
View File

@ -1,26 +1,51 @@
use hyper::server::conn::AddrStream; use std::convert::Infallible;
use hyper::service::{make_service_fn, service_fn}; use std::io;
use hyper::{Body, Request, Response, Server, StatusCode}; use std::net::{IpAddr, SocketAddr};
use std::time::Duration;
use http_body_util::combinators::UnsyncBoxBody;
use http_body_util::{BodyExt, Empty, Full};
use hyper::body::{Bytes, Incoming};
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Request, Response, StatusCode};
use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
use tokio::net::TcpListener;
use hyper_reverse_proxy::ReverseProxy; use hyper_reverse_proxy::ReverseProxy;
use hyper_trust_dns::{RustlsHttpsConnector, TrustDnsResolver}; use hyper_rustls::{ConfigBuilderExt, HttpsConnector};
use std::net::IpAddr; use hyper_util::client::legacy::connect::HttpConnector;
use std::{convert::Infallible, net::SocketAddr};
type Connector = HttpsConnector<HttpConnector>;
type ResponseBody = UnsyncBoxBody<Bytes, std::io::Error>;
lazy_static::lazy_static! { lazy_static::lazy_static! {
static ref PROXY_CLIENT: ReverseProxy<RustlsHttpsConnector> = { static ref PROXY_CLIENT: ReverseProxy<Connector> = {
ReverseProxy::new( let connector: Connector = Connector::builder()
hyper::Client::builder().build::<_, hyper::Body>(TrustDnsResolver::default().into_rustls_webpki_https_connector()), .with_tls_config(
) rustls::ClientConfig::builder()
.with_native_roots()
.expect("with_native_roots")
.with_no_client_auth(),
)
.https_or_http()
.enable_http1()
.build();
ReverseProxy::new(
hyper_util::client::legacy::Builder::new(TokioExecutor::new())
.pool_idle_timeout(Duration::from_secs(3))
.pool_timer(TokioTimer::new())
.build::<_, Incoming>(connector),
)
}; };
} }
fn debug_request(req: &Request<Body>) -> Result<Response<Body>, Infallible> { async fn handle(
let body_str = format!("{:?}", req); client_ip: IpAddr,
Ok(Response::new(Body::from(body_str))) req: Request<Incoming>,
} ) -> Result<Response<ResponseBody>, Infallible> {
let host = req.headers().get("host").and_then(|v| v.to_str().ok());
async fn handle(client_ip: IpAddr, req: Request<Body>) -> Result<Response<Body>, Infallible> { if host.is_some_and(|host| host.starts_with("service1.localhost")) {
if req.uri().path().starts_with("/target/first") {
match PROXY_CLIENT match PROXY_CLIENT
.call(client_ip, "http://127.0.0.1:13901", req) .call(client_ip, "http://127.0.0.1:13901", req)
.await .await
@ -28,10 +53,12 @@ async fn handle(client_ip: IpAddr, req: Request<Body>) -> Result<Response<Body>,
Ok(response) => Ok(response), Ok(response) => Ok(response),
Err(_error) => Ok(Response::builder() Err(_error) => Ok(Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR) .status(StatusCode::INTERNAL_SERVER_ERROR)
.body(Body::empty()) .body(UnsyncBoxBody::new(
Empty::<Bytes>::new().map_err(io::Error::other),
))
.unwrap()), .unwrap()),
} }
} else if req.uri().path().starts_with("/target/second") { } else if host.is_some_and(|host| host.starts_with("service2.localhost")) {
match PROXY_CLIENT match PROXY_CLIENT
.call(client_ip, "http://127.0.0.1:13902", req) .call(client_ip, "http://127.0.0.1:13902", req)
.await .await
@ -39,29 +66,55 @@ async fn handle(client_ip: IpAddr, req: Request<Body>) -> Result<Response<Body>,
Ok(response) => Ok(response), Ok(response) => Ok(response),
Err(_error) => Ok(Response::builder() Err(_error) => Ok(Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR) .status(StatusCode::INTERNAL_SERVER_ERROR)
.body(Body::empty()) .body(UnsyncBoxBody::new(
Empty::<Bytes>::new().map_err(io::Error::other),
))
.unwrap()), .unwrap()),
} }
} else { } else {
debug_request(&req) let body_str = format!("{:?}", req);
Ok(Response::new(UnsyncBoxBody::new(
Full::new(Bytes::from(body_str)).map_err(io::Error::other),
)))
} }
} }
#[tokio::main] #[tokio::main]
async fn main() { async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let bind_addr = "127.0.0.1:8000"; let bind_addr = "127.0.0.1:8000";
let addr: SocketAddr = bind_addr.parse().expect("Could not parse ip:port."); let addr: SocketAddr = bind_addr.parse().expect("Could not parse ip:port.");
let make_svc = make_service_fn(|conn: &AddrStream| { // We create a TcpListener and bind it to the address
let remote_addr = conn.remote_addr().ip(); let listener = TcpListener::bind(addr).await?;
async move { Ok::<_, Infallible>(service_fn(move |req| handle(remote_addr, req))) }
});
let server = Server::bind(&addr).serve(make_svc); println!(
"Access service1 on http://service1.localhost:{}",
addr.port()
);
println!(
"Access service2 on http://service2.localhost:{}",
addr.port()
);
println!("Running server on {:?}", addr); // We start a loop to continuously accept incoming connections
loop {
let (stream, remote_addr) = listener.accept().await?;
let client_ip = remote_addr.ip();
if let Err(e) = server.await { // Use an adapter to access something implementing `tokio::io` traits as if they implement
eprintln!("server error: {}", e); // `hyper::rt` IO traits.
let io = TokioIo::new(stream);
// Spawn a tokio task to serve multiple connections concurrently
tokio::task::spawn(async move {
// Finally, we bind the incoming connection to our `hello` service
if let Err(err) = http1::Builder::new()
// `service_fn` converts our function in a `Service`
.serve_connection(io, service_fn(move |req| handle(client_ip, req)))
.await
{
eprintln!("Error serving connection: {:?}", err);
}
});
} }
} }