flyer 2.1.3

HTTP framework for rust
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

use std::{pin::Pin};
use std::net::SocketAddr;

use anyhow::Result;
use h2::server;
use tokio::{io::{AsyncBufReadExt, AsyncRead, AsyncWrite, BufReader}, net::TcpListener};
use tokio_rustls::TlsAcceptor;

use crate::server::transport::handler::ws;
use crate::{request::Request, server::transport::handler::http2};
use crate::{response::Response, server::transport::handler::http2::H2_PREFACE};
use crate::{server::{Server, transport::{Protocol, handler::http1}}, utils::server::get_tls_acceptor, warn};

pub async fn listen(ptr: usize) {
    let socket = TcpListener::bind(Server::instance(ptr).address()).await.unwrap();
    let tls= Server::instance(ptr).server_config.clone().map(|config| get_tls_acceptor(config.clone()).unwrap());

    listener(ptr, socket, tls).await;
}

async fn listener(ptr: usize, listener: TcpListener, tls: Option<TlsAcceptor>) {
    unsafe  {
        let tls_ptr = &tls as *const Option<TlsAcceptor> as usize;

        while let Ok((stream, addr)) = listener.accept().await {
            tokio::spawn(async move {
                match &*(tls_ptr as *const Option<TlsAcceptor>) {
                    Some(acceptor) => {
                        match acceptor.accept(stream).await {
                            Ok(rw) => connection(ptr, BufReader::new(rw), addr).await,
                            Err(err) => warn!("TLS connection error"; "error" => err),
                        }
                    },
                    None => { connection(ptr, BufReader::new(stream), addr).await },
                }
            });
        }
    }
}

async fn connection_protocol<RW>(rw: &mut BufReader<RW>) -> Result<Protocol>
where
    RW: AsyncRead + AsyncWrite + Unpin  + Sync + Send + 'static,
{
    let buffer = rw.fill_buf().await.unwrap();

    Ok(
        match buffer.len() >= H2_PREFACE.len() && &buffer[..H2_PREFACE.len()] == H2_PREFACE {
            true => Protocol::HTTP2,
            false => Protocol::HTTP1,
        },
    )
}

async fn connection<RW>(ptr: usize, mut rw: BufReader<RW>, addr: SocketAddr)
where
    RW: AsyncRead + AsyncWrite + Unpin  + Sync + Send + 'static,
{
    match connection_protocol(&mut rw).await {
        Ok(protocol) => {
            let result = match protocol {
                Protocol::HTTP1 => { http_1_protocol(ptr, rw, addr).await },
                Protocol::HTTP2 => { http_2_protocol(ptr, rw, addr).await },
                Protocol::HTTP3 => { Err(anyhow::anyhow!("Unsupported request tcp connection")) },
            };

            if let Err(err) = result {
                warn!("failed to process request"; "error" => err);
            }
        }
        Err(err) => {
            warn!("request protocol error"; "error" => err);
        }
    };
}

async fn ws_protocol<RW>(ptr: usize, rw: BufReader<RW>, req: &mut Request, res: &mut Response) -> Result<()>
where
    RW: AsyncRead + AsyncWrite + Unpin  + Sync + Send + 'static,
{
    return Ok(
        match Server::instance(ptr).on_ws_request(req, res).await {
            Some((route, req, res)) => {
                ws::Handler::new(rw, req, res).await.unwrap().handle(route).await.unwrap();
            },
            None => { drop(rw); },
        }
    );
}

async fn http_1_protocol<RW>(ptr: usize, mut rw: BufReader<RW>, addr: SocketAddr) -> Result<()>
where
    RW: AsyncRead + AsyncWrite + Unpin + Sync + Send + 'static,
{
    unsafe {
        let mut handler = http1::Handler::new(ptr, Pin::new(&mut rw), addr);
        let handle_result = handler.handle().await;

        if let Err(err) = handle_result {
            return Err(err);
        }

        let (mut req, mut res) = (handle_result.unwrap(), Response::new());

        if req.header("upgrade").to_lowercase() == "websocket" {
            return ws_protocol(ptr, rw, &mut req, &mut res).await;
        }

        Server::instance(ptr).on_web_request(&mut req, &mut res).await.unwrap();
        handler.write(&mut req, &mut res).await.unwrap();

        drop(rw);

        return Ok(());
    }
}

async fn http_2_protocol<RW>(ptr: usize, rw: BufReader<RW>, addr: SocketAddr) -> Result<()>
where
    RW: AsyncRead + AsyncWrite + Unpin  + Sync + Send + 'static,
{
    match server::handshake(rw).await {
        Ok(mut conn) => {
                while let Some(result) = conn.accept().await {
                match result {
                    Ok((request, send)) => {
                        tokio::spawn(async move {
                            let mut handler = http2::Handler::new(addr, send);
                            let (mut req, mut res) = (handler.handle(request).await.unwrap(), Response::new());

                            Server::instance(ptr).on_web_request(&mut req, &mut res).await.unwrap();

                            handler.write(&mut req, &mut res).await.unwrap();
                        });
                    },
                    Err(_) => {},
                }
            }
            return Ok(());
        },
        Err(err) => { return Err(err.into()); },
    }
}