Repository: tokio-rs/tokio
File: tokio-stream/src/stream_ext.rs (Rust)

Code before:
    /// # let mut int_stream = stream::iter(vec![Ok(1), Err(()), Ok(2), Ok(3)]);
    /// assert_eq!(int_stream.try_next().await, Ok(Some(1)));
    /// assert!(int_stream.try_next().await.is_err());
    /// assert_eq!(int_stream.try_next().await, Ok(Some(2)));
    /// assert_eq!(int_stream.try_next().await, Ok(Some(3)));
    /// assert_eq!(int_stream.try_next().await, Ok(None));
    ///
    /// // If we want to stop consuming the source stream the first time an
    /// // element times out, we can use the `take_while` operator:
    /// # let int_stream = stream::iter(vec![Ok(1), Err(()), Ok(2), Ok(3)]);
    /// let mut int_stream = int_stream.take_while(Result::is_ok);
    ///
    /// assert_eq!(int_stream.try_next().await, Ok(Some(1)));
    /// assert_eq!(int_stream.try_next().await, Ok(None));
    ///
    /// // Timeout errors will be continuously produced at the specified
    /// // interval until the wrapped stream yields a value.
    /// let interval_stream = IntervalStream::new(tokio::time::interval(Duration::from_millis(23)));
    /// let timeout_stream = interval_stream.timeout_repeating(Duration::from_millis(9));
    /// tokio::pin!(timeout_stream);
    ///
    /// // Multiple timeouts will be received between values in the source stream.
    /// assert!(timeout_stream.try_next().await.is_ok());
    /// assert!(timeout_stream.try_next().await.is_err());
    /// assert!(timeout_stream.try_next().await.is_err());
    /// assert!(timeout_stream.try_next().await.is_ok());
    /// # }
    /// ```
    #[cfg(all(feature = "time"))]
    #[cfg_attr(docsrs, doc(cfg(feature = "time")))]
    fn timeout_repeating(self, duration: Duration) -> Timeout<Self>
    where
        Self: Sized,
    {
        Timeout::new(self, duration, TimeoutMode::Repeating)
    }

    /// Slows down a stream by enforcing a delay between items.
    ///
    /// The underlying timer behind this utility has a granularity of one millisecond.
    ///
    /// # Example
    ///
    /// Create a throttled stream.
    /// ```rust,no_run
    /// use std::time::Duration;
    /// use tokio_stream::StreamExt;
    ///
    /// # async fn dox() {
    /// let item_stream = futures::stream::repeat("one").throttle(Duration::from_secs(2));
    /// tokio::pin!(item_stream);

Review comment (bug):
I was afraid this test would be too timing-sensitive and indeed it is failing in CI. Any suggestions on a different approach would be appreciated. I don't know a compact way to create a stream that will produce a few messages and then pause for a while.

Code after:
    ///
    /// let int_stream = int_stream.timeout_repeating(tokio::time::interval(Duration::from_secs(1)));
    /// tokio::pin!(int_stream);
    ///
    /// // When no items time out, we get the 3 elements in succession:
    /// assert_eq!(int_stream.try_next().await, Ok(Some(1)));
    /// assert_eq!(int_stream.try_next().await, Ok(Some(2)));
    /// assert_eq!(int_stream.try_next().await, Ok(Some(3)));
    /// assert_eq!(int_stream.try_next().await, Ok(None));
    ///
    /// // If the second item times out, we get an error and continue polling the stream:
    /// # let mut int_stream = stream::iter(vec![Ok(1), Err(()), Ok(2), Ok(3)]);
    /// assert_eq!(int_stream.try_next().await, Ok(Some(1)));
    /// assert!(int_stream.try_next().await.is_err());
    /// assert_eq!(int_stream.try_next().await, Ok(Some(2)));
    /// assert_eq!(int_stream.try_next().await, Ok(Some(3)));
    /// assert_eq!(int_stream.try_next().await, Ok(None));
    ///
    /// // If we want to stop consuming the source stream the first time an
    /// // element times out, we can use the `take_while` operator:
    /// # let int_stream = stream::iter(vec![Ok(1), Err(()), Ok(2), Ok(3)]);
    /// let mut int_stream = int_stream.take_while(Result::is_ok);
    ///
    /// assert_eq!(int_stream.try_next().await, Ok(Some(1)));
    /// assert_eq!(int_stream.try_next().await, Ok(None));
    /// # }
    /// ```
    ///
    /// Timeout errors will be continuously produced at the specified interval
    /// until the wrapped stream yields a value.
    ///
    /// ```
    /// # #[tokio::main(flavor = "current_thread", start_paused = true)]
    /// # async fn main() {
    /// use tokio_stream::{StreamExt, wrappers::IntervalStream};
    /// use std::time::Duration;
    /// let interval_stream = IntervalStream::new(tokio::time::interval(Duration::from_millis(23)));
    /// let timeout_stream = interval_stream.timeout_repeating(tokio::time::interval(Duration::from_millis(9)));
    /// tokio::pin!(timeout_stream);
    ///
    /// // Multiple timeouts will be received between values in the source stream.
    /// assert!(timeout_stream.try_next().await.is_ok());
    /// assert!(timeout_stream.try_next().await.is_err(), "expected one timeout");
    /// assert!(timeout_stream.try_next().await.is_err(), "expected a second timeout");
    /// // Will eventually receive another value from the source stream...
    /// assert!(timeout_stream.try_next().await.is_ok(), "expected non-timeout");
    /// # }
    /// ```
    #[cfg(all(feature = "time"))]
    #[cfg_attr(docsrs, doc(cfg(feature = "time")))]
    fn timeout_repeating(self, interval: Interval) -> TimeoutRepeating<Self>