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

Code before:
    ///
    /// # async fn dox() {
    /// let item_stream = futures::stream::repeat("one").throttle(Duration::from_secs(2));
    /// tokio::pin!(item_stream);
    ///
    /// loop {
    ///     // The string will be produced at most every 2 seconds
    ///     println!("{:?}", item_stream.next().await);
    /// }
    /// # }
    /// ```
    #[cfg(all(feature = "time"))]
    #[cfg_attr(docsrs, doc(cfg(feature = "time")))]
    fn throttle(self, duration: Duration) -> Throttle<Self>
    where
        Self: Sized,
    {
        throttle(duration, self)
    }

    /// Collects items into batches inside a vector within a deadline.
    ///
    /// Attempts to yield a vector of len `capacity` within the deadline,
    /// otherwise will yield a vector of len less than `capacity` if the deadline is reached.
    /// If the wrapped stream does not yield a value before the deadline is reached, then an
    /// empty vector will be yielded. The deadline will be reset after every batch is yielded.
    ///
    /// # Example
    ///
    /// ```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

Review comment (performance):
If you're looking for inspiration for the documentation, then here is a proposal:

> 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 `capacity` many items since the last batch.
>  2. The time since the last batch is greater than the given duration.
> 
> The length of the returned vector is never greater than `capacity`. Note that the second condition will trigger even if the resulting batch is empty. In this case, the stream yields an empty vector when the duration has elapsed.

Also, I think `max_size` is a better name than `capacity`.

Code after:
    {
        Timeout::new(self, duration)
    }

    /// Slows down a stream by enforcing a delay between items.
    ///
    /// # 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);
    ///
    /// loop {
    ///     // The string will be produced at most every 2 seconds
    ///     println!("{:?}", item_stream.next().await);
    /// }
    /// # }
    /// ```
    #[cfg(all(feature = "time"))]
    #[cfg_attr(docsrs, doc(cfg(feature = "time")))]
    fn throttle(self, duration: Duration) -> Throttle<Self>
    where
        Self: Sized,
    {
        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;