fluke 0.1.1

An HTTP implementation on top of io_uring
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

use std::rc::Rc;

use eyre::Context;
use tracing::debug;

use crate::{
    h1::body::{H1Body, H1BodyKind},
    util::{read_and_parse, SemanticError},
    ExpectResponseHeaders, HeadersExt, Responder, ServerDriver,
};
use fluke_buffet::{ReadOwned, RollMut, WriteOwned};

use super::encode::H1Encoder;

pub struct ServerConf {
    /// Max length of the request line + HTTP headers
    pub max_http_header_len: usize,

    /// Max length of a single header record, e.g. `user-agent: foobar`
    pub max_header_record_len: usize,

    /// Max number of header records
    pub max_header_records: usize,
}

impl Default for ServerConf {
    fn default() -> Self {
        Self {
            max_http_header_len: 64 * 1024,
            max_header_record_len: 4 * 1024,
            max_header_records: 128,
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ServeOutcome {
    ClientRequestedConnectionClose,
    ServerRequestedConnectionClose,
    ClientClosedConnectionBetweenRequests,
    // TODO: return buffer there so we can see what they did write?
    ClientDidntSpeakHttp11,
}

pub async fn serve(
    (mut transport_r, mut transport_w): (impl ReadOwned, impl WriteOwned),
    conf: Rc<ServerConf>,
    mut client_buf: RollMut,
    driver: impl ServerDriver,
) -> eyre::Result<ServeOutcome> {
    loop {
        let req;
        (client_buf, req) = match read_and_parse(
            super::parse::request,
            &mut transport_r,
            client_buf,
            conf.max_http_header_len,
        )
        .await
        {
            Ok(t) => match t {
                Some(t) => t,
                None => {
                    debug!("client went away before sending request headers");
                    return Ok(ServeOutcome::ClientClosedConnectionBetweenRequests);
                }
            },
            Err(e) => {
                if let Some(se) = e.downcast_ref::<SemanticError>() {
                    transport_w
                        .write_all_owned(se.as_http_response())
                        .await
                        .wrap_err("writing error response downstream")?;
                }

                debug!(?e, "error reading request header from downstream");
                return Ok(ServeOutcome::ClientDidntSpeakHttp11);
            }
        };
        debug!("got request {req:?}");

        let chunked = req.headers.is_chunked_transfer_encoding();
        let connection_close = req.headers.is_connection_close();
        let content_len = req.headers.content_length().unwrap_or_default();

        let mut req_body = H1Body::new(
            transport_r,
            client_buf,
            if chunked {
                H1BodyKind::Chunked
            } else {
                H1BodyKind::ContentLength(content_len)
            },
        );

        let responder = Responder {
            encoder: H1Encoder { transport_w },
            state: ExpectResponseHeaders,
        };

        let resp = driver
            .handle(req, &mut req_body, responder)
            .await
            .wrap_err("handling request")?;

        // TODO: if we sent `connection: close` we should close now
        transport_w = resp.into_inner().transport_w;

        (client_buf, transport_r) = req_body
            .into_inner()
            .ok_or_else(|| eyre::eyre!("request body not drained, have to close connection"))?;

        if connection_close {
            debug!("client requested connection close");
            return Ok(ServeOutcome::ClientRequestedConnectionClose);
        }
    }
}