stream: add StreamExt::filter_map_async#7971
stream: add StreamExt::filter_map_async#7971figsoda wants to merge 6 commits intotokio-rs:masterfrom
StreamExt::filter_map_async#7971Conversation
figsoda
changed the title
StreamExt::filter_map_async
There was a problem hiding this comment.
I also wanted to implement StreamExt::filter_async, but that seemed to require AsyncFnMut due to lifetime issues, which hasn't been in stable Rust for 6 months yet which is the MSRV policy. So I will hold on to that until then.
Yeah, the StreamExt::filter_async is tricky due to lifetime issue and it might be less useful before resolving this lifetime issue. I think filter_map_async is useful, but I prefer to hold off on implementing of filter_async as it is less useful at this stage.
ADD-SP
left a comment
ADD-SP
left a comment
There was a problem hiding this comment.
The only test is the doctest, since I couldn't find any standalone tests for the surrounding functions. I have tested it for my use case, but it might still be helpful to have more tests in tree.
Is tokio-stream/tests a good place for a new test file?
|
added some tests and fixed a few oversights |
mattiapitossi
added
A-tokio-stream
| if let Some(future) = me.future.as_mut().as_pin_mut() { | ||
| let item = ready!(future.poll(cx)); | ||
| me.future.set(None); | ||
| if let Some(item) = item { | ||
| return Poll::Ready(Some(item)); |
There was a problem hiding this comment.
So while we are polling the future, we do not poll the stream. It implies that something like stream.buffer_unordered().filter_map_async(...) is really dangerous because ongoing futures inside of the buffer_unordered() just pause out of nowhere.
On the other hand, this is a pre-existing problem with stream.then() too.
There was a problem hiding this comment.
Would the alternative be keeping a buffer of items, so the entire stream can be polled by awaiting on one .next()?
5 participants
Motivation
This is like the async closure version of
StreamExt::filter_map, similar toStreamExt::thenforStreamExt::map.Solution
The code is largely adapted from
StreamExt::thenandStreamExt::filter_map. I'm not super sure about the namefilter_map_async, as this will probably be what other async closure variants will be named after.The only test is the doctest, since I couldn't find any standalone tests for the surrounding functions. I have tested it for my use case, but it might still be helpful to have more tests in tree.
I also wanted to implement
StreamExt::filter_async, but that seemed to requireAsyncFnMutdue to lifetime issues, which hasn't been in stable Rust for 6 months yet which is the MSRV policy. So I will hold on to that until then.