rama 0.3.0

modular service framework
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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154

//! An example to showcase how one can build a service using Rama,
//! built with layers to modify and branch the traffic as it goes through the service.
//! And this on Layer 4 (TCP) all the way to Layer 7 (HTTP).
//!
//! # Run the example
//!
//! ```sh
//! cargo run --example http_service_hello --features=compression,http-full
//! ```
//!
//! # Expected output
//!
//! The server will start and listen on `:62010`. You can use `curl` to interact with the service:
//!
//! ```sh
//! curl -v http://127.0.0.1:62010
//! ```
//!
//! You should see a response with `HTTP/1.1 200 OK` and a HTML body containing
//! the peer address, the path of the request and the stats of the bytes read and written.

#![expect(
    clippy::unwrap_used,
    clippy::expect_used,
    reason = "example/test/bench: panic-on-error and print-for-output are the standard patterns for demos and harnesses"
)]

use rama::{
    Layer,
    bytes::Bytes,
    extensions::ExtensionsRef,
    http::{
        Request, header,
        layer::{
            compression::CompressionLayer,
            error_handling::ErrorHandlerLayer,
            sensitive_headers::{
                SetSensitiveRequestHeadersLayer, SetSensitiveResponseHeadersLayer,
            },
            trace::{DefaultMakeSpan, DefaultOnResponse, TraceLayer},
        },
        server::HttpServer,
        service::web::{IntoEndpointService, response::Html},
    },
    layer::ArcLayer,
    layer::{TimeoutLayer, TraceErrLayer},
    net::stream::{
        SocketInfo,
        layer::{BytesRWTrackerHandle, IncomingBytesTrackerLayer},
    },
    rt::Executor,
    tcp::server::TcpListener,
    telemetry::tracing::{
        self,
        level_filters::LevelFilter,
        subscriber::{EnvFilter, fmt, layer::SubscriberExt, util::SubscriberInitExt},
    },
    utils::latency::LatencyUnit,
};

use std::{sync::Arc, time::Duration};

#[tokio::main]
async fn main() {
    tracing::subscriber::registry()
        .with(fmt::layer())
        .with(
            EnvFilter::builder()
                .with_default_directive(LevelFilter::DEBUG.into())
                .from_env_lossy(),
        )
        .init();

    let graceful = rama::graceful::Shutdown::default();

    let sensitive_headers: Arc<[_]> = vec![header::AUTHORIZATION, header::COOKIE].into();

    graceful.spawn_task_fn(async |guard| {
        let exec = Executor::graceful(guard.clone());

        let http_service = (
            ArcLayer::new(),
            CompressionLayer::new(),
            SetSensitiveRequestHeadersLayer::from_shared(sensitive_headers.clone()),
            TraceLayer::new_for_http()
                .on_body_chunk(|chunk: &Bytes, latency: Duration, _: &tracing::Span| {
                    tracing::trace!(
                        http.request.body.chunk_size = chunk.len(),
                        http.request.body.chunk_read_ms = latency.as_millis(),
                        "sending body chunk"
                    )
                })
                .make_span_with(DefaultMakeSpan::new().with_include_headers(true))
                .on_response(
                    DefaultOnResponse::new()
                        .with_include_headers(true)
                        .with_latency_unit(LatencyUnit::Micros),
                ),
            SetSensitiveResponseHeadersLayer::from_shared(sensitive_headers),
            ErrorHandlerLayer::new(),
        )
            .into_layer(
                (|req: Request| {
                    let uri = req.uri();
                    let socket_info = req.extensions().get_ref::<SocketInfo>().unwrap();
                    let tracker = req.extensions().get_ref::<BytesRWTrackerHandle>().unwrap();

                    std::future::ready(Html(format!(
                        r##"
                        <html>
                            <head>
                                <title>Rama — Http Service Hello</title>
                            </head>
                            <body>
                                <h1>Hello</h1>
                                <p>Peer: {}</p>
                                <p>Path: {}</p>
                                <p>Stats (bytes):</p>
                                <ul>
                                    <li>Read: {}</li>
                                    <li>Written: {}</li>
                                </ul>
                            </body>
                        </html>"##,
                        socket_info.peer_addr(),
                        uri.path_or_root(),
                        tracker.read(),
                        tracker.written(),
                    )))
                })
                .into_endpoint_service(),
            );

        let tcp_http_service = HttpServer::auto(exec.clone()).service(http_service);

        TcpListener::bind_address("127.0.0.1:62010", exec)
            .await
            .expect("bind TCP Listener")
            .serve(
                (
                    TraceErrLayer::new(),
                    TimeoutLayer::new(Duration::from_secs(8)),
                    IncomingBytesTrackerLayer::new(),
                )
                    .into_layer(tcp_http_service),
            )
            .await;
    });

    graceful
        .shutdown_with_limit(Duration::from_secs(30))
        .await
        .expect("graceful shutdown");
}