rws_tower/tower/
mod.rs

1//! Tower interoperability (`tower` feature): adapters between rws's own
2//! [`Application`]/[`Middleware`](crate::middleware::Middleware) and the
3//! [Tower](https://docs.rs/tower) `Service`/`Layer` traits, so tonic gRPC
4//! services and tower-http layers work in an rws app, and an rws app works in
5//! a Tower stack. rws itself doesn't run on Tower: its request path stays the
6//! native one, and nothing here is compiled without the feature.
7//!
8//! - [`TowerApp`] / [`mount`] — a Tower service (a tonic server, a
9//!   tower-http `ServeDir`, an axum `Router`) as an rws [`Application`].
10//! - [`TowerLayer`] — a Tower layer (tower-http's `SetResponseHeader`,
11//!   `ValidateRequestHeader`, `Timeout`, …) as rws middleware.
12//! - [`TowerService`] — an rws app as a Tower service, for hyper or axum.
13//!
14//! Requests and responses cross as `http::Request<Body>` /
15//! `http::Response<Body>` ([`Body`] implements `http_body::Body`).
16//!
17//! Tower futures run on one process-wide runtime, `rws-tower` (2 threads):
18//! tokio I/O and timers belong to the runtime that made them, and an rws
19//! handler may run under the shared runtime, a per-core one or none at all.
20//! Its threads have no trace context, so automatic client spans
21//! ([`crate::otel`]) don't nest under requests a Tower service answers.
22
23#[cfg(test)]
24mod tests;
25
26mod layer;
27mod service;
28
29pub use layer::{Next, TowerLayer};
30pub use service::TowerService;
31
32use std::future::Future;
33use std::io::Read;
34use std::pin::Pin;
35use std::sync::OnceLock;
36use std::task::{Context, Poll};
37
38use bytes::Bytes;
39use http_body::Frame;
40use http_body_util::BodyExt;
41
42use crate::application::{Application, AsyncResponse};
43use crate::core::New;
44use crate::header::Header;
45use crate::range::Range;
46use crate::request::Request;
47use crate::response::{Response, Trailers};
48use crate::server::ConnectionInfo;
49
50/// The error type bodies and services are converted to.
51pub type BoxError = Box<dyn std::error::Error + Send + Sync>;
52
53/// Bodies up to this size with a known length are collected (not streamed)
54/// when they come back from a Tower service, unless they're gRPC.
55const BUFFER_LIMIT: u64 = 1024 * 1024;
56
57/// Bytes per chunk when an rws body (a file, a pipe, a streamed request) is
58/// read into a [`Body`].
59const CHUNK: usize = 64 * 1024;
60
61/// Hop-by-hop and framing headers rws regenerates; never copied across.
62const NOT_COPIED: &[&str] = &["connection", "keep-alive", "transfer-encoding", "upgrade", "proxy-connection", "te", "content-length"];
63
64#[doc(hidden)]
65pub fn runtime() -> &'static tokio::runtime::Runtime {
66    static RUNTIME: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
67    RUNTIME.get_or_init(|| {
68        tokio::runtime::Builder::new_multi_thread()
69            .worker_threads(2)
70            .thread_name("rws-tower")
71            .enable_all()
72            .build()
73            .expect("the tower runtime")
74    })
75}
76
77// ── Body ──────────────────────────────────────────────────────────────────────
78
79/// The `http_body::Body` rws hands to Tower and gets back from its own
80/// handlers: bytes in one frame, or chunks (and trailers) from a channel.
81pub struct Body {
82    inner: Inner,
83}
84
85enum Inner {
86    Full(Option<Bytes>),
87    Channel { rx: tokio::sync::mpsc::Receiver<std::io::Result<Bytes>>, trailers: Option<Trailers> },
88}
89
90impl Body {
91    /// No bytes.
92    pub fn empty() -> Body {
93        Body { inner: Inner::Full(None) }
94    }
95
96    /// `bytes` in one frame.
97    pub fn full(bytes: impl Into<Bytes>) -> Body {
98        let bytes = bytes.into();
99        Body { inner: Inner::Full((!bytes.is_empty()).then_some(bytes)) }
100    }
101
102    /// Chunks read from `reader` on the `rws-tower` blocking pool; `trailers`,
103    /// when given, are sent after the last chunk (read once the reader ends).
104    fn from_reader(mut reader: Box<dyn Read + Send>, trailers: Option<Trailers>) -> Body {
105        let (tx, rx) = tokio::sync::mpsc::channel(4);
106        runtime().spawn_blocking(move || loop {
107            let mut buf = vec![0u8; CHUNK];
108            match reader.read(&mut buf) {
109                Ok(0) => return,
110                Ok(n) => {
111                    buf.truncate(n);
112                    if tx.blocking_send(Ok(Bytes::from(buf))).is_err() {
113                        return;
114                    }
115                }
116                Err(e) => {
117                    let _ = tx.blocking_send(Err(e));
118                    return;
119                }
120            }
121        });
122        Body { inner: Inner::Channel { rx, trailers } }
123    }
124
125    /// An rws request body: bytes as they are, a stream read as it arrives.
126    pub fn from_request_body(body: crate::request::Body) -> Body {
127        if body.is_stream() {
128            Body::from_reader(body.into_reader(), None)
129        } else {
130            Body::full(body.to_vec())
131        }
132    }
133}
134
135impl Default for Body {
136    fn default() -> Body {
137        Body::empty()
138    }
139}
140
141impl http_body::Body for Body {
142    type Data = Bytes;
143    type Error = BoxError;
144
145    fn poll_frame(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Result<Frame<Bytes>, BoxError>>> {
146        match &mut self.get_mut().inner {
147            Inner::Full(bytes) => Poll::Ready(bytes.take().map(|b| Ok(Frame::data(b)))),
148            Inner::Channel { rx, trailers } => match rx.poll_recv(cx) {
149                Poll::Ready(Some(Ok(chunk))) => Poll::Ready(Some(Ok(Frame::data(chunk)))),
150                Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(Box::new(e)))),
151                Poll::Ready(None) => {
152                    let fields = trailers.take().map(|t| t.get()).unwrap_or_default();
153                    if fields.is_empty() {
154                        return Poll::Ready(None);
155                    }
156                    let mut map = http::HeaderMap::new();
157                    for h in fields {
158                        if let (Ok(name), Ok(value)) = (http::HeaderName::from_bytes(h.name.as_bytes()), http::HeaderValue::from_bytes(h.value.as_bytes())) {
159                            map.append(name, value);
160                        }
161                    }
162                    Poll::Ready(Some(Ok(Frame::trailers(map))))
163                }
164                Poll::Pending => Poll::Pending,
165            },
166        }
167    }
168
169    fn is_end_stream(&self) -> bool {
170        matches!(self.inner, Inner::Full(None))
171    }
172
173    fn size_hint(&self) -> http_body::SizeHint {
174        match &self.inner {
175            Inner::Full(bytes) => http_body::SizeHint::with_exact(bytes.as_ref().map_or(0, |b| b.len() as u64)),
176            Inner::Channel { .. } => http_body::SizeHint::default(),
177        }
178    }
179}
180
181// ── conversions ───────────────────────────────────────────────────────────────
182
183fn version_to_http(version: &str) -> http::Version {
184    match version {
185        "HTTP/1.0" => http::Version::HTTP_10,
186        "HTTP/2.0" | "HTTP/2" => http::Version::HTTP_2,
187        "HTTP/3.0" | "HTTP/3" => http::Version::HTTP_3,
188        "HTTP/0.9" => http::Version::HTTP_09,
189        _ => http::Version::HTTP_11,
190    }
191}
192
193fn version_from_http(version: http::Version) -> &'static str {
194    let v = &crate::http::VERSION;
195    match version {
196        http::Version::HTTP_10 => v.http_1_0,
197        http::Version::HTTP_2 => v.http_2_0,
198        http::Version::HTTP_3 => v.http_3_0,
199        _ => v.http_1_1,
200    }
201}
202
203fn header_map(headers: &[Header]) -> http::HeaderMap {
204    let mut map = http::HeaderMap::with_capacity(headers.len());
205    for h in headers {
206        // Anything `http` refuses is left out rather than failing the request.
207        if let (Ok(name), Ok(value)) = (http::HeaderName::from_bytes(h.name.as_bytes()), http::HeaderValue::from_bytes(h.value.as_bytes())) {
208            map.append(name, value);
209        }
210    }
211    map
212}
213
214fn headers_from(map: &http::HeaderMap) -> Vec<Header> {
215    map.iter()
216        .filter_map(|(name, value)| {
217            let value = String::from_utf8_lossy(value.as_bytes()).into_owned();
218            Some(Header::new(name.as_str().to_string(), value))
219        })
220        .collect()
221}
222
223/// An rws request as an `http::Request` (body as given).
224pub fn to_http_request(request: Request) -> http::Request<Body> {
225    let mut builder = http::Request::builder()
226        .method(http::Method::from_bytes(request.method.as_bytes()).unwrap_or(http::Method::GET))
227        .uri(http::Uri::try_from(request.request_uri.as_str()).unwrap_or_else(|_| http::Uri::from_static("/")))
228        .version(version_to_http(&request.http_version));
229    if let Some(headers) = builder.headers_mut() {
230        *headers = header_map(&request.headers);
231    }
232    builder.body(Body::from_request_body(request.body)).expect("method, URI and version are valid")
233}
234
235/// An `http::Request` as an rws request, its body read into memory.
236pub async fn from_http_request<B>(request: http::Request<B>) -> Result<Request, BoxError>
237where
238    B: http_body::Body<Data = Bytes>,
239    B::Error: Into<BoxError>,
240{
241    let (parts, body) = request.into_parts();
242    let bytes = body.collect().await.map_err(Into::into)?.to_bytes();
243    Ok(Request {
244        method: parts.method.as_str().to_string(),
245        request_uri: parts.uri.path_and_query().map(|p| p.as_str().to_string()).unwrap_or_else(|| "/".to_string()),
246        http_version: version_from_http(parts.version).to_string(),
247        headers: headers_from(&parts.headers),
248        body: bytes.to_vec().into(),
249    })
250}
251
252/// An rws response as an `http::Response`: buffered bodies in one frame,
253/// files and pipes streamed (read on the `rws-tower` blocking pool), with
254/// `Response::trailers` after them.
255pub fn to_http_response(mut response: Response) -> http::Response<Body> {
256    let mut headers = header_map(&response.headers);
257    let body = if response.stream_pipe.is_some() || response.stream_file.is_some() {
258        let _enter = runtime().enter();
259        match crate::server::body_stream::spawn_producer(&mut response) {
260            Some(mut chunks) => {
261                let (tx, rx) = tokio::sync::mpsc::channel(4);
262                runtime().spawn(async move {
263                    while let Some(chunk) = chunks.recv().await {
264                        let chunk = chunk.map(Bytes::from).map_err(std::io::Error::other);
265                        let failed = chunk.is_err();
266                        if tx.send(chunk).await.is_err() || failed {
267                            return;
268                        }
269                    }
270                });
271                Body { inner: Inner::Channel { rx, trailers: response.trailers.take() } }
272            }
273            None => Body::empty(),
274        }
275    } else {
276        let ranges = std::mem::take(&mut response.content_range_list);
277        let (content_type, bytes) = match ranges.len() {
278            0 => (None, Vec::new()),
279            1 => {
280                if let Some(value) = crate::response::content_range_header_value(response.status_code, &ranges[0]) {
281                    if let Ok(value) = http::HeaderValue::from_str(&value) {
282                        headers.insert(http::header::CONTENT_RANGE, value);
283                    }
284                }
285                let range = ranges.into_iter().next().expect("one range");
286                (Some(range.content_type), range.body)
287            }
288            _ => (Some(Range::multipart_content_type(&ranges)), Response::generate_body(ranges)),
289        };
290        if let Some(content_type) = content_type.and_then(|t| http::HeaderValue::from_str(&t).ok()) {
291            headers.insert(http::header::CONTENT_TYPE, content_type);
292        }
293        match response.trailers.take() {
294            // Trailers on a buffered body: a one-chunk channel carries them.
295            Some(trailers) if !trailers.get().is_empty() => {
296                let (tx, rx) = tokio::sync::mpsc::channel(1);
297                let _ = tx.try_send(Ok(Bytes::from(bytes)));
298                Body { inner: Inner::Channel { rx, trailers: Some(trailers) } }
299            }
300            _ => Body::full(bytes),
301        }
302    };
303    let mut out = http::Response::new(body);
304    *out.status_mut() = http::StatusCode::from_u16(response.status_code as u16).unwrap_or(http::StatusCode::INTERNAL_SERVER_ERROR);
305    *out.headers_mut() = headers;
306    out
307}
308
309/// An `http::Response` as an rws response. A body that's gRPC, or of unknown
310/// or large size, streams through `stream_pipe` (polled on `rws-tower`),
311/// its trailers in `Response::trailers`; small ones are collected.
312pub async fn from_http_response<B>(response: http::Response<B>) -> Result<Response, BoxError>
313where
314    B: http_body::Body<Data = Bytes> + Send + 'static,
315    B::Error: Into<BoxError>,
316{
317    let (parts, body) = response.into_parts();
318    let mut response = Response::new();
319    response.status_code = parts.status.as_u16() as i16;
320    response.reason_phrase = parts.status.canonical_reason().unwrap_or("").to_string();
321    let content_type = parts.headers.get(http::header::CONTENT_TYPE).and_then(|v| v.to_str().ok()).map(str::to_string);
322    let grpc = content_type.as_deref().is_some_and(|t| t.starts_with("application/grpc"));
323    response.headers = headers_from(&parts.headers)
324        .into_iter()
325        .filter(|h| !NOT_COPIED.iter().any(|n| h.name.eq_ignore_ascii_case(n)) && !h.name.eq_ignore_ascii_case("content-type"))
326        .collect();
327    let content_type = content_type.unwrap_or_else(|| "application/octet-stream".to_string());
328
329    // A gRPC trailers-only answer, a 204, a HEAD: headers only.
330    if body.is_end_stream() {
331        if grpc {
332            response.headers.push(Header::new("content-type".to_string(), content_type));
333        }
334        return Ok(response);
335    }
336    let exact = body.size_hint().exact();
337    if !grpc && exact.is_some_and(|n| n <= BUFFER_LIMIT) {
338        let collected = body.collect().await.map_err(Into::into)?;
339        if let Some(trailers) = collected.trailers() {
340            response.trailers = Some(Trailers::with(headers_from(trailers)));
341        }
342        let bytes = collected.to_bytes();
343        if !bytes.is_empty() {
344            response.content_range_list = vec![Range::get_content_range(bytes.to_vec(), content_type)];
345        }
346        return Ok(response);
347    }
348
349    // Streamed. The content type goes in the headers: a stream has no
350    // `ContentRange` to carry it.
351    response.headers.push(Header::new("content-type".to_string(), content_type));
352    let trailers = Trailers::new();
353    response.trailers = Some(trailers.clone());
354    let (tx, rx) = tokio::sync::mpsc::channel(4);
355    response.stream_pipe = Some(Box::new(crate::server::body_stream::ChannelReader::new(rx)));
356    runtime().spawn(async move {
357        let mut body = std::pin::pin!(body);
358        loop {
359            // The body's own error type needn't be `Send`; don't hold it across an await.
360            let next: Option<Result<Frame<Bytes>, BoxError>> = body.frame().await.map(|r| r.map_err(Into::into));
361            match next {
362                Some(Ok(frame)) => {
363                    let frame = match frame.into_data() {
364                        Ok(data) => {
365                            if tx.send(Ok(data.to_vec())).await.is_err() {
366                                return; // the client went away
367                            }
368                            continue;
369                        }
370                        Err(frame) => frame,
371                    };
372                    if let Ok(fields) = frame.into_trailers() {
373                        // Before the channel closes: the writer reads them at the end.
374                        trailers.set(headers_from(&fields));
375                    }
376                }
377                Some(Err(e)) => {
378                    let _ = tx.send(Err(std::io::Error::other(e))).await;
379                    return;
380                }
381                None => return,
382            }
383        }
384    });
385    Ok(response)
386}
387
388/// `poll_ready` then `call`.
389async fn call<S, Req>(mut service: S, request: Req) -> Result<S::Response, BoxError>
390where
391    S: tower_service::Service<Req>,
392    S::Error: Into<BoxError>,
393{
394    std::future::poll_fn(|cx| service.poll_ready(cx)).await.map_err(Into::into)?;
395    service.call(request).await.map_err(Into::into)
396}
397
398// ── TowerApp ──────────────────────────────────────────────────────────────────
399
400/// A Tower service as an rws [`Application`]: every request goes to it.
401/// [`mount`] puts one under a path prefix in front of another app.
402///
403/// The request body reaches the service as it arrives (a gRPC client or
404/// bidirectional stream works); the response streams back with its trailers
405/// (`grpc-status`). The service runs on the `rws-tower` runtime.
406///
407/// ```rust,ignore
408/// # extern crate rws_tower as rust_web_server;
409/// // A tonic service, e.g. tonic-health's:
410/// let (_, health) = tonic_health::server::health_reporter();
411/// let app = rust_web_server::tower::mount("/grpc.health.v1.Health/", health, App::new());
412/// ```
413#[derive(Clone)]
414pub struct TowerApp<S> {
415    service: S,
416}
417
418impl<S> TowerApp<S> {
419    pub fn new(service: S) -> TowerApp<S> {
420        TowerApp { service }
421    }
422}
423
424/// Requests whose path starts with `prefix` go to `service`, the rest to
425/// `fallback`.
426pub fn mount<S, A>(prefix: &str, service: S, fallback: A) -> Mount<S, A> {
427    Mount { prefix: prefix.to_string(), service: TowerApp::new(service), fallback }
428}
429
430/// What [`mount`] returns.
431#[derive(Clone)]
432pub struct Mount<S, A> {
433    prefix: String,
434    service: TowerApp<S>,
435    fallback: A,
436}
437
438impl<S, A> Mount<S, A> {
439    fn routed(&self, request: &Request) -> bool {
440        crate::url::URL::request_target_path(&request.request_uri).unwrap_or("").starts_with(&self.prefix)
441    }
442}
443
444impl<S, B> TowerApp<S>
445where
446    S: tower_service::Service<http::Request<Body>, Response = http::Response<B>> + Clone + Send + Sync + 'static,
447    S::Future: Send + 'static,
448    S::Error: Into<BoxError>,
449    B: http_body::Body<Data = Bytes> + Send + 'static,
450    B::Error: Into<BoxError>,
451{
452    fn respond(&self, request: Request) -> impl Future<Output = Result<Response, String>> + Send + 'static {
453        let service = self.service.clone();
454        async move {
455            let response = call(service, to_http_request(request)).await.map_err(|e| format!("tower service: {e}"))?;
456            from_http_response(response).await.map_err(|e| format!("tower service body: {e}"))
457        }
458    }
459}
460
461impl<S, B> Application for TowerApp<S>
462where
463    S: tower_service::Service<http::Request<Body>, Response = http::Response<B>> + Clone + Send + Sync + 'static,
464    S::Future: Send + 'static,
465    S::Error: Into<BoxError>,
466    B: http_body::Body<Data = Bytes> + Send + 'static,
467    B::Error: Into<BoxError>,
468{
469    fn execute(&self, request: &Request, _connection: &ConnectionInfo) -> Result<Response, String> {
470        let (tx, rx) = std::sync::mpsc::sync_channel(1);
471        let work = self.respond(request.clone());
472        runtime().spawn(async move {
473            let _ = tx.send(work.await);
474        });
475        rx.recv().map_err(|_| "tower service: dropped".to_string())?
476    }
477
478    fn execute_async<'a>(&'a self, request: &'a Request, _connection: &'a ConnectionInfo) -> Option<AsyncResponse<'a>> {
479        let work = runtime().spawn(self.respond(request.clone()));
480        Some(Box::pin(async move { work.await.map_err(|e| format!("tower service: {e}"))? }))
481    }
482
483    fn streams_body(&self, _request: &Request, _connection: &ConnectionInfo) -> bool {
484        true
485    }
486}
487
488impl<S, B, A> Application for Mount<S, A>
489where
490    S: tower_service::Service<http::Request<Body>, Response = http::Response<B>> + Clone + Send + Sync + 'static,
491    S::Future: Send + 'static,
492    S::Error: Into<BoxError>,
493    B: http_body::Body<Data = Bytes> + Send + 'static,
494    B::Error: Into<BoxError>,
495    A: Application,
496{
497    fn execute(&self, request: &Request, connection: &ConnectionInfo) -> Result<Response, String> {
498        if self.routed(request) {
499            self.service.execute(request, connection)
500        } else {
501            self.fallback.execute(request, connection)
502        }
503    }
504
505    fn runs_inline(&self, request: &Request, connection: &ConnectionInfo) -> bool {
506        !self.routed(request) && self.fallback.runs_inline(request, connection)
507    }
508
509    fn execute_async<'a>(&'a self, request: &'a Request, connection: &'a ConnectionInfo) -> Option<AsyncResponse<'a>> {
510        if self.routed(request) {
511            self.service.execute_async(request, connection)
512        } else {
513            self.fallback.execute_async(request, connection)
514        }
515    }
516
517    fn streams_body(&self, request: &Request, connection: &ConnectionInfo) -> bool {
518        self.routed(request) || self.fallback.streams_body(request, connection)
519    }
520}
521
522// ── blocking on a borrowed future ─────────────────────────────────────────────
523
524/// Drive `future` to completion on this thread (it may borrow from the
525/// caller, so it can't be spawned), inside the `rws-tower` runtime's context
526/// so timers and spawns inside it have a driver.
527#[doc(hidden)]
528pub fn block_on<F: Future>(future: F) -> F::Output {
529    struct Unpark(std::thread::Thread);
530    impl std::task::Wake for Unpark {
531        fn wake(self: std::sync::Arc<Self>) {
532            self.0.unpark();
533        }
534        fn wake_by_ref(self: &std::sync::Arc<Self>) {
535            self.0.unpark();
536        }
537    }
538    let _enter = runtime().enter();
539    let waker = std::task::Waker::from(std::sync::Arc::new(Unpark(std::thread::current())));
540    let mut cx = Context::from_waker(&waker);
541    let mut future = std::pin::pin!(future);
542    loop {
543        match future.as_mut().poll(&mut cx) {
544            Poll::Ready(output) => return output,
545            Poll::Pending => std::thread::park(),
546        }
547    }
548}
rws_tower/tower/mod.rs

rws_tower/tower/
mod.rs
