[FEATURE] more combinators

[Niedzwiedzw]

futures-util StreamExt and TryStreamExt provide bunch of very useful (addictive?) combinators. I wonder if introducing something like that would be accepted as a PR for this project.

[yoshuawuyts]

@Niedzwiedzw it depends – which combinators did you have in mind?

[Niedzwiedzw]

I heavily depend on stuff like .filter_map() .try_filter_map() .flat_map() .scan() .chain() .flatten() .try_flatten() .try_for_each() etc :) those are the ones that come to mind at least

[failable]

For me, that's a big missing part.

[nanoqsh]

@Niedzwiedzw @failable
I think the point of futures_concurrency crate is to provide combinators for concurrency and for other things are other crates with stream combinators like futures or futures_lite. They fit together perfectly, so far they don't have any conflicts and I hope there won't be any:

use {
    futures_concurrency::stream::StreamExt as _,
    futures_lite::{stream, StreamExt as _},
};

let a = stream::iter([1, 2, 3]);
let b = stream::iter([4, 5, 6]);
a
    .merge(b)
//   ^^^^^ merge from futures_concurrency
    .filter(|i| i % 2 == 0)
//   ^^^^^^ filter from futures_lite
    .for_each(|i| println!("{i}"))
    .await;

In my opinion, there is no need to overload the crate in any additionals.

But I wonder is it possible to use ConcurrentStream as a stream and use all this combinators on it?

[tyilo]

@nanoqsh I don't know if this issue refers to StreamExt or ConcurrentStream, but I think it would make sense to add more functions to ConcurrentStream, which take an async function instead of futures::StreamExt where they take a normal function.

So that you can do .filter(|i| async { i % 2 == 0 }).for_each(|i| async { println!("{i}") }) (for_each already exists on ConcurrentStream, but filter does not).

[Niedzwiedzw]

could even be a separate extension trait, like futures-util does