Streams

Streams are similar to futures, but instead of yielding a single value, theyasynchronously yield one or more values. They can be thought of as asynchronousiterators.

Just like futures, streams are able to represent a wide range of things as longas those things produce discrete values at different points sometime in thefuture. For instance:

  • UI Events caused by the user interacting with a GUI in different ways. When anevent happens the stream yields a different message to your app over time.
  • Push Notifications from a server. Sometimes a request/response model is notwhat you need. A client can establish a notification stream with a server to beable to receive messages from the server without specifically being requested.
  • Incoming socket connections. As different clients connect to a server, theconnections stream will yield socket connections.

The Stream trait

Just like Future, implementing Stream is common when using Tokio. TheStream trait is as follows:

  1. trait Stream {
  2.     /// The type of the value yielded by the stream.
  3.     type Item;
  5.     /// The type representing errors that occurred while processing the computation.
  6.     type Error;
  8.     /// The function that will be repeatedly called to see if the stream has
  9.     /// another value it can yield
  10.     fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error>;
  11. }

The Item associated type is the type yielded by the stream. The Errorassociated type is the type of the error yielded when something unexpectedhappens. The poll function is very similar to Future’s poll function. Theonly difference is that, this time, Option<Self::Item> is returned.

Stream implementations have the poll function called many times. When the nextvalue is ready, Ok(Async::Ready(Some(value))) is returned. When the stream isnot ready to yield a value, Ok(Async::NotReady) is returned. When thestream is exhausted and will yield no further values, Ok(Async::Ready(None))is returned. Just like with futures, streams must not returnAsync::NotReady unless Async::NotReady was obtained by an inner stream orfuture.

When the stream encounters an error, Err(error) is returned. Returning anerror does not signify that the stream is exhausted. The error may betransient and the caller may try calling poll again in the future and valuesmay be produced again. If the error is fatal, then the next call to pollshould return Ok(Async::Ready(None)).

Fibonacci

The following example shows how to implement the fibonacci sequence as a stream.

  1. extern crate futures;
  3. use futures::{Stream, Poll, Async};
  5. pub struct Fibonacci {
  6.     curr: u64,
  7.     next: u64,
  8. }
  10. impl Fibonacci {
  11.     fn new() -> Fibonacci {
  12.         Fibonacci {
  13.             curr: 1,
  14.             next: 1,
  15.         }
  16.     }
  17. }
  19. impl Stream for Fibonacci {
  20.     type Item = u64;
  22.     // The stream will never yield an error
  23.     type Error = ();
  25.     fn poll(&mut self) -> Poll<Option<u64>, ()> {
  26.         let curr = self.curr;
  27.         let next = curr + self.next;
  29.         self.curr = self.next;
  30.         self.next = next;
  32.         Ok(Async::Ready(Some(curr)))
  33.     }
  34. }

To use the stream, a future must be built that consumes it. The following futurewill take a stream and display 10 items from it.

  1. #[macro_use]
  2. extern crate futures;
  4. use futures::{Future, Stream, Poll, Async};
  5. use std::fmt;
  7. pub struct Display10<T> {
  8.     stream: T,
  9.     curr: usize,
  10. }
  12. impl<T> Display10<T> {
  13.     fn new(stream: T) -> Display10<T> {
  14.         Display10 {
  15.             stream,
  16.             curr: 0,
  17.         }
  18.     }
  19. }
  21. impl<T> Future for Display10<T>
  22. where
  23.     T: Stream,
  24.     T::Item: fmt::Display,
  25. {
  26.     type Item = ();
  27.     type Error = T::Error;
  29.     fn poll(&mut self) -> Poll<(), Self::Error> {
  30.         while self.curr < 10 {
  31.             let value = match try_ready!(self.stream.poll()) {
  32.                 Some(value) => value,
  33.                 // There were less than 10 values to display, terminate the
  34.                 // future.
  35.                 None => break,
  36.             };
  38.             println!("value #{} = {}", self.curr, value);
  39.             self.curr += 1;
  40.         }
  42.         Ok(Async::Ready(()))
  43.     }
  44. }
  45. # fn main() {}

Now, the fibonacci sequence can be displayed:

  1. extern crate tokio;
  2. # extern crate futures;
  3. # struct Fibonacci;
  4. # impl Fibonacci { fn new() { } }
  5. # struct Display10<T> { v: T };
  6. # impl<T> Display10<T> {
  7. # fn new(_: T) -> futures::future::FutureResult<(), ()> {
  8. # futures::future::ok(())
  9. # }
  10. # }
  12. let fib = Fibonacci::new();
  13. let display = Display10::new(fib);
  15. tokio::run(display);

Getting asynchronous

