tonic/transport/channel/
mod.rs

1/*
2 *
3 * Copyright 2025 gRPC authors.
4 *
5 * Permission is hereby granted, free of charge, to any person obtaining a copy
6 * of this software and associated documentation files (the "Software"), to
7 * deal in the Software without restriction, including without limitation the
8 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
9 * sell copies of the Software, and to permit persons to whom the Software is
10 * furnished to do so, subject to the following conditions:
11 *
12 * The above copyright notice and this permission notice shall be included in
13 * all copies or substantial portions of the Software.
14 *
15 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
20 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
21 * IN THE SOFTWARE.
22 *
23 */
24
25//! Client implementation and builder.
26
27mod endpoint;
28pub(crate) mod service;
29#[cfg(feature = "_tls-any")]
30mod tls;
31mod uds_connector;
32
33pub use self::service::Change;
34pub use endpoint::Endpoint;
35#[cfg(feature = "_tls-any")]
36pub use tls::ClientTlsConfig;
37
38use self::service::{Connection, DynamicServiceStream, Executor, SharedExec};
39use crate::body::Body;
40use bytes::Bytes;
41use http::{
42    Request, Response,
43    uri::{InvalidUri, Uri},
44};
45use std::{
46    fmt,
47    future::Future,
48    hash::Hash,
49    pin::Pin,
50    task::{Context, Poll},
51};
52use tokio::sync::mpsc::{Sender, channel};
53
54use hyper::rt;
55use tower::balance::p2c::Balance;
56use tower::{
57    Service,
58    buffer::{Buffer, future::ResponseFuture as BufferResponseFuture},
59    discover::Discover,
60    util::BoxService,
61};
62
63type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
64
65const DEFAULT_BUFFER_SIZE: usize = 1024;
66
67/// A default batteries included `transport` channel.
68///
69/// This provides a fully featured http2 gRPC client based on `hyper`
70/// and `tower` services.
71///
72/// # Multiplexing requests
73///
74/// Sending a request on a channel requires a `&mut self` and thus can only send
75/// one request in flight. This is intentional and is required to follow the `Service`
76/// contract from the `tower` library which this channel implementation is built on
77/// top of.
78///
79/// `tower` itself has a concept of `poll_ready` which is the main mechanism to apply
80/// back pressure. `poll_ready` takes a `&mut self` and when it returns `Poll::Ready`
81/// we know the `Service` is able to accept only one request before we must `poll_ready`
82/// again. Due to this fact any `async fn` that wants to poll for readiness and submit
83/// the request must have a `&mut self` reference.
84///
85/// To work around this and to ease the use of the channel, `Channel` provides a
86/// `Clone` implementation that is _cheap_. This is because at the very top level
87/// the channel is backed by a `tower_buffer::Buffer` which runs the connection
88/// in a background task and provides a `mpsc` channel interface. Due to this
89/// cloning the `Channel` type is cheap and encouraged.
90#[derive(Clone)]
91pub struct Channel {
92    svc: Buffer<Request<Body>, BoxFuture<'static, Result<Response<Body>, crate::BoxError>>>,
93}
94
95/// A future that resolves to an HTTP response.
96///
97/// This is returned by the `Service::call` on [`Channel`].
98pub struct ResponseFuture {
99    inner: BufferResponseFuture<BoxFuture<'static, Result<Response<Body>, crate::BoxError>>>,
100}
101
102impl Channel {
103    /// Create an [`Endpoint`] builder that can create [`Channel`]s.
104    pub fn builder(uri: Uri) -> Endpoint {
105        Endpoint::from(uri)
106    }
107
108    /// Create an [`Endpoint`] from a static string.
109    ///
110    /// ```
111    /// # use tonic::transport::Channel;
112    /// Channel::from_static("https://example.com");
113    /// ```
114    pub fn from_static(s: &'static str) -> Endpoint {
115        let uri = Uri::from_static(s);
116        Self::builder(uri)
117    }
118
119    /// Create an [`Endpoint`] from shared bytes.
120    ///
121    /// ```
122    /// # use tonic::transport::Channel;
123    /// Channel::from_shared("https://example.com");
124    /// ```
125    pub fn from_shared(s: impl Into<Bytes>) -> Result<Endpoint, InvalidUri> {
126        let uri = Uri::from_maybe_shared(s.into())?;
127        Ok(Self::builder(uri))
128    }
129
130    /// Balance a list of [`Endpoint`]'s.
131    ///
132    /// This creates a [`Channel`] that will load balance across all the
133    /// provided endpoints.
134    pub fn balance_list(list: impl Iterator<Item = Endpoint>) -> Self {
135        let (channel, tx) = Self::balance_channel(DEFAULT_BUFFER_SIZE);
136        list.for_each(|endpoint| {
137            tx.try_send(Change::Insert(endpoint.uri.clone(), endpoint))
138                .unwrap();
139        });
140
141        channel
142    }
143
144    /// Balance a list of [`Endpoint`]'s.
145    ///
146    /// This creates a [`Channel`] that will listen to a stream of change events and will add or remove provided endpoints.
147    pub fn balance_channel<K>(capacity: usize) -> (Self, Sender<Change<K, Endpoint>>)
148    where
149        K: Hash + Eq + Send + Clone + 'static,
150    {
151        Self::balance_channel_with_executor(capacity, SharedExec::tokio())
152    }
153
154    /// Balance a list of [`Endpoint`]'s.
155    ///
156    /// This creates a [`Channel`] that will listen to a stream of change events and will add or remove provided endpoints.
157    ///
158    /// The [`Channel`] will use the given executor to spawn async tasks.
159    pub fn balance_channel_with_executor<K, E>(
160        capacity: usize,
161        executor: E,
162    ) -> (Self, Sender<Change<K, Endpoint>>)
163    where
164        K: Hash + Eq + Send + Clone + 'static,
165        E: Executor<Pin<Box<dyn Future<Output = ()> + Send>>> + Send + Sync + 'static,
166    {
167        let (tx, rx) = channel(capacity);
168        let list = DynamicServiceStream::new(rx);
169        (Self::balance(list, DEFAULT_BUFFER_SIZE, executor), tx)
170    }
171
172    /// Create a new [`Channel`] using a custom connector to the provided [Endpoint].
173    ///
174    /// This is a lower level API, prefer to use [`Endpoint::connect_lazy`] if you are not using a custom connector.
175    pub fn new<C>(connector: C, endpoint: Endpoint) -> Self
176    where
177        C: Service<Uri> + Send + 'static,
178        C::Error: Into<crate::BoxError> + Send,
179        C::Future: Send,
180        C::Response: rt::Read + rt::Write + Unpin + Send + 'static,
181    {
182        let buffer_size = endpoint.buffer_size.unwrap_or(DEFAULT_BUFFER_SIZE);
183        let executor = endpoint.executor.clone();
184
185        let svc = Connection::lazy(connector, endpoint);
186        let (svc, worker) = Buffer::pair(svc, buffer_size);
187
188        executor.execute(worker);
189
190        Channel { svc }
191    }
192
193    /// Connect to the provided [`Endpoint`] using the provided connector, and return a new [`Channel`].
194    ///
195    /// This is a lower level API, prefer to use [`Endpoint::connect`] if you are not using a custom connector.
196    pub async fn connect<C>(connector: C, endpoint: Endpoint) -> Result<Self, super::Error>
197    where
198        C: Service<Uri> + Send + 'static,
199        C::Error: Into<crate::BoxError> + Send,
200        C::Future: Unpin + Send,
201        C::Response: rt::Read + rt::Write + Unpin + Send + 'static,
202    {
203        let buffer_size = endpoint.buffer_size.unwrap_or(DEFAULT_BUFFER_SIZE);
204        let executor = endpoint.executor.clone();
205
206        let svc = Connection::connect(connector, endpoint)
207            .await
208            .map_err(super::Error::from_source)?;
209        let (svc, worker) = Buffer::pair(svc, buffer_size);
210        executor.execute(worker);
211
212        Ok(Channel { svc })
213    }
214
215    pub(crate) fn balance<D, E>(discover: D, buffer_size: usize, executor: E) -> Self
216    where
217        D: Discover<Service = Connection> + Unpin + Send + 'static,
218        D::Error: Into<crate::BoxError>,
219        D::Key: Hash + Send + Clone,
220        E: Executor<BoxFuture<'static, ()>> + Send + Sync + 'static,
221    {
222        let svc = Balance::new(discover);
223
224        let svc = BoxService::new(svc);
225        let (svc, worker) = Buffer::pair(svc, buffer_size);
226        executor.execute(Box::pin(worker));
227
228        Channel { svc }
229    }
230}
231
232impl Service<http::Request<Body>> for Channel {
233    type Response = http::Response<Body>;
234    type Error = super::Error;
235    type Future = ResponseFuture;
236
237    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
238        Service::poll_ready(&mut self.svc, cx).map_err(super::Error::from_source)
239    }
240
241    fn call(&mut self, request: http::Request<Body>) -> Self::Future {
242        let inner = Service::call(&mut self.svc, request);
243
244        ResponseFuture { inner }
245    }
246}
247
248impl Future for ResponseFuture {
249    type Output = Result<Response<Body>, super::Error>;
250
251    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
252        Pin::new(&mut self.inner)
253            .poll(cx)
254            .map_err(super::Error::from_source)
255    }
256}
257
258impl fmt::Debug for Channel {
259    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
260        f.debug_struct("Channel").finish()
261    }
262}
263
264impl fmt::Debug for ResponseFuture {
265    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
266        f.debug_struct("ResponseFuture").finish()
267    }
268}
tonic/transport/channel/mod.rs

tonic/transport/channel/
mod.rs
