loona 0.3.4

HTTP 1 and 2 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
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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168

use buffet::Piece;
use http::{StatusCode, Version};
use tokio::sync::mpsc;
use tracing::debug;

use super::types::{H2Event, H2EventPayload};
use crate::{Encoder, Response};
use loona_h2::StreamId;

#[derive(Debug, PartialEq, Eq, Clone, Copy)]
#[non_exhaustive]
pub enum EncoderState {
    ExpectResponseHeaders,
    ExpectResponseBody,
    ResponseDone,
}

/// Encodes HTTP/2 responses and bodies
pub struct H2Encoder {
    stream_id: StreamId,
    tx: mpsc::Sender<H2Event>,
    state: EncoderState,
}

impl H2Encoder {
    pub(crate) fn new(stream_id: StreamId, tx: mpsc::Sender<H2Event>) -> Self {
        Self {
            stream_id,
            tx,
            state: EncoderState::ExpectResponseHeaders,
        }
    }

    fn event(&self, payload: H2EventPayload) -> H2Event {
        H2Event {
            payload,
            stream_id: self.stream_id,
        }
    }

    async fn send(&self, payload: H2EventPayload) -> Result<(), H2EncoderError> {
        self.tx
            .send(self.event(payload))
            .await
            .map_err(|_| H2EncoderError::StreamReset)?;
        Ok(())
    }
}

#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum H2EncoderError {
    /// The encoder is in the wrong state
    #[error("Wrong state: expected {expected:?}, actual {actual:?}")]
    WrongState {
        expected: EncoderState,
        actual: EncoderState,
    },

    #[error("Stream reset")]
    StreamReset,
}

impl AsRef<dyn std::error::Error> for H2EncoderError {
    fn as_ref(&self) -> &(dyn std::error::Error + 'static) {
        self
    }
}

impl Encoder for H2Encoder {
    type Error = H2EncoderError;

    async fn write_response(&mut self, res: Response) -> Result<(), Self::Error> {
        // FIXME: HTTP/2 _does_ support informational responses, cf. https://github.com/bearcove/loona/issues/190
        assert!(
            !res.status.is_informational(),
            "http/2 does not support informational responses"
        );

        if self.state != EncoderState::ExpectResponseHeaders {
            return Err(H2EncoderError::WrongState {
                expected: EncoderState::ExpectResponseHeaders,
                actual: self.state,
            });
        }

        self.send(H2EventPayload::Headers(res)).await?;
        self.state = EncoderState::ExpectResponseBody;

        Ok(())
    }

    // TODO: BodyWriteMode is not relevant for h2
    async fn write_body_chunk(&mut self, chunk: Piece) -> Result<(), Self::Error> {
        if self.state != EncoderState::ExpectResponseBody {
            return Err(H2EncoderError::WrongState {
                expected: EncoderState::ExpectResponseBody,
                actual: self.state,
            });
        }

        self.send(H2EventPayload::BodyChunk(chunk)).await?;
        Ok(())
    }

    // TODO: BodyWriteMode is not relevant for h2
    async fn write_body_end(&mut self) -> Result<(), Self::Error> {
        if self.state != EncoderState::ExpectResponseBody {
            return Err(H2EncoderError::WrongState {
                expected: EncoderState::ExpectResponseBody,
                actual: self.state,
            });
        }

        self.send(H2EventPayload::BodyEnd).await?;
        self.state = EncoderState::ResponseDone;

        Ok(())
    }

    // TODO: handle trailers
    async fn write_trailers(&mut self, _trailers: Box<crate::Headers>) -> Result<(), Self::Error> {
        if self.state != EncoderState::ResponseDone {
            return Err(H2EncoderError::WrongState {
                expected: EncoderState::ResponseDone,
                actual: self.state,
            });
        }

        todo!("write trailers")
    }
}

impl Drop for H2Encoder {
    fn drop(&mut self) {
        let mut evs = vec![];

        match self.state {
            EncoderState::ExpectResponseHeaders => {
                evs.push(self.event(H2EventPayload::Headers(Response {
                    version: Version::HTTP_11,
                    status: StatusCode::INTERNAL_SERVER_ERROR,
                    headers: Default::default(),
                })));
                evs.push(self.event(H2EventPayload::BodyEnd));
            }
            EncoderState::ExpectResponseBody => {
                // TODO: this should probably be RST_STREAM instead
                evs.push(self.event(H2EventPayload::BodyEnd));
            }
            EncoderState::ResponseDone => {
                // ah, good.
            }
        }

        if !evs.is_empty() {
            let tx = self.tx.clone();
            buffet::spawn(async move {
                for ev in evs {
                    if tx.send(ev).await.is_err() {
                        debug!("could not send event to h2 connection handler");
                        break;
                    }
                }
            });
        }
    }
}