|
| 1 | +//! Composable asynchronous iteration. |
| 2 | +//! |
| 3 | +//! If futures are asynchronous values, then streams are asynchronous |
| 4 | +//! iterators. If you've found yourself with an asynchronous collection of some kind, |
| 5 | +//! and needed to perform an operation on the elements of said collection, |
| 6 | +//! you'll quickly run into 'streams'. Streams are heavily used in idiomatic |
| 7 | +//! asynchronous Rust code, so it's worth becoming familiar with them. |
| 8 | +//! |
| 9 | +//! Before explaining more, let's talk about how this module is structured: |
| 10 | +//! |
| 11 | +//! # Organization |
| 12 | +//! |
| 13 | +//! This module is largely organized by type: |
| 14 | +//! |
| 15 | +//! * [Traits] are the core portion: these traits define what kind of streams |
| 16 | +//! exist and what you can do with them. The methods of these traits are worth |
| 17 | +//! putting some extra study time into. |
| 18 | +//! * Functions provide some helpful ways to create some basic streams. |
| 19 | +//! * Structs are often the return types of the various methods on this |
| 20 | +//! module's traits. You'll usually want to look at the method that creates |
| 21 | +//! the `struct`, rather than the `struct` itself. For more detail about why, |
| 22 | +//! see '[Implementing Stream](#implementing-stream)'. |
| 23 | +//! |
| 24 | +//! [Traits]: #traits |
| 25 | +//! |
| 26 | +//! That's it! Let's dig into streams. |
| 27 | +//! |
| 28 | +//! # Stream |
| 29 | +//! |
| 30 | +//! The heart and soul of this module is the [`Stream`] trait. The core of |
| 31 | +//! [`Stream`] looks like this: |
| 32 | +//! |
| 33 | +//! ``` |
| 34 | +//! # use core::task::{Context, Poll}; |
| 35 | +//! # use core::pin::Pin; |
| 36 | +//! trait Stream { |
| 37 | +//! type Item; |
| 38 | +//! fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>; |
| 39 | +//! } |
| 40 | +//! ``` |
| 41 | +//! |
| 42 | +//! Unlike `Iterator`, `Stream` makes a distinction between the [`poll_next`] |
| 43 | +//! method which is used when implementing a `Stream`, and a (to-be-implemented) |
| 44 | +//! `next` method which is used when consuming a stream. Consumers of `Stream` |
| 45 | +//! only need to consider `next`, which when called, returns a future which |
| 46 | +//! yields `Option<Stream::Item>`. |
| 47 | +//! |
| 48 | +//! The future returned by `next` will yield `Some(Item)` as long as there are |
| 49 | +//! elements, and once they've all been exhausted, will yield `None` to indicate |
| 50 | +//! that iteration is finished. If we're waiting on something asynchronous to |
| 51 | +//! resolve, the future will wait until the stream is ready to yield again. |
| 52 | +//! |
| 53 | +//! Individual streams may choose to resume iteration, and so calling `next` |
| 54 | +//! again may or may not eventually yield `Some(Item)` again at some point. |
| 55 | +//! |
| 56 | +//! [`Stream`]'s full definition includes a number of other methods as well, |
| 57 | +//! but they are default methods, built on top of [`poll_next`], and so you get |
| 58 | +//! them for free. |
| 59 | +//! |
| 60 | +//! [`Poll`]: super::task::Poll |
| 61 | +//! [`poll_next`]: Stream::poll_next |
| 62 | +//! |
| 63 | +//! # Implementing Stream |
| 64 | +//! |
| 65 | +//! Creating a stream of your own involves two steps: creating a `struct` to |
| 66 | +//! hold the stream's state, and then implementing [`Stream`] for that |
| 67 | +//! `struct`. |
| 68 | +//! |
| 69 | +//! Let's make a stream named `Counter` which counts from `1` to `5`: |
| 70 | +//! |
| 71 | +//! ```no_run |
| 72 | +//! #![feature(async_stream)] |
| 73 | +//! # use core::stream::Stream; |
| 74 | +//! # use core::task::{Context, Poll}; |
| 75 | +//! # use core::pin::Pin; |
| 76 | +//! |
| 77 | +//! // First, the struct: |
| 78 | +//! |
| 79 | +//! /// A stream which counts from one to five |
| 80 | +//! struct Counter { |
| 81 | +//! count: usize, |
| 82 | +//! } |
| 83 | +//! |
| 84 | +//! // we want our count to start at one, so let's add a new() method to help. |
| 85 | +//! // This isn't strictly necessary, but is convenient. Note that we start |
| 86 | +//! // `count` at zero, we'll see why in `poll_next()`'s implementation below. |
| 87 | +//! impl Counter { |
| 88 | +//! fn new() -> Counter { |
| 89 | +//! Counter { count: 0 } |
| 90 | +//! } |
| 91 | +//! } |
| 92 | +//! |
| 93 | +//! // Then, we implement `Stream` for our `Counter`: |
| 94 | +//! |
| 95 | +//! impl Stream for Counter { |
| 96 | +//! // we will be counting with usize |
| 97 | +//! type Item = usize; |
| 98 | +//! |
| 99 | +//! // poll_next() is the only required method |
| 100 | +//! fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { |
| 101 | +//! // Increment our count. This is why we started at zero. |
| 102 | +//! self.count += 1; |
| 103 | +//! |
| 104 | +//! // Check to see if we've finished counting or not. |
| 105 | +//! if self.count < 6 { |
| 106 | +//! Poll::Ready(Some(self.count)) |
| 107 | +//! } else { |
| 108 | +//! Poll::Ready(None) |
| 109 | +//! } |
| 110 | +//! } |
| 111 | +//! } |
| 112 | +//! ``` |
| 113 | +//! |
| 114 | +//! # Laziness |
| 115 | +//! |
| 116 | +//! Streams are *lazy*. This means that just creating a stream doesn't _do_ a |
| 117 | +//! whole lot. Nothing really happens until you call `next`. This is sometimes a |
| 118 | +//! source of confusion when creating a stream solely for its side effects. The |
| 119 | +//! compiler will warn us about this kind of behavior: |
| 120 | +//! |
| 121 | +//! ```text |
| 122 | +//! warning: unused result that must be used: streams do nothing unless polled |
| 123 | +//! ``` |
| 124 | +
|
| 125 | +mod stream; |
| 126 | + |
| 127 | +pub use stream::Stream; |
0 commit comments