now_or_never for Streams

[khoover]

The idea is, we want to get all items a Stream is ready to yield, without worrying about future items. We currently have StreamExt::ready_chunks, which can somewhat do this, but it has three downsides: it requires allocation, it caps the number of ready items you get at a time, and it creates a stream itself. The last can be worked around via stream.ready_chunks().next(), but it's an awkward pattern.

I'm proposing StreamExt::iter_ready_now(self), which returns an iterator yielding stream items until either the stream ends or poll_next returns Pending. The intended behaviour can be modeled using existing constructs as:

use std::iter;
use futures::{stream::{StreamExt, Stream}, future::FutureExt};

pub fn iter_ready_now<S: Stream + Unpin>(stream: S) -> impl Iterator<Item = S::Item> {
    let mut stream = stream.fuse();
    iter::from_fn(move || stream.next().now_or_never().flatten())
}

A second possible method would be StreamExt::iter_ready(self), returning a Stream of such iterators outlined above. This one seems more tricky, because both the resulting stream and the iterators it outputs would need access to the original stream.

EDIT: After looking at Iterator again, rather than being like Future and warning that next()-after-None can panic, it explicitly mentions some iterators can decide to start outputting values again. So perhaps instead of iter_ready_now or iter_ready, we have iter_when_ready giving a single iterator struct that is explicitly unfused.