tonic/transport/server/
incoming.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
25use std::{
26    net::{SocketAddr, TcpListener as StdTcpListener},
27    pin::Pin,
28    task::{Context, Poll},
29    time::Duration,
30};
31
32use socket2::TcpKeepalive;
33use tokio::net::{TcpListener, TcpStream};
34use tokio_stream::{Stream, wrappers::TcpListenerStream};
35use tracing::warn;
36
37/// Binds a socket address for a [Router](super::Router)
38///
39/// An incoming stream, usable with [Router::serve_with_incoming](super::Router::serve_with_incoming),
40/// of `AsyncRead + AsyncWrite` that communicate with clients that connect to a socket address.
41#[derive(Debug)]
42pub struct TcpIncoming {
43    inner: TcpListenerStream,
44    nodelay: Option<bool>,
45    keepalive: Option<TcpKeepalive>,
46    keepalive_time: Option<Duration>,
47    keepalive_interval: Option<Duration>,
48    keepalive_retries: Option<u32>,
49}
50
51impl TcpIncoming {
52    /// Creates an instance by binding (opening) the specified socket address.
53    ///
54    /// Returns a TcpIncoming if the socket address was successfully bound.
55    ///
56    /// # Examples
57    /// ```no_run
58    /// # use tower_service::Service;
59    /// # use http::{request::Request, response::Response};
60    /// # use tonic::{body::Body, server::NamedService, transport::{Server, server::TcpIncoming}};
61    /// # use core::convert::Infallible;
62    /// # use std::error::Error;
63    /// # fn main() { }  // Cannot have type parameters, hence instead define:
64    /// # fn run<S>(some_service: S) -> Result<(), Box<dyn Error + Send + Sync>>
65    /// # where
66    /// #   S: Service<Request<Body>, Response = Response<Body>, Error = Infallible> + NamedService + Clone + Send + Sync + 'static,
67    /// #   S::Future: Send + 'static,
68    /// # {
69    /// // Find a free port
70    /// let mut port = 1322;
71    /// let tinc = loop {
72    ///    let addr = format!("127.0.0.1:{}", port).parse().unwrap();
73    ///    match TcpIncoming::bind(addr) {
74    ///       Ok(t) => break t,
75    ///       Err(_) => port += 1
76    ///    }
77    /// };
78    /// Server::builder()
79    ///    .add_service(some_service)
80    ///    .serve_with_incoming(tinc);
81    /// # Ok(())
82    /// # }
83    pub fn bind(addr: SocketAddr) -> std::io::Result<Self> {
84        let std_listener = StdTcpListener::bind(addr)?;
85        std_listener.set_nonblocking(true)?;
86
87        Ok(TcpListener::from_std(std_listener)?.into())
88    }
89
90    /// Sets the `TCP_NODELAY` option on the accepted connection.
91    pub fn with_nodelay(self, nodelay: Option<bool>) -> Self {
92        Self { nodelay, ..self }
93    }
94
95    /// Sets the `TCP_KEEPALIVE` option on the accepted connection.
96    pub fn with_keepalive(self, keepalive_time: Option<Duration>) -> Self {
97        Self {
98            keepalive_time,
99            keepalive: make_keepalive(
100                keepalive_time,
101                self.keepalive_interval,
102                self.keepalive_retries,
103            ),
104            ..self
105        }
106    }
107
108    /// Sets the `TCP_KEEPINTVL` option on the accepted connection.
109    pub fn with_keepalive_interval(self, keepalive_interval: Option<Duration>) -> Self {
110        Self {
111            keepalive_interval,
112            keepalive: make_keepalive(
113                self.keepalive_time,
114                keepalive_interval,
115                self.keepalive_retries,
116            ),
117            ..self
118        }
119    }
120
121    /// Sets the `TCP_KEEPCNT` option on the accepted connection.
122    pub fn with_keepalive_retries(self, keepalive_retries: Option<u32>) -> Self {
123        Self {
124            keepalive_retries,
125            keepalive: make_keepalive(
126                self.keepalive_time,
127                self.keepalive_interval,
128                keepalive_retries,
129            ),
130            ..self
131        }
132    }
133
134    /// Returns the local address that this tcp incoming is bound to.
135    pub fn local_addr(&self) -> std::io::Result<SocketAddr> {
136        self.inner.as_ref().local_addr()
137    }
138}
139
140impl From<TcpListener> for TcpIncoming {
141    fn from(listener: TcpListener) -> Self {
142        Self {
143            inner: TcpListenerStream::new(listener),
144            nodelay: None,
145            keepalive: None,
146            keepalive_time: None,
147            keepalive_interval: None,
148            keepalive_retries: None,
149        }
150    }
151}
152
153impl Stream for TcpIncoming {
154    type Item = std::io::Result<TcpStream>;
155
156    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
157        let polled = Pin::new(&mut self.inner).poll_next(cx);
158
159        if let Poll::Ready(Some(Ok(stream))) = &polled {
160            set_accepted_socket_options(stream, self.nodelay, &self.keepalive);
161        }
162
163        polled
164    }
165}
166
167// Consistent with hyper-0.14, this function does not return an error.
168fn set_accepted_socket_options(
169    stream: &TcpStream,
170    nodelay: Option<bool>,
171    keepalive: &Option<TcpKeepalive>,
172) {
173    if let Some(nodelay) = nodelay
174        && let Err(e) = stream.set_nodelay(nodelay)
175    {
176        warn!("error trying to set TCP_NODELAY: {e}");
177    }
178
179    if let Some(keepalive) = keepalive {
180        let sock_ref = socket2::SockRef::from(&stream);
181        if let Err(e) = sock_ref.set_tcp_keepalive(keepalive) {
182            warn!("error trying to set TCP_KEEPALIVE: {e}");
183        }
184    }
185}
186
187fn make_keepalive(
188    keepalive_time: Option<Duration>,
189    keepalive_interval: Option<Duration>,
190    keepalive_retries: Option<u32>,
191) -> Option<TcpKeepalive> {
192    let mut dirty = false;
193    let mut keepalive = TcpKeepalive::new();
194    if let Some(t) = keepalive_time {
195        keepalive = keepalive.with_time(t);
196        dirty = true;
197    }
198
199    #[cfg(
200        // See https://docs.rs/socket2/0.5.8/src/socket2/lib.rs.html#511-525
201        any(
202            target_os = "android",
203            target_os = "dragonfly",
204            target_os = "freebsd",
205            target_os = "fuchsia",
206            target_os = "illumos",
207            target_os = "ios",
208            target_os = "visionos",
209            target_os = "linux",
210            target_os = "macos",
211            target_os = "netbsd",
212            target_os = "tvos",
213            target_os = "watchos",
214            target_os = "windows",
215        )
216    )]
217    if let Some(t) = keepalive_interval {
218        keepalive = keepalive.with_interval(t);
219        dirty = true;
220    }
221
222    #[cfg(
223        // See https://docs.rs/socket2/0.5.8/src/socket2/lib.rs.html#557-570
224        any(
225            target_os = "android",
226            target_os = "dragonfly",
227            target_os = "freebsd",
228            target_os = "fuchsia",
229            target_os = "illumos",
230            target_os = "ios",
231            target_os = "visionos",
232            target_os = "linux",
233            target_os = "macos",
234            target_os = "netbsd",
235            target_os = "tvos",
236            target_os = "watchos",
237        )
238    )]
239    if let Some(r) = keepalive_retries {
240        keepalive = keepalive.with_retries(r);
241        dirty = true;
242    }
243
244    // avoid clippy errors for targets that do not use these fields.
245    let _ = keepalive_retries;
246    let _ = keepalive_interval;
247
248    dirty.then_some(keepalive)
249}
250
251#[cfg(test)]
252mod tests {
253    use crate::transport::server::TcpIncoming;
254    #[tokio::test]
255    async fn one_tcpincoming_at_a_time() {
256        let addr = "127.0.0.1:1322".parse().unwrap();
257        {
258            let _t1 = TcpIncoming::bind(addr).unwrap();
259            let _t2 = TcpIncoming::bind(addr).unwrap_err();
260        }
261        let _t3 = TcpIncoming::bind(addr).unwrap();
262    }
263}
tonic/transport/server/incoming.rs

tonic/transport/server/
incoming.rs
