如何将Rust中TryStream<Ok=Vec<Item>>转换为TryStream<Ok=Item>？

我正在使用Rust的futures-util crate处理流操作，目前实现了一个用于逐页请求分页API的流，每页返回结果为Vec<Item>，且每次请求页面可能失败。当前函数返回类型为impl TryStream<Ok = Vec<Item>, Error = MyError>，代码如下：

pub fn search(
    &self,
    query: Query,
) -> impl TryStream<Ok = Vec<Item>, Error = MyError> {
    let initial_state = (query, None);
    stream::try_unfold(initial_state, |(query, page_token)| async move {
        let (items, next_page_token) = fetch_page(&query, page_token).await?;
        if items.len() <= 0 {
            Ok(None)
        } else {
            Ok(Some((items, (query, next_page_token))))
        }
    })
}

我希望将返回类型修改为impl TryStream<Ok = Item, Error = MyError>，让流逐个输出Item，且获取页面失败时返回错误并关闭流。请问这是否可行？

完全可行，只需基于现有流做简单转换即可实现需求。futures-util的TryStreamExt trait提供的try_flatten方法可以直接完成流的扁平化：将每个分页返回的Vec<Item>拆分为单个Item的流，同时保留错误处理逻辑——一旦分页请求失败，整个流会立即终止并返回对应的错误。

修改后的代码如下：

use futures_util::{stream, TryStreamExt};

pub fn search(
    &self,
    query: Query,
) -> impl TryStream<Ok = Item, Error = MyError> {
    let initial_state = (query, None);
    stream::try_unfold(initial_state, |(query, page_token)| async move {
        let (items, next_page_token) = fetch_page(&query, page_token).await?;
        // 用is_empty()判断空集合更符合Rust规范
        if items.is_empty() {
            Ok(None)
        } else {
            Ok(Some((items, (query, next_page_token))))
        }
    })
    // 将每个Vec<Item>转换为单个Item的流，再通过try_flatten合并
    .try_map(|items| stream::iter(items).map_err(|_| unreachable!()))
    .try_flatten()
}

核心逻辑说明 #

• try_map：把每个分页返回的Vec<Item>转换成由单个Item组成的流。由于stream::iter生成的普通流不会产生错误，这里用map_err填充一个永远不会触发的MyError，仅为满足类型匹配要求。
• try_flatten：将嵌套的TryStream结构扁平化为单层流。如果上游的分页请求返回错误，try_flatten会立即终止整个流并传递该错误，完全符合你要求的失败处理逻辑。

除了try_flatten，也可以用flat_map系列方法，但try_flatten是最简洁且贴合需求的实现方式，能确保逐个输出Item的同时，在分页失败时及时终止流。

内容的提问来源于stack exchange，提问作者JasterV