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

use std::rc::Rc;

use fluke::{Body, Encoder, ExpectResponseHeaders, Responder, Response, ResponseDone};
use fluke_buffet::{IntoHalves, PipeRead, PipeWrite, ReadOwned, RollMut, WriteOwned};
use http::StatusCode;
use tracing::Level;
use tracing_subscriber::{filter::Targets, layer::SubscriberExt, util::SubscriberInitExt};

/// Note: this will not work with `cargo test`, since it sets up process-level
/// globals. But it will work with `cargo nextest`, and that's what fluke is standardizing on.
pub(crate) fn setup_tracing_and_error_reporting() {
    color_eyre::install().unwrap();

    let targets = if let Ok(rust_log) = std::env::var("RUST_LOG") {
        rust_log.parse::<Targets>().unwrap()
    } else {
        Targets::new()
            .with_default(Level::INFO)
            .with_target("fluke", Level::DEBUG)
            .with_target("httpwg", Level::DEBUG)
            .with_target("want", Level::INFO)
    };

    let fmt_layer = tracing_subscriber::fmt::layer()
        .with_ansi(true)
        .with_file(false)
        .with_line_number(false)
        .without_time();

    tracing_subscriber::registry()
        .with(targets)
        .with(fmt_layer)
        .init();
}

struct TestDriver;

impl fluke::ServerDriver for TestDriver {
    async fn handle<E: Encoder>(
        &self,
        _req: fluke::Request,
        _req_body: &mut impl Body,
        mut res: Responder<E, ExpectResponseHeaders>,
    ) -> eyre::Result<Responder<E, ResponseDone>> {
        let mut buf = RollMut::alloc()?;

        buf.put(b"Continue")?;

        res.write_interim_response(Response {
            status: StatusCode::CONTINUE,
            ..Default::default()
        })
        .await?;

        buf.put(b"OK")?;

        _ = buf;

        let res = res
            .write_final_response(Response {
                status: StatusCode::OK,
                ..Default::default()
            })
            .await?;

        let res = res.finish_body(None).await?;

        Ok(res)
    }
}

pub struct TwoHalves<W, R>(W, R);
impl<W: WriteOwned, R: ReadOwned> IntoHalves for TwoHalves<W, R> {
    type Read = R;
    type Write = W;

    fn into_halves(self) -> (Self::Read, Self::Write) {
        (self.1, self.0)
    }
}

pub fn start_server() -> httpwg::Conn<TwoHalves<PipeWrite, PipeRead>> {
    let (server_write, client_read) = fluke::buffet::pipe();
    let (client_write, server_read) = fluke::buffet::pipe();

    let serve_fut = async move {
        let server_conf = Rc::new(fluke::h2::ServerConf {
            ..Default::default()
        });

        let client_buf = RollMut::alloc()?;
        let driver = Rc::new(TestDriver);
        let io = (server_read, server_write);
        fluke::h2::serve(io, server_conf, client_buf, driver).await?;
        tracing::debug!("http/2 server done");
        Ok::<_, eyre::Report>(())
    };

    fluke_buffet::spawn(async move {
        serve_fut.await.unwrap();
    });

    let config = Rc::new(httpwg::Config::default());
    httpwg::Conn::new(config, TwoHalves(client_write, client_read))
}

#[cfg(test)]
httpwg_macros::tests! {{
   crate::setup_tracing_and_error_reporting();

   fluke_buffet::start(async move {
       let conn = crate::start_server();
       let result = test(conn).await;
       result.unwrap()
   });
}}