grpc_core/
session.rs

1//! Synchronous HTTP/2 gRPC sessions with bounded per-stream receive and send queues.
2//!
3//! Receive budgets cover metadata, message descriptors and reserved partial envelopes.
4//! Native HTTP/2 storage, temporary copies and queued IncomingRequest metadata are separate.
5mod receive;
6use std::{
7    cell::RefCell,
8    collections::{BTreeMap, VecDeque},
9    io,
10    marker::PhantomData,
11    rc::Rc,
12    time::Duration,
13};
14
15use crate::{
16    compression::{CompressionAlgorithm, CompressionConfig},
17    framing::{AlignedMessageBuf, LengthPrefixedFrame},
18    http2::{
19        DataPoll, DataProvider, DataSource, FrameType, HeaderField, Http2Role, Http2Session,
20        Http2Settings, SessionEvent,
21    },
22    metadata::{Metadata, format_grpc_timeout, parse_grpc_timeout},
23    slot_map::{EndpointRole, Headers, SlotMapError, StreamLimits, StreamMessage, StreamSlotMap},
24    status::{Code, Status, decode_grpc_message},
25    transport::SyncTransport,
26};
27
28/// Relative request deadline represented by the grpc-timeout header.
29pub type Timeout = Duration;
30/// Client role marker.
31pub enum Client {}
32/// Server role marker.
33pub enum Server {}
34/// Client RPC engine.
35pub type GrpcClientSession = GrpcSession<Client>;
36/// Server RPC engine.
37pub type GrpcServerSession = GrpcSession<Server>;
38const ACCEPT: &str = "zstd,lz4,gzip,deflate,snappy,identity";
39/// HTTP/2 settings and application-owned stream limits.
40#[derive(Clone, Debug)]
41pub struct SessionConfig {
42    pub max_concurrent_streams: u32,
43    pub initial_window_size: u32,
44    pub max_frame_size: u32,
45    pub max_header_list_size: u32,
46    pub max_inbound_message_size: usize,
47    pub per_stream_byte_budget: usize,
48    pub max_active_streams: usize,
49    pub compression: CompressionConfig,
50    pub user_agent: String,
51    pub content_type: String,
52    pub scheme: String,
53}
54#[rustfmt::skip]
55impl Default for SessionConfig {
56    fn default() -> Self {
57        Self {
58            max_concurrent_streams: 100, initial_window_size: 1024 * 1024,
59            max_frame_size: 16_384, max_header_list_size: 64 * 1024,
60            max_inbound_message_size: 16 * 1024 * 1024, per_stream_byte_budget: 16 * 1024 * 1024,
61            max_active_streams: 1024, compression: CompressionConfig::default(),
62            user_agent: "grpc-rs/0.1.0".into(), scheme: "http".into(), content_type: "application/grpc".into(),
63        }
64    }
65}
66/// A validated request header block; message bodies are consumed separately.
67#[derive(Debug)]
68pub struct IncomingRequest {
69    pub stream_id: i32,
70    pub path: String,
71    pub content_type: String,
72    pub authority: String,
73    pub timeout: Option<Timeout>,
74    pub request_encoding: CompressionAlgorithm,
75    pub accept_encoding: String,
76    pub initial_metadata: Metadata,
77}
78#[derive(Default)]
79struct Outbound {
80    bytes: VecDeque<u8>,
81    eof: bool,
82    deferred: bool,
83    finished: bool,
84    trailers: bool,
85}
86struct Source {
87    queue: Rc<RefCell<Outbound>>,
88    chunk: Vec<u8>,
89}
90#[rustfmt::skip]
91impl DataSource for Source {
92    fn read(&mut self, max_len: usize) -> DataPoll<'_> {
93        let mut queue = self.queue.borrow_mut();
94        self.chunk.clear();
95        if !queue.bytes.is_empty() {
96            let count = max_len.min(queue.bytes.len());
97            self.chunk.extend(queue.bytes.drain(..count));
98            return DataPoll::Copy(&self.chunk);
99        }
100        if queue.eof {
101            queue.finished = true;
102            if queue.trailers { DataPoll::EofWithTrailers } else { DataPoll::Eof }
103        } else {
104            queue.deferred = true;
105            DataPoll::Deferred
106        }
107    }
108}
109#[rustfmt::skip]
110struct Stream {
111    decoder: LengthPrefixedFrame,
112    encoding: CompressionAlgorithm,
113    outgoing_encoding: CompressionAlgorithm,
114    outgoing: Rc<RefCell<Outbound>>,
115    headers: Headers, header_bytes: usize,
116    initial: bool, sent_initial: bool, terminal: bool, closed: bool, end_stream_serialized: bool, dispatched: bool,
117    failure: Option<Status>, failure_taken: bool, http_status: u16,
118    pending_trailers: Option<Vec<(String, String)>>,
119    prefix: Vec<u8>, remaining: usize, reserved: usize,
120}
121/// Single-threaded session. Closed slots remain readable until explicitly removed.
122#[rustfmt::skip]
123pub struct GrpcSession<Role> {
124    protocol: Http2Session, config: SessionConfig, slots: StreamSlotMap,
125    streams: BTreeMap<i32, Stream>,
126    incoming: VecDeque<IncomingRequest>,
127    wire: Vec<u8>, transport_pending: Vec<u8>, transport_offset: usize, transport_flushed: bool,
128    failed: Option<Status>, draining: bool, role: PhantomData<Role>,
129}
130#[rustfmt::skip]
131fn internal(error: impl std::fmt::Display) -> Status { Status::internal(error.to_string()) }
132#[rustfmt::skip]
133fn resource() -> Status { Status::resource_exhausted("per-stream byte budget exceeded") }
134#[rustfmt::skip]
135fn slot_error(error: SlotMapError) -> Status {
136    match error { SlotMapError::MemoryLimit { .. } | SlotMapError::MessageLimit { .. } => Status::resource_exhausted(error.to_string()), _ => internal(error) }
137}
138#[rustfmt::skip]
139fn fields(headers: &[(String, String)]) -> Vec<HeaderField<'_>> {
140    headers.iter().map(|(k, v)| HeaderField::new(k.as_bytes(), v.as_bytes())).collect()
141}
142#[rustfmt::skip]
143fn metadata(headers: &Headers) -> Result<Metadata, Status> {
144    let mut result = Metadata::new();
145    for (key, value) in headers {
146        if key.starts_with(':') || matches!(key.as_str(), "content-type" | "te" | "user-agent") || key.starts_with("grpc-") {
147            continue;
148        }
149        result.append_header(key, std::str::from_utf8(value).map_err(internal)?).map_err(internal)?;
150    }
151    Ok(result)
152}
153#[rustfmt::skip]
154fn validate_content_type(value: &str) -> Result<(), Status> {
155    let media = value.split(';').next().unwrap_or("");
156    let suffix = media.strip_prefix("application/grpc+");
157    if !value.bytes().all(|b| (32..127).contains(&b)) || !(media == "application/grpc" || suffix.is_some_and(|s| !s.is_empty() && s.bytes().all(|b| b.is_ascii_alphanumeric() || b"!#$%&'*+-.^_`|~".contains(&b)))) { return Err(Status::invalid_argument("invalid gRPC content-type")); }
158    Ok(())
159}
160#[rustfmt::skip]
161fn validate_metadata(metadata: &Metadata) -> Result<(), Status> {
162    if metadata.iter().any(|(key, _)| matches!(key, "content-type" | "te" | "user-agent")) { return Err(Status::invalid_argument("application metadata contains protocol header")); }
163    Ok(())
164}
165#[rustfmt::skip]
166fn value<'a>(headers: &'a Headers, key: &str) -> Result<Option<&'a str>, Status> {
167    let mut values = headers.iter().filter(|(k, _)| k == key);
168    let result = values.next().map(|(_, v)| std::str::from_utf8(v).map_err(internal)).transpose()?;
169    if values.next().is_some() { return Err(Status::internal(format!("duplicate {key}"))); }
170    Ok(result)
171}
172#[rustfmt::skip]
173macro_rules! constructor {
174    ($role:ty, $variant:ident, $endpoint:expr) => {
175        impl GrpcSession<$role> {
176            /// Creates an engine and queues its initial HTTP/2 SETTINGS.
177            /// # Errors
178            /// Rejects invalid settings and protocol initialization failure.
179            pub fn new(config: SessionConfig) -> Result<Self, Status> {
180                if config.max_active_streams == 0
181                    || config.initial_window_size > i32::MAX as u32
182                    || !(16_384..=16_777_215).contains(&config.max_frame_size)
183                    || !matches!(config.scheme.as_str(), "http" | "https")
184                {
185                    return Err(Status::invalid_argument("invalid HTTP/2 session configuration"));
186                }
187                validate_content_type(&config.content_type)?;
188                let protocol = Http2Session::new(Http2Role::$variant, Http2Settings {
189                    max_concurrent_streams: config.max_concurrent_streams,
190                    initial_window_size: config.initial_window_size,
191                    max_frame_size: config.max_frame_size,
192                    max_header_list_size: config.max_header_list_size,
193                    enable_push: false, ..Http2Settings::default()
194                }).map_err(internal)?;
195                let slots = StreamSlotMap::with_role(StreamLimits { max_buffered_bytes: config.per_stream_byte_budget, max_queued_messages: 1024 }, $endpoint);
196                Ok(Self {
197                    protocol, config, slots,
198                    streams: BTreeMap::new(), incoming: VecDeque::new(),
199                    wire: Vec::new(), transport_pending: Vec::new(),
200                    transport_offset: 0, transport_flushed: true, failed: None, draining: false, role: PhantomData,
201                })
202            }
203        }
204    };
205}
206constructor!(Client, Client, EndpointRole::Client);
207constructor!(Server, Server, EndpointRole::Server);
208#[rustfmt::skip]
209impl GrpcClientSession {
210    /// Starts a POST request; false END_STREAM enables incremental message sending.
211    /// # Errors
212    /// Rejects stream limits, invalid metadata/timeout or protocol submission failure.
213    pub fn start_request(&mut self, authority: &str, path: &str, metadata: &Metadata, timeout: Option<Timeout>, encoding: CompressionAlgorithm, end_stream: bool) -> Result<i32, Status> {
214        let content_type = self.config.content_type.clone();
215        self.start_request_with_content_type(authority, path, metadata, timeout, encoding, end_stream, &content_type)
216    }
217    /// Starts a request with a codec media type independently of other RPCs.
218    /// # Errors
219    /// Rejects invalid media types, metadata, stream limits or protocol submission failure.
220    #[expect(clippy::too_many_arguments, reason = "The request carries routing, metadata, deadline, encoding and codec identity.")]
221    pub fn start_request_with_content_type(&mut self, authority: &str, path: &str, metadata: &Metadata, timeout: Option<Timeout>, encoding: CompressionAlgorithm, end_stream: bool, content_type: &str) -> Result<i32, Status> {
222        validate_content_type(content_type)?;
223        self.check_capacity()?;
224        if authority.is_empty() || !path.starts_with('/') { return Err(Status::invalid_argument("invalid RPC authority or path")); }
225        let mut headers = vec![
226            (":method".into(), "POST".into()),
227            (":scheme".into(), self.config.scheme.clone()),
228            (":authority".into(), authority.into()),
229            (":path".into(), path.into()),
230            ("te".into(), "trailers".into()),
231            ("content-type".into(), content_type.into()),
232            ("user-agent".into(), self.config.user_agent.clone()),
233            ("grpc-accept-encoding".into(), ACCEPT.into()),
234        ];
235        if encoding != CompressionAlgorithm::Identity {
236            headers.push(("grpc-encoding".into(), encoding.as_str().into()));
237        }
238        if let Some(timeout) = timeout {
239            headers.push(("grpc-timeout".into(), format_grpc_timeout(timeout).map_err(internal)?));
240        }
241        validate_metadata(metadata)?;
242        headers.extend(metadata.to_headers());
243        self.check_headers(&headers)?;
244        let queue = Rc::new(RefCell::new(Outbound::default()));
245        let provider = (!end_stream).then(|| DataProvider::new(Source { queue: Rc::clone(&queue), chunk: Vec::new() }));
246        let id = self.protocol.submit_request(&fields(&headers), provider).map_err(internal)?;
247        self.open(id, queue, encoding)?;
248        if end_stream {
249            self.slots.mark_local_end_stream(id).map_err(slot_error)?;
250            self.stream_mut(id)?.outgoing.borrow_mut().eof = true;
251        }
252        Ok(id)
253    }
254    /// Ends the request body after all queued messages.
255    /// # Errors
256    /// Rejects unknown, reset or already half-closed streams.
257    pub fn half_close(&mut self, id: i32) -> Result<(), Status> {
258        self.check_send(id)?;
259        self.stream_mut(id)?.outgoing.borrow_mut().eof = true;
260        self.slots.mark_local_end_stream(id).map_err(slot_error)?;
261        self.resume(id)
262    }
263}
264#[rustfmt::skip]
265impl GrpcServerSession {
266    /// Releases safely closed streams that were never handed to a request consumer.
267    /// Dispatched streams retain their terminal state until explicitly removed.
268    ///
269    /// # Errors
270    /// Returns a stream-removal error if its closure state is inconsistent.
271    pub fn reap_closed_undispatched_streams(&mut self) -> Result<usize, Status> {
272        let closed: Vec<_> = self
273            .streams
274            .iter()
275            .filter(|(id, stream)| {
276                !stream.dispatched && stream.closed && self.stream_outbound_complete(**id)
277            })
278            .map(|(id, _)| *id)
279            .collect();
280        for id in &closed {
281            self.remove_closed_stream(*id)?;
282        }
283        Ok(closed.len())
284    }
285    /// Transfers the next received request's routing information.
286    pub fn pop_incoming_request(&mut self) -> Option<IncomingRequest> { let request = self.incoming.pop_front()?; if let Some(stream) = self.streams.get_mut(&request.stream_id) { stream.dispatched = true; } Some(request) }
287    /// Sends response headers and installs an incremental body source.
288    /// # Errors
289    /// Rejects duplicate headers, ended streams and protocol submission failure.
290    pub fn send_initial_metadata(&mut self, id: i32, metadata: &Metadata, encoding: CompressionAlgorithm) -> Result<(), Status> { self.send_initial_metadata_with_content_type(id, metadata, encoding, &self.config.content_type.clone()) }
291    /// Starts a response using the codec selected for this RPC's route.
292    /// # Errors
293    /// Rejects invalid media types, duplicate headers, ended streams or protocol failure.
294    pub fn send_initial_metadata_with_content_type(&mut self, id: i32, metadata: &Metadata, encoding: CompressionAlgorithm, content_type: &str) -> Result<(), Status> { self.begin_response(id, metadata, encoding, content_type) }
295    /// Completes a response, emitting trailers only when its queued DATA has drained.
296    /// # Errors
297    /// Rejects duplicate completion, invalid streams, budgets or protocol failure.
298    pub fn finish_response(&mut self, id: i32, status: &Status, metadata: &Metadata) -> Result<(), Status> {
299        self.check_send(id)?;
300        let mut headers = vec![("grpc-status".into(), status.code().as_i32().to_string())];
301        if !status.message().is_empty() {
302            headers.push(("grpc-message".into(), status.encoded_message()));
303        }
304        validate_metadata(metadata)?;
305        headers.extend(metadata.to_headers());
306        if !self.stream_mut(id)?.sent_initial {
307            headers.insert(0, ("content-type".into(), self.config.content_type.clone()));
308            headers.insert(0, (":status".into(), "200".into()));
309            self.check_headers(&headers)?;
310            self.protocol.submit_response(id, &fields(&headers), None).map_err(internal)?;
311        } else {
312            self.check_headers(&headers)?;
313            let stream = self.stream_mut(id)?;
314            let mut queue = stream.outgoing.borrow_mut();
315            queue.eof = true;
316            queue.trailers = true;
317            stream.pending_trailers = Some(headers);
318            drop(queue);
319            self.resume(id)?;
320        }
321        self.stream_mut(id)?.dispatched = true; self.slots.mark_local_end_stream(id).map_err(slot_error)
322    }
323}
324#[rustfmt::skip]
325impl<Role> GrpcSession<Role> {
326    fn begin_response(&mut self, id: i32, metadata: &Metadata, encoding: CompressionAlgorithm, content_type: &str) -> Result<(), Status> {
327        validate_content_type(content_type)?;
328        self.check_send(id)?;
329        if self.stream_mut(id)?.sent_initial { return Err(Status::failed_precondition("response headers already sent")); }
330        let mut headers = vec![(":status".into(), "200".into()), ("content-type".into(), content_type.into()), ("grpc-accept-encoding".into(), ACCEPT.into())];
331        if encoding != CompressionAlgorithm::Identity {
332            headers.push(("grpc-encoding".into(), encoding.as_str().into()));
333        }
334        validate_metadata(metadata)?;
335        headers.extend(metadata.to_headers());
336        self.check_headers(&headers)?;
337        let queue = Rc::clone(&self.stream_mut(id)?.outgoing);
338        self.protocol.submit_response(id, &fields(&headers), Some(DataProvider::new(Source { queue, chunk: Vec::new() }))).map_err(internal)?;
339        let stream = self.stream_mut(id)?;
340        stream.sent_initial = true; stream.dispatched = true;
341        stream.outgoing_encoding = encoding;
342        Ok(())
343    }
344
345    /// Queues GOAWAY at the last processed peer stream; existing RPCs may finish.
346    /// # Errors
347    /// Returns protocol submission or a previous connection failure.
348    pub fn submit_goaway(&mut self, error_code: u32) -> Result<(), Status> {
349        self.check_live()?;
350        let last = self.protocol.last_proc_stream_id();
351        self.protocol.submit_goaway(last, error_code, &[]).map_err(internal)?;
352        self.draining = true; Ok(())
353    }
354    /// Whether the protocol connection can accept additional inbound bytes.
355    pub fn want_read(&self) -> bool { self.failed.is_none() && self.protocol.want_read() }
356    /// Whether protocol frames or partially written transport bytes need sending.
357    pub fn want_write(&self) -> bool { !self.transport_flushed || self.transport_offset < self.transport_pending.len() || self.streams.values().any(|s| s.pending_trailers.is_some() && s.outgoing.borrow().finished) || self.protocol.want_write() }
358    /// The most restrictive peer GOAWAY boundary and its HTTP/2 error code.
359    pub fn goaway_received(&self) -> Option<(i32, u32)> { self.slots.goaway() }
360    /// Whether terminal output was serialized and tracked transport writes/flushes drained.
361    pub fn stream_outbound_complete(&self, id: i32) -> bool { self.transport_flushed && self.transport_offset == self.transport_pending.len() && self.streams.get(&id).is_some_and(|s| s.closed || s.end_stream_serialized) }
362    /// Whether local or peer GOAWAY has stopped admission of new RPCs.
363    pub fn is_draining(&self) -> bool { self.draining || self.slots.goaway().is_some() }
364    fn client(&self) -> bool { self.protocol.role() == Http2Role::Client }
365    fn check_live(&self) -> Result<(), Status> { self.failed.clone().map_or(Ok(()), Err) }
366    fn check_capacity(&self) -> Result<(), Status> {
367        self.check_live()?;
368        if self.is_draining() { return Err(Status::unavailable("connection is draining")); }
369        if self.streams.len() >= self.config.max_active_streams { return Err(Status::resource_exhausted("active stream limit exceeded")); }
370        Ok(())
371    }
372    fn check_headers(&self, headers: &[(String, String)]) -> Result<(), Status> {
373        let bytes = headers.iter().try_fold(0usize, |n, (k, v)| n.checked_add(k.len() + v.len() + 32)).ok_or_else(resource)?;
374        if bytes > self.config.max_header_list_size as usize || bytes > self.config.per_stream_byte_budget { return Err(resource()); }
375        Ok(())
376    }
377    fn stream_mut(&mut self, id: i32) -> Result<&mut Stream, Status> { self.streams.get_mut(&id).ok_or_else(|| Status::not_found("unknown stream")) }
378    fn open(&mut self, id: i32, queue: Rc<RefCell<Outbound>>, encoding: CompressionAlgorithm) -> Result<(), Status> {
379        self.check_capacity()?;
380        self.slots.open_stream(id).map_err(slot_error)?;
381        self.streams.insert(
382            id,
383            Stream {
384                decoder: LengthPrefixedFrame::new(self.config.max_inbound_message_size.min(self.config.per_stream_byte_budget)),
385                encoding: CompressionAlgorithm::Identity,
386                outgoing_encoding: encoding,
387                outgoing: queue,
388                headers: vec![], header_bytes: 0, initial: false, sent_initial: false, terminal: false, closed: false, end_stream_serialized: false, dispatched: false,
389                failure: None, failure_taken: false, http_status: 200, pending_trailers: None,
390                prefix: vec![], remaining: 0, reserved: 0,
391            },
392        );
393        Ok(())
394    }
395    fn check_send(&self, id: i32) -> Result<(), Status> {
396        self.check_live()?;
397        let stream = self.streams.get(&id).ok_or_else(|| Status::not_found("unknown stream"))?;
398        if let Some(error) = &stream.failure { return Err(error.clone()); }
399        let slot = self.slots.get(id).ok_or_else(|| Status::not_found("unknown stream"))?;
400        if slot.local_end_stream() { return Err(Status::failed_precondition("sending half already closed")); }
401        Ok(())
402    }
403    fn resume(&mut self, id: i32) -> Result<(), Status> {
404        let deferred = self.stream_mut(id)?.outgoing.borrow().deferred;
405        if deferred {
406            self.protocol.resume_data(id).map_err(internal)?;
407            self.stream_mut(id)?.outgoing.borrow_mut().deferred = false;
408        }
409        Ok(())
410    }
411    /// Queues one framed message, preserving the stream's negotiated encoding.
412    /// # Errors
413    /// Rejects ended streams, mismatched encodings, compression failure or byte excess.
414    pub fn send_message(&mut self, id: i32, payload: &[u8], encoding: CompressionAlgorithm) -> Result<(), Status> {
415        self.check_send(id)?;
416        if !self.client() && !self.stream_mut(id)?.sent_initial {
417            self.begin_response(id, &Metadata::new(), encoding, &self.config.content_type.clone())?;
418        }
419
420        if encoding != CompressionAlgorithm::Identity && encoding != self.stream_mut(id)?.outgoing_encoding { return Err(Status::invalid_argument("message encoding differs from stream encoding")); }
421        if payload.len() > self.config.per_stream_byte_budget { return Err(resource()); }
422        let compressed = CompressionConfig { algorithm: encoding, ..self.config.compression }.compress(payload)?;
423        let frame = AlignedMessageBuf::new(&compressed.data, compressed.compressed)?;
424        let budget = self.config.per_stream_byte_budget;
425        let stream = self.stream_mut(id)?;
426        let mut queue = stream.outgoing.borrow_mut();
427        if queue.bytes.len().checked_add(frame.wire_bytes().len()).is_none_or(|n| n > budget) { return Err(resource()); }
428        queue.bytes.extend(frame.wire_bytes());
429        drop(queue);
430        self.resume(id)
431    }
432    /// Resets a stream and releases its queued outbound body.
433    /// Completed NO_ERROR peer closure needs no additional local upload reset.
434    /// # Errors
435    /// Rejects unknown streams or protocol reset submission failure.
436    pub fn cancel_stream(&mut self, id: i32, error_code: u32) -> Result<(), Status> {
437        self.check_read(id)?;
438        if error_code == 0 && self.streams.get(&id).is_some_and(|s| s.closed && s.terminal) { return Ok(()); }
439        self.protocol.submit_rst_stream(id, error_code).map_err(internal)?;
440        self.fail_stream(id, Status::cancelled("stream cancelled"), error_code)
441    }
442    fn fail_stream(&mut self, id: i32, error: Status, code: u32) -> Result<(), Status> {
443        let stream = self.stream_mut(id)?;
444        if stream.failure.is_some() { return Ok(()); }
445        stream.failure = Some(error);
446        stream.decoder.reset();
447        stream.reserved = 0;
448        stream.prefix.clear();
449        stream.headers.clear();
450        stream.pending_trailers = None;
451        stream.outgoing.borrow_mut().bytes.clear();
452        self.incoming.retain(|r| r.stream_id != id);
453        self.slots.record_rst_stream(id, code).map_err(slot_error)
454    }
455    /// Feeds received HTTP/2 bytes and dispatches every resulting callback event.
456    /// # Errors
457    /// Returns protocol connection errors; malformed RPC streams are reset individually.
458    pub fn mem_recv(&mut self, bytes: &[u8]) -> Result<usize, Status> {
459        self.check_live()?;
460        let mut count = 0;
461        for chunk in bytes.chunks(16 * 1024) {
462            let n = self.protocol.mem_recv(chunk).map_err(internal)?;
463            count += n;
464            self.events()?;
465            if n < chunk.len() {
466                break;
467            }
468        }
469        Ok(count)
470    }
471    /// Produces the next HTTP/2 wire chunk, valid until the next mutable session call.
472    /// # Errors
473    /// Returns protocol connection or callback-processing errors.
474    pub fn mem_send(&mut self) -> Result<&[u8], Status> {
475        if self.transport_offset < self.transport_pending.len() { return Err(Status::failed_precondition("transport output must drain before mem_send")); }
476        self.check_live()?;
477        self.submit_ready_trailers()?;
478        self.wire = self.protocol.mem_send().map_err(internal)?.to_vec();
479        self.events()?;
480        if self.wire.is_empty() {
481            self.submit_ready_trailers()?;
482            self.wire = self.protocol.mem_send().map_err(internal)?.to_vec();
483            self.events()?;
484        }
485        Ok(&self.wire)
486    }
487    fn submit_ready_trailers(&mut self) -> Result<(), Status> {
488        let ready: Vec<_> = self.streams.iter().filter(|(_, s)| s.pending_trailers.is_some() && s.outgoing.borrow().finished).map(|(id, _)| *id).collect();
489        for id in ready {
490            if let Some(headers) = self.stream_mut(id)?.pending_trailers.take() {
491                self.protocol.submit_trailer(id, &fields(&headers)).map_err(internal)?;
492            }
493        }
494        Ok(())
495    }
496    /// Flushes wire output, retaining partially written chunks across WouldBlock.
497    /// # Errors
498    /// Returns transport or session failures; WouldBlock means retry later.
499    pub fn flush_to_transport<T: SyncTransport>(&mut self, transport: &mut T) -> Result<(), Status> {
500        loop {
501            if self.transport_offset == self.transport_pending.len() {
502                self.transport_pending = self.mem_send()?.to_vec();
503                self.transport_offset = 0;
504                if !self.transport_pending.is_empty() { self.transport_flushed = false; }
505                if self.transport_pending.is_empty() {
506                    break;
507                }
508            }
509            match transport.write(&self.transport_pending[self.transport_offset..]) {
510                Ok(0) => return Err(Status::unavailable("transport write returned zero")),
511                Ok(n) => self.transport_offset += n,
512                Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
513                Err(e) if e.kind() == io::ErrorKind::WouldBlock => return Ok(()),
514                Err(e) => return Err(Status::unavailable(e.to_string())),
515            }
516        }
517        match transport.flush() {
518            Ok(()) => { self.transport_flushed = true; Ok(()) },
519            Err(e) if e.kind() == io::ErrorKind::WouldBlock => { self.transport_flushed = false; Ok(()) },
520            Err(e) => Err(Status::unavailable(e.to_string())),
521        }
522    }
523    /// Flushes outgoing data, reads once and routes inbound bytes.
524    /// # Errors
525    /// Returns transport EOF/errors or session errors. WouldBlock returns zero.
526    pub fn step_transport<T: SyncTransport>(&mut self, transport: &mut T, buf: &mut [u8]) -> Result<usize, Status> {
527        if buf.is_empty() { return Err(Status::invalid_argument("receive buffer is empty")); }
528        self.flush_to_transport(transport)?;
529        let n = loop {
530            match transport.read(buf) {
531                Ok(0) => return Err(Status::unavailable("transport EOF")),
532                Ok(n) => break n,
533                Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
534                Err(e) if e.kind() == io::ErrorKind::WouldBlock => return Ok(0),
535                Err(e) => return Err(Status::unavailable(e.to_string())),
536            }
537        };
538        let mut used = 0;
539        while used < n {
540            let count = self.mem_recv(&buf[used..n])?;
541            if count == 0 { return Err(Status::internal("HTTP/2 receive made no progress")); }
542            used += count;
543        }
544        self.flush_to_transport(transport)?;
545        Ok(n)
546    }
547    /// Transfers custom initial metadata, preserving stream lifecycle history.
548    /// # Errors
549    /// Returns unknown/reset stream or metadata decoding errors.
550    pub fn pop_initial_metadata(&mut self, id: i32) -> Result<Option<Metadata>, Status> {
551        self.check_read(id)?;
552        self.slots.pop_initial_metadata(id).map_err(slot_error)?.map(|h| metadata(&h)).transpose()
553    }
554    /// Transfers one decompressed message into eight-byte-aligned storage.
555    /// # Errors
556    /// Returns unknown/reset stream or allocation errors.
557    pub fn pop_message(&mut self, id: i32) -> Result<Option<AlignedMessageBuf>, Status> {
558        self.check_read(id)?;
559        self.slots.pop_message(id).map_err(slot_error)?.map(|m| AlignedMessageBuf::new(&m.payload, false)).transpose()
560    }
561    /// Whether the peer's request or response half has ended.
562    /// # Errors
563    /// Returns unknown/reset stream errors.
564    pub fn is_remote_end_stream(&self, id: i32) -> Result<bool, Status> {
565        self.check_read(id)?;
566        Ok(self.slots.get(id).is_some_and(|s| s.remote_end_stream()))
567    }
568    fn check_read(&self, id: i32) -> Result<(), Status> {
569        let s = self.streams.get(&id).ok_or_else(|| Status::not_found("unknown stream"))?;
570        s.failure.clone().map_or(Ok(()), Err)
571    }
572    /// Takes terminal status and custom trailing metadata exactly once.
573    /// # Errors
574    /// Returns unknown/reset stream or invalid metadata errors.
575    pub fn take_trailers(&mut self, id: i32) -> Result<Option<(Status, Metadata)>, Status> {
576        if let Some(error) = self.stream_mut(id)?.failure.clone() {
577            if self.stream_mut(id)?.failure_taken { return Ok(None); }
578            self.stream_mut(id)?.failure_taken = true;
579            return Ok(Some((error, Metadata::new())));
580        }
581        self.slots
582            .take_trailers(id)
583            .map_err(slot_error)?
584            .map(|h| {
585                let code = value(&h, "grpc-status")?.and_then(|v| v.parse::<i32>().ok()).map_or(Code::Unknown, Code::from_i32);
586                let message = decode_grpc_message(value(&h, "grpc-message")?.unwrap_or(""));
587                Ok((Status::new(code, message), metadata(&h)?))
588            })
589            .transpose()
590    }
591    /// Releases a closed stream and all unread events.
592    /// # Errors
593    /// Rejects unknown or still-open streams.
594    pub fn remove_closed_stream(&mut self, id: i32) -> Result<(), Status> {
595        if !self.stream_mut(id)?.closed { return Err(Status::failed_precondition("protocol stream is not closed")); }
596        if self.slots.get(id).is_some() {
597            self.slots.remove_closed_stream(id).map_err(slot_error)?;
598        }
599        self.protocol.release_stream(id).map_err(internal)?;
600        self.streams.remove(&id);
601        self.incoming.retain(|r| r.stream_id != id);
602        Ok(())
603    }
604}
grpc_core/session.rs

grpc_core/
session.rs
