rustio-core 2.0.0

Runtime core for RustIO: HTTP server, router, middleware, ORM, admin, and migrations.
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

//! Hyper-backed HTTP/1 server.
//!
//! Bind an address with [`Server::bind`] and serve either a raw handler
//! via [`Server::serve`] or a [`Router`] via [`Server::serve_router`].

use std::convert::Infallible;
use std::future::Future;
use std::net::SocketAddr;
use std::sync::Arc;

use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper_util::rt::TokioIo;
use tokio::net::TcpListener;

use crate::http::{Request, Response};
use crate::router::Router;

pub struct Server {
    addr: SocketAddr,
}

impl Server {
    pub fn bind(addr: SocketAddr) -> Self {
        Self { addr }
    }

    pub async fn serve<F, Fut>(self, handler: F) -> std::io::Result<()>
    where
        F: Fn(Request) -> Fut + Clone + Send + Sync + 'static,
        Fut: Future<Output = Response> + Send + 'static,
    {
        let listener = TcpListener::bind(self.addr).await?;
        eprintln!("rustio-core: listening on http://{}", self.addr);

        loop {
            let (stream, peer) = listener.accept().await?;
            let io = TokioIo::new(stream);
            let handler = handler.clone();

            tokio::spawn(async move {
                let service = service_fn(move |raw: hyper::Request<hyper::body::Incoming>| {
                    let handler = handler.clone();
                    async move {
                        // `peer` is captured by `Copy` (SocketAddr is
                        // Copy); attached to every request off this
                        // connection so handlers can read it via
                        // `Request::peer_addr`.
                        let req = Request::new(raw, Some(peer));
                        Ok::<Response, Infallible>(handler(req).await)
                    }
                });

                if let Err(err) = http1::Builder::new().serve_connection(io, service).await {
                    eprintln!("rustio-core: connection error: {err}");
                }
            });
        }
    }

    pub async fn serve_router(self, router: Router) -> std::io::Result<()> {
        let router = Arc::new(router);
        self.serve(move |req| {
            let router = router.clone();
            async move { router.dispatch(req).await }
        })
        .await
    }

    /// Serve a router on an already-bound `TcpListener`.
    ///
    /// Use when the caller needs to own the socket — for example to
    /// bind to port 0 and read back the kernel-assigned address
    /// before spawning the server (integration tests, pre-fork
    /// servers that drop privileges after binding).
    pub async fn serve_router_on(listener: TcpListener, router: Router) -> std::io::Result<()> {
        let router = Arc::new(router);
        loop {
            let (stream, peer) = listener.accept().await?;
            let io = TokioIo::new(stream);
            let router = router.clone();

            tokio::spawn(async move {
                let service = service_fn(move |raw: hyper::Request<hyper::body::Incoming>| {
                    let router = router.clone();
                    async move {
                        let req = Request::new(raw, Some(peer));
                        Ok::<Response, Infallible>(router.dispatch(req).await)
                    }
                });
                if let Err(err) = http1::Builder::new().serve_connection(io, service).await {
                    eprintln!("rustio-core: connection error: {err}");
                }
            });
        }
    }
}