camber 0.1.5

Opinionated async Rust for IO-bound services on top of Tokio
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

use super::async_proxy::ProxyRequest;
use super::trie::Handler;
use super::{Request, Response};
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

/// An async middleware function that wraps request handling.
///
/// Receives the request and a `Next` handle. Returns a future
/// that resolves to a response. Can short-circuit or modify the
/// response by returning early without calling `next.call()`.
pub type MiddlewareFn =
    Box<dyn Fn(&Request, Next) -> Pin<Box<dyn Future<Output = Response> + Send>> + Send + Sync>;

/// Terminal handler for the middleware chain.
pub(crate) enum Terminal<'a> {
    Handler(&'a Handler),
    /// Passthrough terminal for middleware gating.
    /// Sets the flag to `true` when reached, indicating middleware did not short-circuit.
    Gate(Arc<AtomicBool>),
    /// Proxy terminal — forwards the request to a backend without boxing a closure.
    Proxy {
        backend: Arc<str>,
        prefix: Arc<str>,
    },
}

/// Handle to the next layer in the middleware chain.
///
/// Calling `next.call(req)` returns a future that resolves to the
/// response from the remaining middleware and terminal handler.
pub struct Next<'a> {
    remaining: &'a [MiddlewareFn],
    terminal: Terminal<'a>,
}

impl<'a> Next<'a> {
    pub(crate) fn new(remaining: &'a [MiddlewareFn], terminal: Terminal<'a>) -> Self {
        Self {
            remaining,
            terminal,
        }
    }

    /// Run the next middleware layer or terminal handler.
    pub fn call(self, req: &Request) -> Pin<Box<dyn Future<Output = Response> + Send>> {
        match (self.remaining.split_first(), &self.terminal) {
            (Some((mw, rest)), _) => {
                let next = Next {
                    remaining: rest,
                    terminal: self.terminal,
                };
                mw(req, next)
            }
            (None, Terminal::Handler(handler)) => handler(req),
            (None, Terminal::Gate(flag)) => {
                flag.store(true, Ordering::Release);
                Box::pin(async { Response::empty_raw(200) })
            }
            (None, Terminal::Proxy { backend, prefix }) => {
                let proxy_req = ProxyRequest::from_request(req);
                let backend = Arc::clone(backend);
                let prefix = Arc::clone(prefix);
                Box::pin(forward_proxy(proxy_req, backend, prefix))
            }
        }
    }
}

/// Forward a proxy request to the backend, mapping errors to 502 responses.
async fn forward_proxy(proxy_req: ProxyRequest, backend: Arc<str>, prefix: Arc<str>) -> Response {
    match super::async_proxy::forward_request_buffered(proxy_req, &backend, &prefix).await {
        Ok(resp) => resp,
        Err(e) => {
            tracing::warn!(error = %e, "proxy upstream failed");
            Response::text_raw(502, "proxy upstream failed")
        }
    }
}