diff --git a/Cargo.toml b/Cargo.toml index 707f2881eb..1f5b051ff6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -67,6 +67,7 @@ tokio = { version = "1", features = [ ] } tokio-test = "0.4" tokio-util = "0.7.10" +tower = { version = "0.5.2", default-features = false, features = ["limit", "timeout", "util"] } [features] # Nothing by default @@ -288,6 +289,11 @@ name = "single_threaded" path = "examples/single_threaded.rs" required-features = ["full"] +[[example]] +name = "tower_layers" +path = "examples/tower_layers.rs" +required-features = ["full"] + [[example]] name = "state" path = "examples/state.rs" diff --git a/examples/README.md b/examples/README.md index de38911e9c..c45e8a661f 100644 --- a/examples/README.md +++ b/examples/README.md @@ -22,6 +22,7 @@ serde_json = "1.0" form_urlencoded = "1" http = "1" futures-util = { version = "0.3", default-features = false } +tower = { version = "0.5", features = ["limit", "timeout", "util"] } ``` ## Getting Started @@ -58,6 +59,8 @@ futures-util = { version = "0.3", default-features = false } * [`state`](state.rs) - A webserver showing basic state sharing among requests. A counter is shared, incremented for every request, and every response is sent the last count. +* [`tower_layers`](tower_layers.rs) - A server whose handler is wrapped in a stack of `tower` layers, with the small adapter needed to hand a `tower::Service` to `serve_connection`. + * [`upgrades`](upgrades.rs) - A server and client demonstrating how to do HTTP upgrades (such as WebSockets). * [`web_api`](web_api.rs) - A server consisting in a service that returns incoming POST request's content in the response in uppercase and a service that calls the first service and includes the first service response in its own response. diff --git a/examples/tower_layers.rs b/examples/tower_layers.rs new file mode 100644 index 0000000000..ee4c85999d --- /dev/null +++ b/examples/tower_layers.rs @@ -0,0 +1,109 @@ +#![deny(warnings)] + +use std::convert::Infallible; +use std::net::SocketAddr; +use std::time::Duration; + +use bytes::Bytes; +use http_body_util::Full; +use hyper::server::conn::http1; +use hyper::{body::Incoming, Request, Response}; +use tokio::net::TcpListener; +use tower::{ServiceBuilder, ServiceExt}; + +// This would normally come from the `hyper-util` crate, but we can't depend +// on that here because it would be a cyclical dependency. +#[path = "../benches/support/mod.rs"] +mod support; +use support::{TokioIo, TokioTimer}; + +// The same handler as in `hello.rs`. `tower::service_fn` turns it into a +// `tower::Service`, which is what a `tower::Layer` knows how to wrap. +async fn hello(_: Request) -> Result>, Infallible> { + Ok(Response::new(Full::new(Bytes::from("Hello World!")))) +} + +// `tower::Service` and `hyper::service::Service` are different traits, so a +// tower stack cannot be given to `serve_connection` as it is. They differ in +// two ways: +// +// - tower's `call` takes `&mut self` while hyper's takes `&self`, so the tower +// service has to be cloned for each request. +// - a tower service must be driven to readiness with `poll_ready` before every +// `call`, and hyper has no `poll_ready` to drive it from. Layers such as +// `concurrency_limit` acquire their resources there, so skipping it would +// break them. +// +// `ServiceExt::oneshot` does both: it takes a service by value, polls it until +// it is ready, and then calls it. +// +// `hyper_util::service::TowerToHyperService` is this same adapter and is what +// to use in a real application. It cannot be used here because hyper cannot +// depend on hyper-util. +#[derive(Clone)] +struct TowerToHyperService { + service: S, +} + +impl hyper::service::Service for TowerToHyperService +where + S: tower::Service + Clone, +{ + type Response = S::Response; + type Error = S::Error; + type Future = tower::util::Oneshot; + + fn call(&self, req: R) -> Self::Future { + self.service.clone().oneshot(req) + } +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + pretty_env_logger::init(); + + // This address is localhost + let addr: SocketAddr = ([127, 0, 0, 1], 3000).into(); + + // Layers wrap outside-in: a request passes the concurrency limit, then the + // timeout, and only then reaches `hello`. + // + // The stack is built here and not inside the accept loop, so that the + // semaphore behind `concurrency_limit` is shared by every connection. + // Building it per connection would limit each connection on its own. + // + // The concurrency limit takes its permit in `poll_ready`, which the adapter + // above awaits before `timeout` starts its timer in `call`. The timeout + // therefore bounds the handler and not the time spent waiting for a permit, + // whichever order the two layers are written in. + let service = ServiceBuilder::new() + .concurrency_limit(128) + .timeout(Duration::from_secs(10)) + .service(tower::service_fn(hello)); + let service = TowerToHyperService { service }; + + let listener = TcpListener::bind(addr).await?; + println!("Listening on http://{}", addr); + + loop { + let (tcp, _) = listener.accept().await?; + let io = TokioIo::new(tcp); + + // Cloning the stack is cheap and keeps the shared semaphore. + let service = service.clone(); + + tokio::task::spawn(async move { + // An error from the service, an elapsed timeout for instance, + // aborts the connection. A server that would rather answer with a + // 503 or a 504 needs one more layer turning the error into a + // response. + if let Err(err) = http1::Builder::new() + .timer(TokioTimer::new()) + .serve_connection(io, service) + .await + { + println!("Error serving connection: {:?}", err); + } + }); + } +} diff --git a/src/service/mod.rs b/src/service/mod.rs index 0e49c45bd7..0c89b94218 100644 --- a/src/service/mod.rs +++ b/src/service/mod.rs @@ -20,6 +20,17 @@ //! The helper [`service_fn`] should be sufficient for most cases, but //! if you need to implement `Service` for a type manually, you can follow the example //! in `service_struct_impl.rs`. +//! +//! # Middleware +//! +//! This trait is not [`tower::Service`][tower], so a tower stack has to be wrapped in +//! [`hyper_util::service::TowerToHyperService`][t2h] before a server can be given it. +//! The `tower_layers.rs` example shows the shape, and the [middleware guide][guide] +//! goes further. +//! +//! [tower]: https://docs.rs/tower/latest/tower/trait.Service.html +//! [t2h]: https://docs.rs/hyper-util/latest/hyper_util/service/struct.TowerToHyperService.html +//! [guide]: https://hyper.rs/guides/1/server/middleware/ mod http; mod service;