Use AsyncFn family of traits (async closures)

[taiki-e]

async closures have been stabilized in rust-lang/rust#132706 (Rust 1.85).

[tisonkun]

This looks interesting and I'd like to investigate it in futrues-rs.

One prerequisite is bumping the MSRV and I wonder what our MSRV strategy is.

That said, I'm trying to replace the dependency to pin-utils (only used for pin_utils::pin_mut) with core::pin::pin but noticed that futures-utils' MSRV is 1.63 < 1.68.

This issue requires an MSRV >= 1.85 and I'm unsure when we can bump the MSRV to unblock the actual migration work.

[taiki-e]

The next MSRV bump will probably be to 1.85 or 1.86 when Debian 13 is released. It will probably be 4 to 8 months later.

However, you probably don't need to worry about that since the master branch is not scheduled for immediate release.

---

Btw, this is actually already supported at least for some simple cases:

use futures::stream::{self, StreamExt};

let stream = stream::iter(1..=3);
let stream = stream.then(async move |x| x + 3);

assert_eq!(vec![4, 5, 6], stream.collect::<Vec<_>>().await);

playground

However, doesn't seem to work if the closure bound is FnMut/Fn and argument of the closure contains a reference.

use futures::stream::{self, StreamExt};

let stream = stream::iter(1..=3);
// ok
let stream = stream.filter(|_x| async { true });
// error
let stream = stream.filter(async |_x| true);

error[E0308]: mismatched types
   --> src/main.rs:9:18
    |
9   |     let stream = stream.filter(async |_x| true);
    |                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ one type is more general than the other
    |
    = note: expected `async` closure body `{async closure body@src/main.rs:9:43: 9:47}`
               found `async` closure body `{async closure body@src/main.rs:9:43: 9:47}`
    = note: no two async blocks, even if identical, have the same type
    = help: consider pinning your async block and casting it to a trait object
note: the lifetime requirement is introduced here
   --> /playground/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/futures-util-0.3.31/src/stream/stream/mod.rs:422:34
    |
422 |         F: FnMut(&Self::Item) -> Fut,
    |                                  ^^^

playground

rust-lang/rust#132706 says:

Interactions between async closures and the Fn* family of traits

Async closures always implement FnOnce, since they always can be called once. They may also implement Fn or FnMut if their body is compatible with the calling mode (i.e. if they do not mutate their captures, or they do not capture their captures, respectively) and if the future returned by the async closure is not lending.

let id = String::new();

let mapped: Vec</* impl Future */> =
    [/* elements */]
    .into_iter()
    // `Iterator::map` takes an `impl FnMut`
    .map(async |element| {
        do_something(&id, element).await;
    })
    .collect();

[umbrageodotus]

progress?

[oconnor663]

Will the plan be to have StreamExt::for_each and friends switch to taking AsyncFnMut, or will there need to be new methods for backwards compatibility reasons? (I can't tell if that switch is backwards compatible with the regular-closure-returning-async-block pattern. In simple cases it seems like it is.)

[tisonkun]

Here is a case where using AsyncFnOnce doesn't break compatibility - fast/mea#84