Add a Stream::try_iter() method

[notgull]

This PR adds a try_iter method to StreamExt that consumes all of the values until it gets Pending, at which point it returns None. The goal is to get a method similar to Receiver::try_iter() on the async_channel::Receiver type that would allow users to consume all of the values in the channel without waiting.

[taiki-e]

I'm confused about this is named _iter. Implementation has no relationship to the Iterator trait.

[notgull]

I basically just took the name from libstd's try_iter. Maybe a better name for this would be s.ready() or s.immediate()?

[taiki-e]

poll_immediate and now_or_never in futures are somewhat similar to this, but I don't think they are exactly the same.

Well, this doesn't even require implementing Stream in the first place. Implementing Iterator might actually work in more contexts.

[notgull]

Well, this doesn't even require implementing Stream in the first place. Implementing Iterator might actually work in more contexts.

Unfortunately we still need a Context to pass down to the underlying Stream, just like for future::poll_once(). We could add functionality to create a "no-op waker" for this if we wanted to.

[taiki-e]

Unfortunately we still need a Context to pass down to the underlying Stream, just like for future::poll_once().

Oh, good point. I forgot that because now_or_never uses noop_context.

[fogti]

Suggested change

	match self.stream.poll_next(cx) {
	Poll::Ready(x) => Poll::Ready(x),
	Poll::Pending => Poll::Ready(None),
	}
	Poll::Ready(match self.stream.poll_next(cx) {
	Poll::Ready(x) => x,
	Poll::Pending => None,
	})

[notgull]

I don't think this looks nice

[fogti]

Usage of this would probably introduce unnecessary spurious wake-ups due to the lack of smth. like noop_context.

[notgull]

Yes, it would be nice to have a noop_waker or a panic_waker for futures that are guaranteed to not return Pending. Unfortunately defining that using only safe code requires an unnecessary allocation.

[fogti]

Could the allocation be cached (similar to the caching in block_on)?

[notgull]

It could, but you would still need an atomic operation to clone it out of the thread local. While this is acceptable overhead for block_on, I imagine that try_iter() would be called much more frequently.

I don't think there's anything wrong with just passing try_iter() to stream::block_on, just like poll_once() is passed to future::block_on or spin_on.

[fogti]

I think the concern about the name still applies here. I'd propose try_stream in analogue to try_iter, but idk if that's already in use...

[fogti]

The stream isn't necessarily fused, right? I don't think "until" does capture what happens correctly... It might also be a good idea to provide a way to access the inner stream, in case one wants that....

[notgull]

It might also be a good idea to provide a way to access the inner stream, in case one wants that....

I've added accessor methods. However there are many other streams in this crate that don't provide access to their underlying streams. I'm not sure if it's a good idea to add those accessor methods there as well...

Definitely out of scope for this PR though.

[fogti]

Also, what about pinning? How does this interact with that?

[notgull]

Also, what about pinning? How does this interact with that?

It only takes Unpin streams. However it's possible to make it take a !Unpin stream by first pinning the stream to the stack.

let x = stream::once_future(async { 1 });
pin!(x);
let y = x.try_stream();

[fogti]

Error: cannot find function block_on in module futures_lite::future