grpc_core/http2/
mod.rs

1//! Sans-IO streaming HTTP/2 with independent connection and stream flow control.
2//!
3//! Frame and HPACK codecs are pure Rust. Application consumption returns receive
4//! credit; body producers run only when both peer windows permit transmission.
5mod headers;
6mod receive;
7mod send;
8mod types;
9
10use std::collections::{BTreeMap, VecDeque};
11
12pub use nexora_proto::http2::{
13    HpackDecoder, Http2FrameType as FrameType, Http2Role, Http2Settings,
14};
15use nexora_proto::{
16    BytesMut,
17    http2::{CLIENT_PREFACE, HpackEncoder},
18};
19pub use types::{DataPoll, DataProvider, DataSource, HeaderField, ProtocolError, SessionEvent};
20use types::{Result, error};
21
22struct Stream {
23    send_window: i64,
24    recv_window: i64,
25    unconsumed: usize,
26    credit: u32,
27    provider: Option<DataProvider>,
28    deferred: bool,
29    initial: bool,
30    local_closed: bool,
31    remote_closed: bool,
32    close_emitted: bool,
33    reset: bool,
34}
35impl Stream {
36    fn new(local: Http2Settings, peer: Http2Settings) -> Self {
37        Self {
38            send_window: peer.initial_window_size.into(),
39            recv_window: local.initial_window_size.into(),
40            unconsumed: 0,
41            credit: 0,
42            provider: None,
43            deferred: false,
44            initial: false,
45            local_closed: false,
46            remote_closed: false,
47            close_emitted: false,
48            reset: false,
49        }
50    }
51}
52enum Outgoing {
53    Frame {
54        bytes: Vec<u8>,
55        id: i32,
56        kind: FrameType,
57        flags: u8,
58    },
59    Headers {
60        id: i32,
61        block: Vec<u8>,
62        end: bool,
63    },
64}
65struct HeaderBlock {
66    id: i32,
67    end: bool,
68    bytes: Vec<u8>,
69    reject_code: Option<u32>,
70}
71/// A single-threaded, incremental HTTP/2 connection with no socket ownership.
72pub struct Http2Session {
73    role: Http2Role,
74    local: Http2Settings,
75    peer: Http2Settings,
76    streams: BTreeMap<i32, Stream>,
77    next_id: i32,
78    last_peer: i32,
79    last_processed: i32,
80    last_sent: i32,
81    input: BytesMut,
82    preface: usize,
83    received_settings: bool,
84    pending_ack: bool,
85    encoder: HpackEncoder,
86    decoder: HpackDecoder,
87    headers: Option<HeaderBlock>,
88    outgoing: VecDeque<Outgoing>,
89    events: VecDeque<SessionEvent>,
90    output: Vec<u8>,
91    send_window: i64,
92    recv_window: i64,
93    credit: u32,
94    sent_goaway: Option<i32>,
95    received_goaway: Option<i32>,
96    failed: Option<ProtocolError>,
97}
98impl Http2Session {
99    /// Creates a peer and queues its connection preface and local settings.
100    /// # Errors
101    /// Rejects invalid frame and flow-control window sizes.
102    pub fn new(role: Http2Role, local: Http2Settings) -> Result<Self> {
103        if local.initial_window_size > i32::MAX as u32
104            || !(16_384..=16_777_215).contains(&local.max_frame_size)
105        {
106            return Err(error(1, "invalid local settings"));
107        }
108        let mut session = Self {
109            role,
110            local,
111            peer: Http2Settings::default(),
112            streams: BTreeMap::new(),
113            next_id: 1,
114            last_peer: 0,
115            last_processed: 0,
116            last_sent: 0,
117            input: BytesMut::new(),
118            preface: if role == Http2Role::Client {
119                CLIENT_PREFACE.len()
120            } else {
121                0
122            },
123            received_settings: false,
124            pending_ack: true,
125            encoder: HpackEncoder::default(),
126            decoder: HpackDecoder::new(
127                local.header_table_size as usize,
128                local.max_header_list_size as usize,
129            ),
130            headers: None,
131            outgoing: VecDeque::new(),
132            events: VecDeque::new(),
133            output: Vec::new(),
134            send_window: 65_535,
135            recv_window: 65_535,
136            credit: 0,
137            sent_goaway: None,
138            received_goaway: None,
139            failed: None,
140        };
141        if role == Http2Role::Client {
142            session.outgoing.push_back(Outgoing::Frame {
143                bytes: CLIENT_PREFACE.to_vec(),
144                id: 0,
145                kind: FrameType::Unknown(255),
146                flags: 0,
147            });
148        }
149        let mut settings = Vec::new();
150        for (id, value) in [
151            (1u16, local.header_table_size),
152            (3, local.max_concurrent_streams),
153            (4, local.initial_window_size),
154            (5, local.max_frame_size),
155            (6, local.max_header_list_size),
156        ] {
157            settings.extend_from_slice(&id.to_be_bytes());
158            settings.extend_from_slice(&value.to_be_bytes());
159        }
160        if role == Http2Role::Client {
161            settings.extend_from_slice(&[0, 2, 0, 0, 0, 0]);
162        }
163        session.queue_frame(0, FrameType::Settings, 0, &settings);
164        Ok(session)
165    }
166    pub const fn role(&self) -> Http2Role {
167        self.role
168    }
169    pub const fn last_proc_stream_id(&self) -> i32 {
170        self.last_processed
171    }
172    pub fn next_event(&mut self) -> Option<SessionEvent> {
173        self.events.pop_front()
174    }
175    pub fn want_read(&self) -> bool {
176        self.failed.is_none()
177            && (self.received_goaway.is_none() || self.streams.values().any(|s| !s.close_emitted))
178    }
179    pub fn want_write(&self) -> bool {
180        !self.outgoing.is_empty()
181            || (self.send_window > 0
182                && self
183                    .streams
184                    .values()
185                    .any(|s| s.provider.is_some() && !s.deferred && !s.reset && s.send_window > 0))
186    }
187    /// Releases a closed stream's protocol bookkeeping.
188    /// # Errors
189    /// Rejects removal before the closure event has been emitted.
190    pub fn release_stream(&mut self, id: i32) -> Result<()> {
191        if !self.streams.get(&id).is_some_and(|s| s.close_emitted) {
192            return Err(error(1, "stream still open"));
193        }
194        self.streams.remove(&id);
195        Ok(())
196    }
197    fn check_live(&self) -> Result<()> {
198        self.failed.clone().map_or(Ok(()), Err)
199    }
200    fn close_if_complete(&mut self, id: i32) {
201        if let Some(stream) = self.streams.get_mut(&id)
202            && stream.local_closed
203            && stream.remote_closed
204            && !stream.close_emitted
205        {
206            stream.close_emitted = true;
207            stream.provider = None;
208            self.events.push_back(SessionEvent::StreamClose {
209                stream_id: id,
210                error_code: 0,
211            });
212        }
213    }
214    fn stream_failure(&mut self, id: i32, resource: bool, message: impl Into<String>) {
215        self.events.push_back(SessionEvent::StreamError {
216            stream_id: id,
217            resource,
218            message: message.into(),
219        });
220    }
221}
grpc_core/http2/mod.rs

grpc_core/http2/
mod.rs