So far, the fibonacci stream is synchronous. Lets make it asynchronous bywaiting a second between values. To do this,tokio::timer::Interval is used. Interval is, itself, a streamthat yields () values at the requested time interval. Calling Interval::pollbetween intervals results in Async::NotReady being returned.

The Fibonacci stream is updated as such:

  1. #[macro_use]
  2. extern crate futures;
  3. extern crate tokio;
  5. use tokio::timer::Interval;
  6. use futures::{Stream, Poll, Async};
  7. use std::time::Duration;
  9. pub struct Fibonacci {
  10.     interval: Interval,
  11.     curr: u64,
  12.     next: u64,
  13. }
  15. impl Fibonacci {
  16.     fn new(duration: Duration) -> Fibonacci {
  17.         Fibonacci {
  18.             interval: Interval::new_interval(duration),
  19.             curr: 1,
  20.             next: 1,
  21.         }
  22.     }
  23. }
  25. impl Stream for Fibonacci {
  26.     type Item = u64;
  28.     // The stream will never yield an error
  29.     type Error = ();
  31.     fn poll(&mut self) -> Poll<Option<u64>, ()> {
  32.         // Wait until the next interval
  33.         try_ready!(
  34.             self.interval.poll()
  35.                 // The interval can fail if the Tokio runtime is unavailable.
  36.                 // In this example, the error is ignored.
  37.                 .map_err(|_| ())
  38.         );
  40.         let curr = self.curr;
  41.         let next = curr + self.next;
  43.         self.curr = self.next;
  44.         self.next = next;
  46.         Ok(Async::Ready(Some(curr)))
  47.     }
  48. }
  49. # fn main() {}

The Display10 future already supports asynchronicity so it does not need to beupdated.

To run the throttled fibonacci sequence, include an interval:

  1. extern crate tokio;
  2. # extern crate futures;
  3. # struct Fibonacci;
  4. # impl Fibonacci { fn new(dur: Duration) { } }
  5. # struct Display10<T> { v: T };
  6. # impl<T> Display10<T> {
  7. # fn new(_: T) -> futures::future::FutureResult<(), ()> {
  8. # futures::future::ok(())
  9. # }
  10. # }
  12. use std::time::Duration;
  14. let fib = Fibonacci::new(Duration::from_secs(1));
  15. let display = Display10::new(fib);
  17. tokio::run(display);

Combinators

Just like futures, streams come with a number of combinators for reducingboilerplate. Many of these combinators exist as functions on theStream trait.

Updating fibonacci stream can be rewritten using the unfold function:

  1. extern crate futures;
  3. use futures::{stream, Stream};
  5. fn fibonacci() -> impl Stream<Item = u64, Error = ()> {
  6.     stream::unfold((1, 1), |(curr, next)| {
  7.         let new_next = curr + next;
  9.         Some(Ok((curr, (next, new_next))))
  10.     })
  11. }

Just like with futures, using stream combinators requires a functional style ofprogramming. Also, impl Stream is used to return the stream from the function.The returning futures strategies apply equally to returning streams.

Display10 is reimplemented using take and for_each:

  1. extern crate tokio;
  2. extern crate futures;
  4. use futures::Stream;
  5. # use futures::stream;
  6. # fn fibonacci() -> impl Stream<Item = u64, Error = ()> {
  7. # stream::once(Ok(1))
  8. # }
  10. tokio::run(
  11.     fibonacci().take(10)
  12.         .for_each(|num| {
  13.             println!("{}", num);
  14.             Ok(())
  15.         })
  16. );

The take combinator limits the fibonacci stream to 10 values. The for_eachcombinator asynchronously iterates the stream values. for_each consumes thestream and returns a future that completes once the closure was called once foreach stream value. It is the asynchronous equivalent to a rust for loop.

Essential combinators

It is worth spending some time with the Stream trait andmodule documentation to gain some familiarity with the full set ofavailable combinators. This guide will provide a very quick overview.

Concrete streams

The stream module contains functions for converting values anditerators into streams.

  • once converts the provided value into an immediately ready stream thatyields a single item: the provided value.
  • iter_ok and iter_result both take IntoIterator values and convertsthem to an immediately ready stream that yields the iterator values.
  • empty returns a stream that immediately yields None.
    For example:
  1. extern crate tokio;
  2. extern crate futures;
  4. use futures::{stream, Stream};
  6. let values = vec!["one", "two", "three"];
  8. tokio::run(
  9.     stream::iter_ok(values).for_each(|value| {
  10.         println!("{}", value);
  11.         Ok(())
  12.     })
  13. )

Adapters

Like Iterator, the Stream trait includes a broad range of “adapter”methods. These methods all consume the stream, returning a new stream providingthe requested behavior. Using these adapter combinators, it is possible to: