Trait futures_time::stream::StreamExt

source · [−]
pub trait StreamExt: Stream {
    fn sample<I>(self, interval: I) -> Sample<Self, I::IntoStream>
    where
        Self: Sized,
        I: IntoStream,
    { ... }
    fn buffer<I>(self, interval: I) -> Buffer<Self, I::IntoStream>
    where
        Self: Sized,
        I: IntoStream,
    { ... }
    fn debounce<D>(self, window: D) -> Debounce<Self, D::IntoFuture>
    where
        Self: Sized,
        D: IntoFuture,
        D::IntoFuture: Timer,
    { ... }
    fn delay<D>(self, deadline: D) -> Delay<Self, D::IntoFuture>
    where
        Self: Sized,
        D: IntoFuture,
    { ... }
    fn park<I>(self, interval: I) -> Park<Self, I::IntoStream>
    where
        Self: Sized,
        I: IntoStream<Item = Parker>,
    { ... }
    fn throttle<I>(self, interval: I) -> Throttle<Self, I::IntoStream>
    where
        Self: Sized,
        I: IntoStream,
    { ... }
    fn timeout<D>(self, deadline: D) -> Timeout<Self, D::IntoFuture>
    where
        Self: Sized,
        D: IntoFuture,
        D::IntoFuture: Timer,
    { ... }
}
Expand description

Extend Stream with time-based operations.

Provided Methods

Yield the last item received at the end of each interval.

If no items have been received during an interval, the stream will not yield any items. In addition to using a time-based interval, this method can take any stream as a source. This enables throttling based on alternative event sources, such as variable-rate timers.

See also throttle() and debounce().

Data Loss

This method will discard data between intervals. Though the discarded items will have their destuctors run, using this method incorrectly may lead to unintended data loss. This method is best used to reduce the number of duplicate items after the first has been received, such as repeated mouse clicks or key presses. This method may lead to unintended data loss when used to discard unique items, such as network request.

Example
use futures_lite::prelude::*;
use futures_time::prelude::*;
use futures_time::time::{Instant, Duration};
use futures_time::stream;

fn main() {
   async_io::block_on(async {
       let mut counter = 0;
       stream::interval(Duration::from_millis(100))
           .take(4)
           .sample(Duration::from_millis(200))
           .for_each(|_| counter += 1)
           .await;

       assert_eq!(counter, 2);
   })
}

Group items into vectors which are yielded at every interval.

In addition to using a time source as a deadline, any stream can be used as a deadline too. This enables more interesting buffer strategies to be built on top of this primitive.

Future Improvements
  • Lending iterators would allow for internal reusing of the buffer. Though different from Iterator::windows, it could be more efficient.
  • Contexts/capabilities would enable custom allocators to be used.
Example
use futures_lite::prelude::*;
use futures_time::prelude::*;
use futures_time::time::{Instant, Duration};
use futures_time::stream;

fn main() {
    async_io::block_on(async {
        let mut counter = 0;
        stream::interval(Duration::from_millis(5))
            .take(10)
            .buffer(Duration::from_millis(20))
            .for_each(|buf| counter += buf.len())
            .await;

        assert_eq!(counter, 10);
    })
}

Yield the last item received at the end of a window which resets with each item received.

Every time an item is yielded by the underlying stream, the window is reset. Once the window expires, the last item seen will be yielded. This means that in order to yield an item, no items must be received for the entire window, or else the window will reset.

This method is useful to perform actions at the end of bursts of events, where performing that same action on every event might not be economical.

See also sample() and throttle().

Example
use futures_lite::prelude::*;
use futures_time::prelude::*;
use futures_time::time::{Instant, Duration};
use futures_time::stream;

fn main() {
    async_io::block_on(async {
        let mut counter = 0;
        stream::interval(Duration::from_millis(10))
            .take(10)
            .debounce(Duration::from_millis(20)) // the window is greater than the interval
            .for_each(|_| counter += 1)
            .await;

        assert_eq!(counter, 1); // so only the last item is received
    })
}

Delay the yielding of items from the stream until the given deadline.

The underlying stream will not be polled until the deadline has expired. In addition to using a time source as a deadline, any future can be used as a deadline too. When used in combination with a multi-consumer channel, this method can be used to synchronize the start of multiple streams and futures.

Example
use futures_lite::prelude::*;
use futures_time::prelude::*;
use futures_time::time::{Instant, Duration};
use futures_lite::stream;

fn main() {
    async_io::block_on(async {
        let now = Instant::now();
        let delay = Duration::from_millis(100);
        let _ = stream::once("meow").delay(delay).next().await;
        assert!(now.elapsed() >= *delay);
    });
}

Suspend or resume execution of a stream.

When this method is called the execution of the stream will be put into a suspended state until the channel returns Parker::Unpark or the channel’s senders are dropped. The underlying stream will not be polled while the it is paused.

Yield an item, then ignore subsequent items for a duration.

In addition to using a time-based interval, this method can take any stream as a source. This enables throttling based on alternative event sources, such as variable-rate timers.

See also sample() and debounce().

Data Loss

This method will discard data between intervals. Though the discarded items will have their destuctors run, using this method incorrectly may lead to unintended data loss. This method is best used to reduce the number of duplicate items after the first has been received, such as repeated mouse clicks or key presses. This method may lead to unintended data loss when used to discard unique items, such as network request.

Examples
use futures_lite::prelude::*;
use futures_time::prelude::*;
use futures_time::time::Duration;
use futures_time::stream;

fn main() {
    async_io::block_on(async {
        let mut counter = 0;
        stream::interval(Duration::from_millis(100))  // Yield an item every 100ms
            .take(4)                                  // Stop after 4 items
            .throttle(Duration::from_millis(300))     // Only let an item through every 300ms
            .for_each(|_| counter += 1)               // Increment a counter for each item
            .await;

        assert_eq!(counter, 2);
    })
}

Return an error if a stream does not yield an item within a given time span.

Typically timeouts are, as the name implies, based on time. However this method can time out based on any future. This can be useful in combination with channels, as it allows (long-lived) streams to be cancelled based on some external event.

When a timeout is returned, the stream will be dropped and destructors will be run.

Example
use futures_lite::prelude::*;
use futures_time::prelude::*;
use futures_time::time::{Instant, Duration};
use futures_lite::stream;
use std::io;

fn main() {
    async_io::block_on(async {
        let res = stream::once("meow")
            .delay(Duration::from_millis(100))  // longer delay
            .timeout(Duration::from_millis(50)) // shorter timeout
            .next()
            .await;
        assert_eq!(res.unwrap().unwrap_err().kind(), io::ErrorKind::TimedOut); // error

        let res = stream::once("meow")
            .delay(Duration::from_millis(50))    // shorter delay
            .timeout(Duration::from_millis(100)) // longer timeout
            .next()
            .await;
        assert_eq!(res.unwrap().unwrap(), "meow"); // success
    });
}

Implementors