并发执行任务时不能使用filter_map和buffer_unordered？

Question

我正在查看在 Rust 中同时下载东西。

大致是这样的：

#![feature(async_closure)]

use futures::{stream, StreamExt}; // 0.3.13

async fn foo() {
    let xs_new = stream::once(async { 42 })
        .map(async move |x| {
            Some(x + 1)
        })
        .buffer_unordered(42);
}

但是，我希望使用 filter_map 来做这样的事情：

#![feature(async_closure)]

use futures::{stream, StreamExt}; // 0.3.13

async fn foo() {
    let xs_new = stream::once(async { 42 })
        .filter_map(async move |x| if x % 2 == 0 { Some(x + 1) } else { None })
        .buffer_unordered(42);
}

然而，这失败并出现错误：“{integer} 不是 Future，特征 ... 未为 {integer} 实现”。

有谁知道为什么 filter_map 失败但 map 工作正常？

Answer 1

buffer_unordered requires the Items of the Stream to themselves be Futures. Using map 和 async 闭包有效，因为它将整数转换为 Futures 产生一个整数。

使用 filter_map 需要 return 一个 Future 产生一个 Option 决定是否过滤。但是，您忘记了通过 return Some(Future):

将整数转换为 Futures

async fn foo() {
    let xs_new = stream::once(async { 42 })
        .filter_map(async move |x| {
            if x % 2 == 0 {
                Some(async move { x + 1 }) // <--------
            } else {
                None
            }
        })
        .buffer_unordered(42);
}

并发执行任务时不能使用filter_map和buffer_unordered？

Cannot use filter_map with buffer_unordered when concurrently executing tasks?

rust

rust-tokio