http3_datagram/
datagram.rs

1use bytes::Buf;
2use http3::{
3    error::{Code, internal_error::InternalConnectionError},
4    proto::varint::VarInt,
5    quic::StreamId,
6};
7
8/// HTTP datagram frames
9/// See: <https://www.rfc-editor.org/rfc/rfc9297#section-2.1>
10#[derive(Debug, Clone)]
11pub struct Datagram<B> {
12    /// Stream id divided by 4
13    stream_id: StreamId,
14    /// The data contained in the datagram
15    payload: B,
16}
17
18impl<B> Datagram<B>
19where
20    B: Buf,
21{
22    /// Creates a new datagram frame
23    // TODO: remove for MSRV >= 1.87 https://github.com/rust-lang/rust/issues/128101
24    #[allow(unknown_lints, clippy::manual_is_multiple_of)]
25    pub fn new(stream_id: StreamId, payload: B) -> Self {
26        assert!(
27            stream_id.into_inner() % 4 == 0,
28            "StreamId is not divisible by 4"
29        );
30        // StreamId will be divided by 4 when encoding the Datagram header
31        Self { stream_id, payload }
32    }
33
34    /// Decodes a datagram frame from the QUIC datagram
35    pub fn decode(mut buf: B) -> Result<Self, InternalConnectionError> {
36        let q_stream_id = VarInt::decode(&mut buf).map_err(|_| {
37            InternalConnectionError::new(Code::H3_DATAGRAM_ERROR, "invalid stream id".to_string())
38        })?;
39
40        //= https://www.rfc-editor.org/rfc/rfc9297#section-2.1
41        // Quarter Stream ID: A variable-length integer that contains the value of the
42        // client-initiated bidirectional stream that this datagram is associated with
43        // divided by four (the division by four stems from the fact that HTTP requests are
44        // sent on client-initiated bidirectional streams, which have stream IDs that are
45        // divisible by four). The largest legal QUIC stream ID value is 262-1, so the
46        // largest legal value of the Quarter Stream ID field is 260-1. Receipt of an HTTP/3
47        // Datagram that includes a larger value MUST be treated as an HTTP/3 connection
48        // error of type H3_DATAGRAM_ERROR (0x33).
49        let stream_id = StreamId::try_from(u64::from(q_stream_id) * 4).map_err(|_| {
50            InternalConnectionError::new(Code::H3_DATAGRAM_ERROR, "invalid stream id".to_string())
51        })?;
52
53        let payload = buf;
54
55        Ok(Self { stream_id, payload })
56    }
57
58    #[inline]
59    /// Returns the associated stream id of the datagram
60    pub fn stream_id(&self) -> StreamId {
61        self.stream_id
62    }
63
64    #[inline]
65    /// Returns the datagram payload
66    pub fn payload(&self) -> &B {
67        &self.payload
68    }
69
70    /// Encode the datagram to wire format
71    pub fn encode(self) -> EncodedDatagram<B> {
72        let mut buffer = [0; VarInt::MAX_SIZE];
73        let varint = VarInt::from(self.stream_id) / 4;
74        varint.encode(&mut buffer.as_mut_slice());
75        EncodedDatagram {
76            stream_id: buffer,
77            len: varint.size(),
78            pos: 0,
79            payload: self.payload,
80        }
81    }
82
83    /// Returns the datagram payload
84    pub fn into_payload(self) -> B {
85        self.payload
86    }
87}
88
89#[cfg(test)]
90mod tests {
91    use bytes::{Buf, Bytes};
92    use http3::quic::StreamId;
93
94    use super::Datagram;
95
96    #[test]
97    fn encode_preserves_quarter_stream_id() {
98        let stream_id = StreamId::try_from(4).unwrap();
99        let mut encoded = Datagram::new(stream_id, Bytes::from_static(b"payload")).encode();
100
101        let decoded = Datagram::decode(encoded.copy_to_bytes(encoded.remaining())).unwrap();
102
103        assert_eq!(decoded.stream_id(), stream_id);
104        assert_eq!(decoded.into_payload(), Bytes::from_static(b"payload"));
105    }
106}
107
108#[derive(Debug)]
109pub struct EncodedDatagram<B: Buf> {
110    /// Encoded datagram stream ID as Varint
111    stream_id: [u8; VarInt::MAX_SIZE],
112    /// Length of the varint
113    len: usize,
114    /// Position of the stream_id buffer
115    pos: usize,
116    /// Datagram Payload
117    payload: B,
118}
119
120/// Implementation of [`Buf`] for [`Datagram`]
121impl<B> Buf for EncodedDatagram<B>
122where
123    B: Buf,
124{
125    fn remaining(&self) -> usize {
126        self.len - self.pos + self.payload.remaining()
127    }
128
129    fn chunk(&self) -> &[u8] {
130        if self.len - self.pos > 0 {
131            &self.stream_id[self.pos..self.len]
132        } else {
133            self.payload.chunk()
134        }
135    }
136
137    fn advance(&mut self, mut cnt: usize) {
138        let remaining_header = self.len - self.pos;
139        if remaining_header > 0 {
140            let advanced = usize::min(cnt, remaining_header);
141            self.pos += advanced;
142            cnt -= advanced;
143        }
144        self.payload.advance(cnt);
145    }
146}
http3_datagram/datagram.rs

http3_datagram/
datagram.rs
