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

Code before:
    ///
    /// ```rust,no_run
    /// use std::time::Duration;
    /// use tokio::time;
    /// use tokio_stream::{self as stream, StreamExt};
    /// use futures::FutureExt;
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let iter = vec![1, 2, 3].into_iter();
    ///     let stream0 = stream::iter(iter);
    ///
    ///     let iter = vec![4].into_iter();
    ///     let stream1 = stream::iter(iter)
    ///          .then(move |n| time::sleep(Duration::from_secs(5)).map(move |_| n));
    ///
    ///     let chunk_stream = stream0
    ///         .chain(stream1)
    ///         .chunks_timeout(3, Duration::from_secs(2));
    ///
    ///     // a full batch was received
    ///     assert_eq!(chunk_stream.next().await, Some(vec![1,2,3]));
    ///     // deadline was reached before capacity was reached
    ///     assert_eq!(chunk_stream.next().await, Some(vec![4]));
    ///     // no items were emitted within the deadline
    ///     assert_eq!(chunk_stream.next().await, Some(vec![]));
    ///     // deadline was reached before capacity was reached
    ///     assert_eq!(chunk_stream.next().await, Some(vec![5]));
    /// }
    /// ```
    #[cfg(feature = "time")]
    #[cfg_attr(docsrs, doc(cfg(feature = "time")))]
    fn chunks_timeout(self, capacity: usize, duration: Duration) -> ChunksTimeout<Self>
    where
        Self: Sized,
    {
        ChunksTimeout::new(self, capacity, duration)
    }
}

impl<St: ?Sized> StreamExt for St where St: Stream {}

/// Merge the size hints from two streams.
fn merge_size_hints(
    (left_low, left_high): (usize, Option<usize>),
    (right_low, right_hign): (usize, Option<usize>),
) -> (usize, Option<usize>) {
    let low = left_low.saturating_add(right_low);
    let high = match (left_high, right_hign) {
        (Some(h1), Some(h2)) => h1.checked_add(h2),
        _ => None,

Review comment (performance):
I was wondering something: maybe it would be more useful to change the stream so that the timer only starts counting down when the first item of the batch arrives? 

Code after:
    {
        throttle(duration, self)
    }

    /// Batches the items in the given stream using a maximum duration and size for each batch.
    ///
    /// This stream returns the next batch of items in the following situations:
    ///  1. The inner stream has returned at least `max_size` many items since the last batch.
    ///  2. The time since the first item of a batch is greater than the given duration.
    ///  3. The end of the stream is reached.
    ///
    /// The length of the returned vector is never empty or greater than the maximum size. Empty batches
    /// will not be emitted if no items are received upstream.
    ///
    /// # Panics
    ///
    /// This function panics if `max_size` is zero
    ///
    /// # Example
    ///
    /// ```rust
    /// use std::time::Duration;
    /// use tokio::time;
    /// use tokio_stream::{self as stream, StreamExt};
    /// use futures::FutureExt;
    ///
    /// #[tokio::main]
    /// # async fn _unused() {}
    /// # #[tokio::main(flavor = "current_thread", start_paused = true)]
    /// async fn main() {
    ///     let iter = vec![1, 2, 3, 4].into_iter();
    ///     let stream0 = stream::iter(iter);
    ///
    ///     let iter = vec![5].into_iter();
    ///     let stream1 = stream::iter(iter)
    ///          .then(move |n| time::sleep(Duration::from_secs(5)).map(move |_| n));
    ///
    ///     let chunk_stream = stream0
    ///         .chain(stream1)
    ///         .chunks_timeout(3, Duration::from_secs(2));
    ///     tokio::pin!(chunk_stream);
    ///
    ///     // a full batch was received
    ///     assert_eq!(chunk_stream.next().await, Some(vec![1,2,3]));
    ///     // deadline was reached before max_size was reached
    ///     assert_eq!(chunk_stream.next().await, Some(vec![4]));
    ///     // last element in the stream
    ///     assert_eq!(chunk_stream.next().await, Some(vec![5]));
    /// }
    /// ```
    #[cfg(feature = "time")]