Rust中async_stream生成的Stream如何确定正确返回类型

问题参考示例 #

请参考如下最简Stream示例：

use futures::StreamExt;
use async_stream::stream;

#[tokio::main]
async fn main() {
    let ticks = stream! {
        yield Ok(0);
        yield Err("");
    };
    futures::pin_mut!(ticks);
    while let Some(x) = ticks.next().await {
        println!("{:?}", x);
    }
}

上述代码中变量ticks的正确类型是什么？
封装需求：需要找到适配的函数签名，将流生成逻辑封装为独立函数，待补全的代码结构如下：

use futures::StreamExt;
use async_stream::stream;

fn query() -> ???? {
    return stream! {
        yield Ok(0);
        yield Err("");
    };
}

#[tokio::main]
async fn main() {
    let ticks = query();
    futures::pin_mut!(ticks);
    while let Some(x) = ticks.next().await {
        println!("{:?}", x);
    }
}

核心问题：如何正确推导async_stream宏生成值的具体类型？

解答 #

async_stream::stream!宏生成的流属于编译器生成的匿名不透明类型，和async块生成的Future类型类似，没有可手动书写的公开具名类型，只能通过以下两种方式作为函数返回值：

方案1：使用impl Stream静态分发（零开销，优先推荐） #

首先推导流的Item类型：查看stream!块内所有yield输出值的公共统一类型即可，逻辑和迭代器元素类型推导完全一致。
示例中两个yield值分别是Ok(0)（Result<i32, _>的Ok变体）、Err("")（Result<_, &'static str>的Err变体），统一后Item类型为Result<i32, &'static str>。
直接将返回值写为impl Stream<Item = 推导出来的Item类型>即可，补全后可运行代码如下：

// 引入Stream trait
use futures::Stream;
use futures::StreamExt;
use async_stream::stream;

fn query() -> impl Stream<Item = Result<i32, &'static str>> {
    stream! {
        yield Ok(0);
        yield Err("");
    }
}

#[tokio::main]
async fn main() {
    let ticks = query();
    futures::pin_mut!(ticks);
    while let Some(x) = ticks.next().await {
        println!("{:?}", x);
    }
}

该方式为静态分发，无任何运行时开销，适用于函数所有分支返回同一种流类型的场景，是封装async_stream流的首选方案。

方案2：使用Pin<Box<dyn Stream>>动态分发（适配多分支返回不同流的场景） #

如果函数需要根据条件返回不同的stream!生成的流（不同宏生成的匿名类型不同，无法用impl Stream统一返回），可以用特征对象做类型擦除：

use futures::Stream;
use futures::StreamExt;
use async_stream::stream;
use std::pin::Pin;

// 需要跨线程await时加Send绑定，单线程使用可移除
fn query(flag: bool) -> Pin<Box<dyn Stream<Item = Result<i32, &'static str>> + Send>> {
    if flag {
        Box::pin(stream! {
            yield Ok(0);
            yield Err("first stream error");
        })
    } else {
        Box::pin(stream! {
            yield Ok(1);
            yield Err("second stream error");
        })
    }
}

该方式存在轻微的指针间接访问开销，但适配性更强。

注意事项 #

不要尝试通过编译器报错、类型打印等方式获取stream!生成的具体具名类型：该类型是内部生成的状态机类型，名称极长且随编译器版本、代码逻辑变化，没有任何手动书写的可行性。

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