alux_http_rocket/
server.rs

1use crate::RocketRoute;
2use alux_http::{HttpServerAlg, HttpServerSetup};
3use rocket::config::{LogLevel, Shutdown as ShutdownConfig};
4use rocket::fairing::AdHoc;
5#[cfg(unix)]
6use std::collections::HashSet;
7use std::io;
8use tokio::sync::oneshot;
9use tokio::task::JoinHandle;
10
11type RocketServerError = Box<dyn std::error::Error + Send + Sync>;
12
13/// Bounds how long Rocket finishes outstanding requests before it ends them, in seconds.
14///
15/// The same bound the other interpretations state, so a request still being served this long after
16/// a close is ended wherever it was made. Rocket's default is 2, which would end one sooner here
17/// than anywhere else.
18const GRACE: u32 = 5;
19
20/// Bounds how long Rocket then finishes outstanding connection I/O, in seconds.
21///
22/// Nothing, because it is spent after the grace period, by which point every request has been
23/// answered or ended. Rocket takes about a second of its own past these two either way, so closing
24/// returns a second after the grace period rather than at it. Its default is 3.
25const MERCY: u32 = 0;
26
27/// Serves a Rocket route at the address its setup names.
28#[derive(Debug, Default)]
29pub struct RocketServer;
30
31/// Carries an open Rocket server that is accepting requests.
32pub struct RocketOpen {
33    shutdown: rocket::Shutdown,
34    task: JoinHandle<Result<(), RocketServerError>>,
35}
36
37impl Drop for RocketOpen {
38    fn drop(&mut self) {
39        self.shutdown.clone().notify();
40        self.task.abort();
41    }
42}
43
44/// Shuts Rocket down if the open that launched it is cancelled.
45///
46/// This open is the only one that awaits after spawning, so cancelling it would otherwise leave
47/// the address held by a task with no handle.
48struct Launching(Option<rocket::Shutdown>);
49
50impl Launching {
51    /// Disarms the guard once the open has finished.
52    fn launched(&mut self) {
53        self.0 = None;
54    }
55}
56
57impl Drop for Launching {
58    fn drop(&mut self) {
59        if let Some(shutdown) = self.0.take() {
60            shutdown.notify();
61        }
62    }
63}
64
65/// Reads a Rocket failure as text, which is also what keeps it from aborting the process.
66fn rocket_error(error: &rocket::Error) -> RocketServerError {
67    io::Error::other(error.to_string()).into()
68}
69
70impl HttpServerAlg for RocketServer {
71    type Program = RocketRoute;
72    type Open = RocketOpen;
73    type Error = RocketServerError;
74
75    async fn open(&mut self, setup: HttpServerSetup<Self::Program>) -> Result<Self::Open, Self::Error> {
76        let (bind, route) = setup.into_parts();
77        let (listening, mut bound) = oneshot::channel();
78        let rocket = route
79            .mount(rocket::custom(rocket::Config {
80                address: bind.address().ip(),
81                port: bind.address().port(),
82                // Rocket writes a launch banner and a line per request to stdout. Whoever opened
83                // this server states what it says, so it says nothing.
84                log_level: LogLevel::Off,
85                shutdown: ShutdownConfig {
86                    grace: GRACE,
87                    mercy: MERCY,
88                    // Whoever opened this server decides when it closes, so Rocket takes no
89                    // signals. Left on, it installs process-wide handlers and answers Ctrl-C and
90                    // SIGTERM itself, which stops an application from ever seeing them.
91                    ctrlc: false,
92                    #[cfg(unix)]
93                    signals: HashSet::new(),
94                    ..ShutdownConfig::default()
95                },
96                ..rocket::Config::default()
97            }))
98            // Rocket binds inside `launch`, so liftoff is the only reliable signal that the
99            // address is held. Probing it instead would answer for whatever server is there.
100            .attach(AdHoc::on_liftoff("alux-http-rocket bound", move |_| {
101                Box::pin(async move {
102                    let _ = listening.send(());
103                })
104            }))
105            .ignite()
106            .await
107            .map_err(|error| rocket_error(&error))?;
108        let shutdown = rocket.shutdown();
109        let mut launching = Launching(Some(shutdown.clone()));
110        let mut task =
111            tokio::task::spawn_local(async move { rocket.launch().await.map(|_| ()).map_err(|e| rocket_error(&e)) });
112
113        // Rocket either lifts off or stops; stopping first carries the reason it failed.
114        let open = tokio::select! {
115            result = &mut task => match result? {
116                Ok(()) => Err(io::Error::other("Rocket stopped before binding").into()),
117                Err(error) => Err(error),
118            },
119            Ok(()) = &mut bound => Ok(RocketOpen { shutdown, task }),
120        };
121        launching.launched();
122
123        open
124    }
125
126    async fn close(&mut self, open: &mut Self::Open) -> Result<(), Self::Error> {
127        open.shutdown.clone().notify();
128        if !open.task.is_finished() {
129            let _ = (&mut open.task).await;
130        }
131
132        Ok(())
133    }
134
135    // Nothing here can tell the address being released from the drain being over, because both
136    // happen inside the framework's own shutdown. So ending is closing, which satisfies both.
137    async fn end(&mut self, open: &mut Self::Open) -> Result<(), Self::Error> {
138        self.close(open).await
139    }
140}
alux_http_rocket/server.rs

alux_http_rocket/
server.rs
