grpc/
channel.rs

1//! Caller-thread client channels with shared HTTP/2 sessions and typed RPC streams.
2use std::{
3    cell::RefCell,
4    collections::BTreeSet,
5    io,
6    rc::Rc,
7    time::{Duration, Instant},
8};
9
10use grpc_core::{
11    Code, Codec, CompressionAlgorithm, GrpcClientSession, Metadata, SessionConfig, Status, Timeout,
12    transport::{SyncTransport, TcpTransport},
13};
14
15use crate::stream::{
16    BidiStreamingCall, ClientStreamingCall, ServerStreamingCall, StreamDriver, StreamError,
17};
18
19/// Connection defaults applied before starting each RPC.
20#[derive(Clone, Debug)]
21pub struct ChannelConfig {
22    pub authority: String,
23    pub scheme: String,
24    pub default_timeout: Option<Timeout>,
25    pub default_compression: CompressionAlgorithm,
26    pub session_config: SessionConfig,
27}
28impl Default for ChannelConfig {
29    fn default() -> Self {
30        Self {
31            authority: "localhost".into(),
32            scheme: "http".into(),
33            default_timeout: None,
34            default_compression: CompressionAlgorithm::Identity,
35            session_config: SessionConfig::default(),
36        }
37    }
38}
39impl ChannelConfig {
40    /// Selects the default outgoing compression (Zstd uses level one).
41    pub fn with_compression(mut self, compression: CompressionAlgorithm) -> Self {
42        self.default_compression = compression;
43        self
44    }
45}
46/// Metadata and overrides for one RPC.
47#[derive(Clone, Debug, Default)]
48pub struct CallOptions {
49    pub metadata: Metadata,
50    pub timeout: Option<Timeout>,
51    pub compression: Option<CompressionAlgorithm>,
52}
53struct ChannelInner<T: SyncTransport> {
54    session: GrpcClientSession,
55    transport: T,
56    buffer: Vec<u8>,
57    config: ChannelConfig,
58    cleanup_error: Option<Status>,
59    pending_reap: BTreeSet<i32>,
60    pending_cancellations: Rc<RefCell<Vec<i32>>>,
61}
62impl<T: SyncTransport> ChannelInner<T> {
63    fn pump(&mut self, wait: bool, deadline: Option<Instant>) -> Result<(), Status> {
64        check_deadline(deadline)?;
65        self.transport.set_nonblocking(true).map_err(io_status)?;
66        self.flush()?;
67        let mut read = loop {
68            match self.transport.read(&mut self.buffer) {
69                Err(error) if error.kind() == io::ErrorKind::Interrupted => {
70                    check_deadline(deadline)?;
71                }
72                result => break result,
73            }
74        };
75        if matches!(&read, Err(error) if error.kind() == io::ErrorKind::WouldBlock) && wait {
76            let remaining = deadline.map(|at| at.saturating_duration_since(Instant::now()));
77            check_deadline(deadline)?;
78            let timeout = if self.session.want_write() {
79                Some(remaining.map_or(Duration::from_millis(1), |left| {
80                    left.min(Duration::from_millis(1))
81                }))
82            } else {
83                remaining
84            };
85            self.transport
86                .set_read_timeout(timeout)
87                .map_err(io_status)?;
88            self.transport.set_nonblocking(false).map_err(io_status)?;
89            read = self.transport.read(&mut self.buffer);
90            let restored = self.transport.set_nonblocking(true).map_err(io_status);
91            restored?;
92        }
93        let count = match read {
94            Ok(0) => return Err(Status::unavailable("transport EOF")),
95            Ok(count) => count,
96            Err(error)
97                if matches!(
98                    error.kind(),
99                    io::ErrorKind::WouldBlock
100                        | io::ErrorKind::TimedOut
101                        | io::ErrorKind::Interrupted
102                ) =>
103            {
104                check_deadline(deadline)?;
105                return self.reap();
106            }
107            Err(error) => return Err(io_status(error)),
108        };
109        let mut offset = 0;
110        while offset < count {
111            let used = self.session.mem_recv(&self.buffer[offset..count])?;
112            if used == 0 {
113                return Err(Status::internal("HTTP/2 receive made no progress"));
114            }
115            offset += used;
116        }
117        self.flush()?;
118        check_deadline(deadline)
119    }
120    fn drain_cancellations(&mut self) -> Result<(), Status> {
121        let ids = std::mem::take(&mut *self.pending_cancellations.borrow_mut());
122        for id in ids {
123            let reset = self.session.cancel_stream(id, 8);
124            self.pending_reap.insert(id);
125            if let Err(reset_error) = reset {
126                match self.session.remove_closed_stream(id) {
127                    Ok(()) => {
128                        self.pending_reap.remove(&id);
129                    }
130                    Err(error) if error.code() == Code::NotFound => {
131                        self.pending_reap.remove(&id);
132                    }
133                    Err(_) => {
134                        self.cleanup_error = Some(reset_error.clone());
135                        return Err(reset_error);
136                    }
137                }
138            }
139        }
140        Ok(())
141    }
142    fn reap(&mut self) -> Result<(), Status> {
143        let ids: Vec<_> = self.pending_reap.iter().copied().collect();
144        for id in ids {
145            match self.session.remove_closed_stream(id) {
146                Ok(()) => {
147                    self.pending_reap.remove(&id);
148                }
149                Err(error) if error.code() == Code::NotFound => {
150                    self.pending_reap.remove(&id);
151                }
152                Err(error) if error.code() == Code::FailedPrecondition => {}
153                Err(error) => {
154                    self.cleanup_error = Some(error.clone());
155                    return Err(error);
156                }
157            }
158        }
159        Ok(())
160    }
161    fn flush(&mut self) -> Result<(), Status> {
162        self.drain_cancellations()?;
163        self.session.flush_to_transport(&mut self.transport)?;
164        self.reap()
165    }
166}
167fn io_status(error: io::Error) -> Status {
168    Status::unavailable(error.to_string())
169}
170fn codec_status(error: impl std::fmt::Display) -> Status {
171    Status::internal(format!("codec failure: {error}"))
172}
173
174/// Shared, single-threaded connection. Clones share one HTTP/2 session.
175///
176/// Connections stay on their owning OS thread.
177/// ```compile_fail
178/// use grpc::{SyncChannel, core::transport::MemoryPipe};
179/// fn require_send<T: Send>() {}
180/// require_send::<SyncChannel<MemoryPipe>>();
181/// ```
182/// ```compile_fail
183/// use grpc::{SyncChannel, core::transport::MemoryPipe};
184/// fn require_sync<T: Sync>() {}
185/// require_sync::<SyncChannel<MemoryPipe>>();
186/// ```
187pub struct SyncChannel<T: SyncTransport> {
188    inner: Rc<RefCell<ChannelInner<T>>>,
189}
190impl<T: SyncTransport> Clone for SyncChannel<T> {
191    fn clone(&self) -> Self {
192        Self {
193            inner: Rc::clone(&self.inner),
194        }
195    }
196}
197impl<T: SyncTransport + 'static> SyncChannel<T> {
198    /// Creates a channel using standard connection defaults.
199    /// # Errors
200    /// Returns invalid configuration or transport/session initialization failures.
201    pub fn new(transport: T) -> Result<Self, Status> {
202        Self::with_config(transport, ChannelConfig::default())
203    }
204    /// Creates a nonblocking session; blocking RPC methods wait on the caller thread.
205    /// # Errors
206    /// Returns invalid authority/scheme, transport or session initialization errors.
207    pub fn with_config(mut transport: T, mut config: ChannelConfig) -> Result<Self, Status> {
208        if config.authority.is_empty()
209            || config
210                .authority
211                .bytes()
212                .any(|byte| byte.is_ascii_control() || byte.is_ascii_whitespace())
213        {
214            return Err(Status::invalid_argument("invalid channel authority"));
215        }
216        config.session_config.scheme.clone_from(&config.scheme);
217        let session = GrpcClientSession::new(config.session_config.clone())?;
218        transport.set_nonblocking(true).map_err(io_status)?;
219        Ok(Self {
220            inner: Rc::new(RefCell::new(ChannelInner {
221                session,
222                transport,
223                buffer: vec![0; 16 * 1024],
224                config,
225                cleanup_error: None,
226                pending_reap: BTreeSet::new(),
227                pending_cancellations: Rc::new(RefCell::new(Vec::new())),
228            })),
229        })
230    }
231    fn deadline(&self, options: &CallOptions) -> Result<Option<Instant>, Status> {
232        let inner = self
233            .inner
234            .try_borrow()
235            .map_err(|_| Status::failed_precondition("reentrant channel access"))?;
236        options
237            .timeout
238            .or(inner.config.default_timeout)
239            .map(|duration| {
240                Instant::now()
241                    .checked_add(duration)
242                    .ok_or_else(|| Status::invalid_argument("RPC deadline overflow"))
243            })
244            .transpose()
245    }
246    fn start<C: Codec>(
247        &mut self,
248        path: &str,
249        codec: C,
250        options: CallOptions,
251        content_type: &str,
252        absolute_deadline: Option<Option<Instant>>,
253    ) -> Result<ChannelStreamDriver<T, C>, Status> {
254        let mut inner = self
255            .inner
256            .try_borrow_mut()
257            .map_err(|_| Status::failed_precondition("reentrant channel access"))?;
258        inner.flush()?;
259        if let Some(error) = &inner.cleanup_error {
260            return Err(error.clone());
261        }
262        let deadline = match absolute_deadline {
263            Some(deadline) => deadline,
264            None => options
265                .timeout
266                .or(inner.config.default_timeout)
267                .map(|duration| {
268                    Instant::now()
269                        .checked_add(duration)
270                        .ok_or_else(|| Status::invalid_argument("RPC deadline overflow"))
271                })
272                .transpose()?,
273        };
274        let timeout = deadline
275            .map(|end| {
276                end.checked_duration_since(Instant::now())
277                    .filter(|duration| !duration.is_zero())
278                    .ok_or_else(|| Status::deadline_exceeded("RPC deadline elapsed"))
279            })
280            .transpose()?;
281        let encoding = options
282            .compression
283            .unwrap_or(inner.config.default_compression);
284        let authority = inner.config.authority.clone();
285        let id = inner.session.start_request_with_content_type(
286            &authority,
287            path,
288            &options.metadata,
289            timeout,
290            encoding,
291            false,
292            content_type,
293        )?;
294        Ok(ChannelStreamDriver {
295            inner: Rc::clone(&self.inner),
296            id,
297            codec,
298            encoding,
299            deadline,
300            initial: None,
301            trailing: None,
302            complete: false,
303            closed: false,
304            buffer: Vec::new(),
305            pending_cancellations: Rc::clone(&inner.pending_cancellations),
306        })
307    }
308    /// Performs a unary RPC, returning its response and both metadata collections.
309    ///
310    /// A mutable codec also supports stateful serializers and generated service adapters.
311    /// # Errors
312    /// Returns transport, codec, deadline, remote status or response cardinality errors.
313    pub fn unary<C, Req, Resp>(
314        &mut self,
315        path: &str,
316        req: &Req,
317        codec: &mut C,
318        options: CallOptions,
319    ) -> Result<(Resp, Metadata, Metadata), Status>
320    where
321        C: Codec<EncodeItem = Req, DecodeItem = Resp>,
322    {
323        let deadline = self.deadline(&options)?;
324        check_deadline(deadline)?;
325        let mut payload = Vec::new();
326        codec.encode(req, &mut payload).map_err(codec_status)?;
327        check_deadline(deadline)?;
328        let mut driver = self.start(path, BytesCodec, options, C::CONTENT_TYPE, Some(deadline))?;
329        driver.send(&payload).map_err(status_only)?;
330        driver.half_close().map_err(status_only)?;
331        let mut response = None;
332        loop {
333            match driver.try_recv_next() {
334                Ok(Some(bytes)) => {
335                    if response.replace(bytes).is_some() {
336                        driver.cancel();
337                        return Err(Status::internal("multiple unary responses"));
338                    }
339                }
340                Ok(None) => break,
341                Err(StreamError::WouldBlock) => driver.wait_ready().map_err(status_only)?,
342                Err(error) => return Err(status_only(error)),
343            }
344        }
345        let bytes = response.ok_or_else(|| Status::internal("missing unary response"))?;
346        let response = codec.decode(bytes.payload()).map_err(codec_status)?;
347        check_deadline(deadline)?;
348        Ok((
349            response,
350            driver.initial.take().unwrap_or_default(),
351            driver.trailing.take().unwrap_or_default(),
352        ))
353    }
354    /// Performs a unary call with default call options.
355    /// # Errors
356    /// Returns the same failures as [`Self::unary`].
357    pub fn call_unary<C, Req, Resp>(
358        &mut self,
359        path: &str,
360        req: &Req,
361        codec: &mut C,
362    ) -> Result<Resp, Status>
363    where
364        C: Codec<EncodeItem = Req, DecodeItem = Resp>,
365    {
366        self.unary(path, req, codec, CallOptions::default())
367            .map(|(response, _, _)| response)
368    }
369    /// Starts one request followed by a response stream.
370    /// # Errors
371    /// Returns setup, encoding or request submission failures.
372    pub fn server_streaming<C, Req, Resp>(
373        &mut self,
374        path: &str,
375        req: &Req,
376        codec: C,
377        options: CallOptions,
378    ) -> Result<ServerStreamingCall<Resp>, Status>
379    where
380        C: Codec<EncodeItem = Req, DecodeItem = Resp> + 'static,
381        Req: 'static,
382        Resp: 'static,
383    {
384        let mut driver = self.start(path, codec, options, C::CONTENT_TYPE, None)?;
385        driver.send(req).map_err(status_only)?;
386        driver.half_close().map_err(status_only)?;
387        Ok(ServerStreamingCall::new(driver))
388    }
389    /// Starts a request stream with one terminal response.
390    /// # Errors
391    /// Returns invalid options or session submission failures.
392    pub fn client_streaming<C, Req, Resp>(
393        &mut self,
394        path: &str,
395        codec: C,
396        options: CallOptions,
397    ) -> Result<ClientStreamingCall<Req, Resp>, Status>
398    where
399        C: Codec<EncodeItem = Req, DecodeItem = Resp> + 'static,
400        Req: 'static,
401        Resp: 'static,
402    {
403        Ok(ClientStreamingCall::new(self.start(
404            path,
405            codec,
406            options,
407            C::CONTENT_TYPE,
408            None,
409        )?))
410    }
411    /// Starts independent request and response streams.
412    /// # Errors
413    /// Returns invalid options or session submission failures.
414    pub fn bidi_streaming<C, Req, Resp>(
415        &mut self,
416        path: &str,
417        codec: C,
418        options: CallOptions,
419    ) -> Result<BidiStreamingCall<Req, Resp>, Status>
420    where
421        C: Codec<EncodeItem = Req, DecodeItem = Resp> + 'static,
422        Req: 'static,
423        Resp: 'static,
424    {
425        Ok(BidiStreamingCall::new(self.start(
426            path,
427            codec,
428            options,
429            C::CONTENT_TYPE,
430            None,
431        )?))
432    }
433}
434impl SyncChannel<TcpTransport> {
435    /// Connects a TCP channel with authority derived from the peer address.
436    /// # Errors
437    /// Returns connection, address or session initialization errors.
438    pub fn connect(address: impl std::net::ToSocketAddrs) -> Result<Self, Status> {
439        let transport = TcpTransport::connect(address).map_err(io_status)?;
440        let authority = transport.peer_endpoint().map_err(io_status)?;
441        Self::with_config(
442            transport,
443            ChannelConfig {
444                authority,
445                ..ChannelConfig::default()
446            },
447        )
448    }
449    /// Connects and verifies an HTTP/2 TLS session before creating the channel.
450    /// # Errors
451    /// Returns invalid server name, TCP, certificate, ALPN or handshake errors.
452    pub fn connect_tls(
453        address: impl std::net::ToSocketAddrs,
454        config: &grpc_core::tls::TlsClientConfig,
455        server_name: &str,
456    ) -> Result<SyncChannel<impl SyncTransport + 'static>, Status> {
457        let name = server_name.to_owned().try_into().map_err(codec_status)?;
458        let socket = TcpTransport::connect(address).map_err(io_status)?;
459        let mut tls =
460            grpc_core::tls::TlsTransport::client(socket, config, name).map_err(io_status)?;
461        tls.handshake().map_err(io_status)?;
462        SyncChannel::with_config(
463            tls,
464            ChannelConfig {
465                authority: server_name.into(),
466                scheme: "https".into(),
467                ..ChannelConfig::default()
468            },
469        )
470    }
471}
472#[cfg(unix)]
473impl SyncChannel<grpc_core::transport::UnixTransport> {
474    /// Connects a Unix-domain channel.
475    /// # Errors
476    /// Returns socket connection or session initialization errors.
477    pub fn connect_unix(path: impl AsRef<std::path::Path>) -> Result<Self, Status> {
478        Self::new(grpc_core::transport::UnixTransport::connect(path).map_err(io_status)?)
479    }
480}
481fn status_only(error: StreamError) -> Status {
482    match error {
483        StreamError::Status(status) => status,
484        StreamError::WouldBlock => Status::unavailable("RPC operation would block"),
485    }
486}
487struct BytesCodec;
488impl Codec for BytesCodec {
489    type EncodeItem = Vec<u8>;
490    type DecodeItem = grpc_core::AlignedMessageBuf;
491    type Error = Status;
492    const CONTENT_TYPE: &'static str = "application/grpc";
493    fn encode(&mut self, item: &Vec<u8>, destination: &mut Vec<u8>) -> Result<(), Self::Error> {
494        destination.extend_from_slice(item);
495        Ok(())
496    }
497    fn decode(&mut self, payload: &[u8]) -> Result<Self::DecodeItem, Self::Error> {
498        grpc_core::AlignedMessageBuf::new(payload, false)
499    }
500}
501/// Typed streaming driver over one shared caller-thread channel connection.
502pub struct ChannelStreamDriver<T: SyncTransport, C: Codec> {
503    inner: Rc<RefCell<ChannelInner<T>>>,
504    id: i32,
505    codec: C,
506    encoding: CompressionAlgorithm,
507    deadline: Option<Instant>,
508    initial: Option<Metadata>,
509    trailing: Option<Metadata>,
510    complete: bool,
511    closed: bool,
512    buffer: Vec<u8>,
513    pending_cancellations: Rc<RefCell<Vec<i32>>>,
514}
515impl<T: SyncTransport, C: Codec> ChannelStreamDriver<T, C> {
516    fn check_deadline(&self) -> Result<(), StreamError> {
517        check_deadline(self.deadline).map_err(Into::into)
518    }
519    fn receive(&mut self) -> Result<Option<C::DecodeItem>, StreamError> {
520        let mut inner = self
521            .inner
522            .try_borrow_mut()
523            .map_err(|_| Status::failed_precondition("reentrant channel access"))?;
524        if self.initial.is_none() {
525            self.initial = inner.session.pop_initial_metadata(self.id)?;
526        }
527        if let Some(message) = inner.session.pop_message(self.id)? {
528            drop(inner);
529            let value = self.codec.decode(message.payload()).map_err(codec_status)?;
530            self.check_deadline()?;
531            return Ok(Some(value));
532        }
533        if let Some((status, trailers)) = inner.session.take_trailers(self.id)? {
534            self.trailing = Some(trailers);
535            if !self.closed {
536                inner.session.cancel_stream(self.id, 0)?;
537                self.closed = true;
538            }
539            inner.pending_reap.insert(self.id);
540            inner.flush()?;
541            self.complete = true;
542            return if status.code() == Code::Ok {
543                Ok(None)
544            } else {
545                Err(status.into())
546            };
547        }
548        Err(StreamError::WouldBlock)
549    }
550}
551impl<T: SyncTransport, C: Codec> StreamDriver for ChannelStreamDriver<T, C>
552where
553    C::EncodeItem: Sized,
554{
555    type Send = C::EncodeItem;
556    type Receive = C::DecodeItem;
557    fn send(&mut self, message: &Self::Send) -> Result<(), StreamError> {
558        self.check_deadline()?;
559        if self.closed || self.complete {
560            return Err(Status::failed_precondition("request sending half is closed").into());
561        }
562        self.buffer.clear();
563        self.codec
564            .encode(message, &mut self.buffer)
565            .map_err(codec_status)?;
566        self.check_deadline()?;
567        let mut inner = self
568            .inner
569            .try_borrow_mut()
570            .map_err(|_| Status::failed_precondition("reentrant channel access"))?;
571        inner
572            .session
573            .send_message(self.id, &self.buffer, self.encoding)?;
574        inner.flush()?;
575        Ok(())
576    }
577    fn half_close(&mut self) -> Result<(), StreamError> {
578        self.check_deadline()?;
579        if self.closed {
580            return Ok(());
581        }
582        let mut inner = self
583            .inner
584            .try_borrow_mut()
585            .map_err(|_| Status::failed_precondition("reentrant channel access"))?;
586        inner.session.half_close(self.id)?;
587        self.closed = true;
588        inner.flush()?;
589        Ok(())
590    }
591    fn try_recv_next(&mut self) -> Result<Option<Self::Receive>, StreamError> {
592        self.check_deadline()?;
593        if self.complete {
594            return Ok(None);
595        }
596        match self.receive() {
597            Err(StreamError::WouldBlock) => {}
598            result => return result,
599        }
600        self.inner
601            .try_borrow_mut()
602            .map_err(|_| Status::failed_precondition("reentrant channel access"))?
603            .pump(false, self.deadline)?;
604        self.receive()
605    }
606    fn wait_ready(&mut self) -> Result<(), StreamError> {
607        self.check_deadline()?;
608        self.inner
609            .try_borrow_mut()
610            .map_err(|_| Status::failed_precondition("reentrant channel access"))?
611            .pump(true, self.deadline)?;
612        Ok(())
613    }
614    fn finish(&mut self, _: &Status, _: &Metadata) -> Result<(), StreamError> {
615        Err(Status::failed_precondition("client cannot finish server response").into())
616    }
617    fn initial_metadata(&self) -> Option<&Metadata> {
618        self.initial.as_ref()
619    }
620    fn trailing_metadata(&self) -> Option<&Metadata> {
621        self.trailing.as_ref()
622    }
623    fn cancel(&mut self) {
624        if self.complete {
625            return;
626        }
627        self.pending_cancellations.borrow_mut().push(self.id);
628        if let Ok(mut inner) = self.inner.try_borrow_mut()
629            && let Err(error) = inner.flush()
630        {
631            inner.cleanup_error = Some(error);
632        }
633        self.complete = true;
634    }
635}
636
637impl<T: SyncTransport, C: Codec> Drop for ChannelStreamDriver<T, C> {
638    fn drop(&mut self) {
639        if self.complete {
640            return;
641        }
642        self.pending_cancellations.borrow_mut().push(self.id);
643        if let Ok(mut inner) = self.inner.try_borrow_mut()
644            && let Err(error) = inner.flush()
645        {
646            inner.cleanup_error = Some(error);
647        }
648    }
649}
650fn check_deadline(deadline: Option<Instant>) -> Result<(), Status> {
651    if deadline.is_some_and(|at| Instant::now() >= at) {
652        Err(Status::deadline_exceeded("RPC deadline elapsed"))
653    } else {
654        Ok(())
655    }
656}
grpc/channel.rs

grpc/
channel.rs
