thruster 1.3.13

A middleware based http async web server.
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

use std::error::Error;
use std::net::ToSocketAddrs;
use std::sync::Arc;

use async_trait::async_trait;
use futures::sink::SinkExt;
use futures::stream::StreamExt;
use futures::FutureExt;
use native_tls::Identity;
use tokio::net::{TcpListener, TcpStream};
use tokio_stream::wrappers::TcpListenerStream;
use tokio_util::codec::Framed;
use tokio_util::sync::ReusableBoxFuture;

use crate::app::App;
use crate::core::context::Context;
use crate::core::http::Http;
use crate::core::request::Request;
use crate::core::response::Response;

use crate::server::ThrusterServer;

pub struct SSLServer<T: 'static + Context<Response = Response> + Clone + Send + Sync, S: Send> {
    app: App<Request, T, S>,
    cert: Option<Vec<u8>>,
    cert_pass: &'static str,
}

impl<T: 'static + Context<Response = Response> + Clone + Send + Sync, S: Send> SSLServer<T, S> {
    ///
    /// Sets the cert on the server
    ///
    pub fn cert(&mut self, cert: Vec<u8>) {
        self.cert = Some(cert);
    }

    pub fn cert_pass(&mut self, cert_pass: &'static str) {
        self.cert_pass = cert_pass;
    }
}

#[async_trait]
impl<T: Context<Response = Response> + Clone + Send + Sync, S: 'static + Send + Sync> ThrusterServer
    for SSLServer<T, S>
{
    type Context = T;
    type Response = Response;
    type Request = Request;
    type State = S;

    fn new(mut app: App<Self::Request, T, Self::State>) -> Self {
        app = app.commit();

        SSLServer {
            app,
            cert: None,
            cert_pass: "",
        }
    }

    ///
    /// Alias for start_work_stealing_optimized
    ///
    fn build(self, host: &str, port: u16) -> ReusableBoxFuture<()> {
        if self.cert.is_none() {
            panic!("Cert is required to be set via SSLServer::cert() before starting the server");
        }

        let addr = (host, port).to_socket_addrs().unwrap().next().unwrap();

        let cert = self.cert.unwrap();
        let cert_pass = self.cert_pass;
        let cert = Identity::from_pkcs12(&cert, cert_pass).expect("Could not decrypt p12 file");
        let tls_acceptor = tokio_native_tls::TlsAcceptor::from(
            native_tls::TlsAcceptor::builder(cert)
                .build()
                .expect("Could not create TLS acceptor."),
        );
        let arc_app = Arc::new(self.app);
        let arc_acceptor = Arc::new(tls_acceptor);

        let listener_fut = TcpListener::bind(addr).then(move |listener| {
            TcpListenerStream::new(listener.unwrap()).for_each(move |res| {
                if let Ok(stream) = res {
                    let cloned_app = arc_app.clone();
                    let cloned_tls_acceptor = arc_acceptor.clone();
                    tokio::spawn(async move {
                        if let Err(e) = process(cloned_app, cloned_tls_acceptor, stream).await {
                            println!("failed to process connection; error = {}", e);
                        }
                    });
                }

                async {}
            })
        });

        ReusableBoxFuture::new(listener_fut)
    }
}

async fn process<T: Context<Response = Response> + Clone + Send + Sync, S: 'static + Send>(
    app: Arc<App<Request, T, S>>,
    tls_acceptor: Arc<tokio_native_tls::TlsAcceptor>,
    socket: TcpStream,
) -> Result<(), Box<dyn Error>> {
    let tls = tls_acceptor.accept(socket).await?;
    let mut framed = Framed::new(tls, Http);

    while let Some(request) = framed.next().await {
        match request {
            Ok(request) => {
                let matched =
                    app.resolve_from_method_and_path(request.method(), request.path().to_owned());
                let response = app.resolve(request, matched).await?;
                framed.send(response).await?;
            }
            Err(e) => return Err(e.into()),
        }
    }

    Ok(())
}