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
//! An example to showcase how to optionally support HaProxy.
//! which is typically used in case your service is behind a loadbalancer.
//!
//! Our server implementation can handle both v1 and v2 alike.
//!
//! # Run the example
//!
//! ```sh
//! cargo run --example haproxy_client_ip --features=haproxy,http-full
//! ```
//!
//! # Expected output
//!
//! The server will start and listen on `:62025`. You can use `curl` to interact with the service:
//!
//! ```sh
//! curl -v http://127.0.0.1:62025
//! ```
//!
//! You should see a response with `HTTP/1.1 200 OK` and the client IP as the body payload.
//! In case you are doing this with HaProxy data at the start of your Tcp stream,
//! you'll see the client IP Address advertised in there, otherwise you'll see
//! the socket peer addr.
#![expect(
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,
error::ErrorContext,
http::{
Request, StatusCode,
layer::{
error_handling::ErrorHandlerLayer, required_header::AddRequiredResponseHeadersLayer,
},
server::HttpServer,
service::web::{Router, response::ErrorResponse},
},
layer::ArcLayer,
net::ClientIp,
proxy::haproxy::server::HaProxyLayer,
rt::Executor,
tcp::server::TcpListener,
telemetry::tracing::{
self,
level_filters::LevelFilter,
subscriber::{EnvFilter, fmt, layer::SubscriberExt, util::SubscriberInitExt},
},
};
use std::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();
graceful.spawn_task_fn(async |guard| {
let exec = Executor::graceful(guard);
let tcp_http_service = HttpServer::auto(exec.clone()).service(
(
ArcLayer::new(),
AddRequiredResponseHeadersLayer::new(),
ErrorHandlerLayer::new(),
)
.into_layer(Router::new().with_get(
"/",
async |req: Request| -> Result<String, ErrorResponse> {
let client_ip = req
.client_ip()
.context("failed to fetch client IP")
.map_err(|err| (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
Ok(client_ip.to_string())
},
)),
);
TcpListener::bind_address("127.0.0.1:62025", exec)
.await
.expect("bind TCP Listener")
.serve(
HaProxyLayer::new()
// by default [`HaProxyLayer`] is enforced,
// setting peek=true allows you to make it optional,
// which is pretty useful to easily run cloud services locally
.with_peek(true)
.into_layer(tcp_http_service),
)
.await;
});
graceful
.shutdown_with_limit(Duration::from_secs(30))
.await
.expect("graceful shutdown");
}