mwc_libp2p_core/transport/
timeout.rs

1// Copyright 2018 Parity Technologies (UK) Ltd.
2//
3// Permission is hereby granted, free of charge, to any person obtaining a
4// copy of this software and associated documentation files (the "Software"),
5// to deal in the Software without restriction, including without limitation
6// the rights to use, copy, modify, merge, publish, distribute, sublicense,
7// and/or sell copies of the Software, and to permit persons to whom the
8// Software is furnished to do so, subject to the following conditions:
9//
10// The above copyright notice and this permission notice shall be included in
11// all copies or substantial portions of the Software.
12//
13// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
14// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
15// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
16// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
17// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
18// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
19// DEALINGS IN THE SOFTWARE.
20
21//! Transports with timeouts on the connection setup.
22//!
23//! The connection setup includes all protocol upgrades applied on the
24//! underlying `Transport`.
25// TODO: add example
26
27use crate::{Multiaddr, Transport, transport::{TransportError, ListenerEvent}};
28use futures::prelude::*;
29use futures_timer::Delay;
30use std::{error, fmt, io, pin::Pin, task::Context, task::Poll, time::Duration};
31
32/// A `TransportTimeout` is a `Transport` that wraps another `Transport` and adds
33/// timeouts to all inbound and outbound connection attempts.
34///
35/// **Note**: `listen_on` is never subject to a timeout, only the setup of each
36/// individual accepted connection.
37#[derive(Debug, Copy, Clone)]
38pub struct TransportTimeout<InnerTrans> {
39    inner: InnerTrans,
40    outgoing_timeout: Duration,
41    incoming_timeout: Duration,
42}
43
44impl<InnerTrans> TransportTimeout<InnerTrans> {
45    /// Wraps around a `Transport` to add timeouts to all the sockets created by it.
46    pub fn new(trans: InnerTrans, timeout: Duration) -> Self {
47        TransportTimeout {
48            inner: trans,
49            outgoing_timeout: timeout,
50            incoming_timeout: timeout,
51        }
52    }
53
54    /// Wraps around a `Transport` to add timeouts to the outgoing connections.
55    pub fn with_outgoing_timeout(trans: InnerTrans, timeout: Duration) -> Self {
56        TransportTimeout {
57            inner: trans,
58            outgoing_timeout: timeout,
59            incoming_timeout: Duration::from_secs(100 * 365 * 24 * 3600), // 100 years
60        }
61    }
62
63    /// Wraps around a `Transport` to add timeouts to the ingoing connections.
64    pub fn with_ingoing_timeout(trans: InnerTrans, timeout: Duration) -> Self {
65        TransportTimeout {
66            inner: trans,
67            outgoing_timeout: Duration::from_secs(100 * 365 * 24 * 3600), // 100 years
68            incoming_timeout: timeout,
69        }
70    }
71}
72
73impl<InnerTrans> Transport for TransportTimeout<InnerTrans>
74where
75    InnerTrans: Transport,
76    InnerTrans::Error: 'static,
77{
78    type Output = InnerTrans::Output;
79    type Error = TransportTimeoutError<InnerTrans::Error>;
80    type Listener = TimeoutListener<InnerTrans::Listener>;
81    type ListenerUpgrade = Timeout<InnerTrans::ListenerUpgrade>;
82    type Dial = Timeout<InnerTrans::Dial>;
83
84    fn listen_on(self, addr: Multiaddr) -> Result<Self::Listener, TransportError<Self::Error>> {
85        let listener = self.inner.listen_on(addr)
86            .map_err(|err| err.map(TransportTimeoutError::Other))?;
87
88        let listener = TimeoutListener {
89            inner: listener,
90            timeout: self.incoming_timeout,
91        };
92
93        Ok(listener)
94    }
95
96    fn dial(self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
97        let dial = self.inner.dial(addr)
98            .map_err(|err| err.map(TransportTimeoutError::Other))?;
99        Ok(Timeout {
100            inner: dial,
101            timer: Delay::new(self.outgoing_timeout),
102        })
103    }
104
105    fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
106        self.inner.address_translation(server, observed)
107    }
108}
109
110// TODO: can be removed and replaced with an `impl Stream` once impl Trait is fully stable
111//       in Rust (https://github.com/rust-lang/rust/issues/34511)
112#[pin_project::pin_project]
113pub struct TimeoutListener<InnerStream> {
114    #[pin]
115    inner: InnerStream,
116    timeout: Duration,
117}
118
119impl<InnerStream, O, E> Stream for TimeoutListener<InnerStream>
120where
121    InnerStream: TryStream<Ok = ListenerEvent<O, E>, Error = E>,
122{
123    type Item = Result<ListenerEvent<Timeout<O>, TransportTimeoutError<E>>, TransportTimeoutError<E>>;
124
125    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
126        let this = self.project();
127
128        let poll_out = match TryStream::try_poll_next(this.inner, cx) {
129            Poll::Ready(Some(Err(err))) => return Poll::Ready(Some(Err(TransportTimeoutError::Other(err)))),
130            Poll::Ready(Some(Ok(v))) => v,
131            Poll::Ready(None) => return Poll::Ready(None),
132            Poll::Pending => return Poll::Pending,
133        };
134
135        let timeout = *this.timeout;
136        let event = poll_out
137            .map(move |inner_fut| {
138                Timeout {
139                    inner: inner_fut,
140                    timer: Delay::new(timeout),
141                }
142            })
143            .map_err(TransportTimeoutError::Other);
144
145        Poll::Ready(Some(Ok(event)))
146    }
147}
148
149/// Wraps around a `Future`. Turns the error type from `TimeoutError<Err>` to
150/// `TransportTimeoutError<Err>`.
151// TODO: can be replaced with `impl Future` once `impl Trait` are fully stable in Rust
152//       (https://github.com/rust-lang/rust/issues/34511)
153#[pin_project::pin_project]
154#[must_use = "futures do nothing unless polled"]
155pub struct Timeout<InnerFut> {
156    #[pin]
157    inner: InnerFut,
158    timer: Delay,
159}
160
161impl<InnerFut> Future for Timeout<InnerFut>
162where
163    InnerFut: TryFuture,
164{
165    type Output = Result<InnerFut::Ok, TransportTimeoutError<InnerFut::Error>>;
166
167    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
168        // It is debatable whether we should poll the inner future first or the timer first.
169        // For example, if you start dialing with a timeout of 10 seconds, then after 15 seconds
170        // the dialing succeeds on the wire, then after 20 seconds you poll, then depending on
171        // which gets polled first, the outcome will be success or failure.
172
173        let mut this = self.project();
174
175        match TryFuture::try_poll(this.inner, cx) {
176            Poll::Pending => {},
177            Poll::Ready(Ok(v)) => return Poll::Ready(Ok(v)),
178            Poll::Ready(Err(err)) => return Poll::Ready(Err(TransportTimeoutError::Other(err))),
179        }
180
181        match Pin::new(&mut this.timer).poll(cx) {
182            Poll::Pending => Poll::Pending,
183            Poll::Ready(()) => Poll::Ready(Err(TransportTimeoutError::Timeout))
184        }
185    }
186}
187
188/// Error that can be produced by the `TransportTimeout` layer.
189#[derive(Debug)]
190pub enum TransportTimeoutError<TErr> {
191    /// The transport timed out.
192    Timeout,
193    /// An error happened in the timer.
194    TimerError(io::Error),
195    /// Other kind of error.
196    Other(TErr),
197}
198
199impl<TErr> fmt::Display for TransportTimeoutError<TErr>
200where TErr: fmt::Display,
201{
202    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
203        match self {
204            TransportTimeoutError::Timeout => write!(f, "Timeout has been reached"),
205            TransportTimeoutError::TimerError(err) => write!(f, "Error in the timer: {}", err),
206            TransportTimeoutError::Other(err) => write!(f, "{}", err),
207        }
208    }
209}
210
211impl<TErr> error::Error for TransportTimeoutError<TErr>
212where TErr: error::Error + 'static,
213{
214    fn source(&self) -> Option<&(dyn error::Error + 'static)> {
215        match self {
216            TransportTimeoutError::Timeout => None,
217            TransportTimeoutError::TimerError(err) => Some(err),
218            TransportTimeoutError::Other(err) => Some(err),
219        }
220    }
221}
mwc_libp2p_core/transport/timeout.rs

mwc_libp2p_core/transport/
timeout.rs
