外壳：stream! 与 stream_inner

主线位置：宏展开的第一步和最后一步。stream! 把输入交给 proc-macro，stream_inner 解析输入、交给 Scrub 改写（下一章），再把改写后的语句包进一个固定的骨架。本章讲两头，中间的改写留给 Scrub。

flowchart TB
  user["用户：stream! { 语句 }"] --> mr["macro_rules! stream"]
  mr --> inner["__private::stream_inner!(($crate) 语句)"]
  inner --> pi["parse_input"]
  pi --> path["第一个 token 组：crate 路径"]
  pi --> stmts["其余：replace_for_await 后解析成语句"]
  stmts --> scrub["Scrub 逐条改写"]
  scrub --> skel["骨架：pair() + AsyncStream::new"]
  path --> skel

先看结果

一个最简单的 stream!：

pub fn numbers() -> impl Stream<Item = u32> {
  stream! {
    for i in 0..3 {
      yield i;
    }
  }
}

cargo expand 的输出（cargo-expand 1.0.104，函数体部分，未改动）：

pub fn numbers() -> impl Stream<Item = u32> {
    {
        let (mut __yield_tx, __yield_rx) = unsafe {
            ::async_stream::__private::yielder::pair()
        };
        ::async_stream::__private::AsyncStream::new(
            __yield_rx,
            async move {
                '__async_stream_private_check_scope: {
                    for i in 0..3 {
                        {
                            #[allow(unreachable_code)]
                            if false {
                                break '__async_stream_private_check_scope (loop {});
                            }
                            __yield_tx.send(i).await
                        };
                    }
                }
            },
        )
    }
}

几样东西：

一对 __yield_tx / __yield_rx，由 unsafe 的 pair() 造出来。
一个 AsyncStream，把接收端和一个 async move 块放在一起。impl Stream<Item = u32> 就是它。
async 块里是一个带标签 '__async_stream_private_check_scope 的块，你写的语句原样放在里面。
yield i; 变成了一个块：一个永远不执行的 break 到那个标签，再加 __yield_tx.send(i).await。这是 Scrub 的工作。

本章逐个讲这几样东西是怎么来的。

macro_rules! 外壳

用户调用的 stream! 不是 proc-macro，是 async-stream/src/lib.rs 里的一个 macro_rules!：

#[macro_export]
macro_rules! stream {
    ($($tt:tt)*) => {
        $crate::__private::stream_inner!(($crate) $($tt)*)
    }
}

try_stream! 一模一样，只是转给 try_stream_inner!。它接收任意 token，原样转发，只在最前面多塞了一个括号组 ($crate)。这一层壳解决了两个问题。

第一个问题：路径。 生成的代码要调用 pair()、AsyncStream::new，得写出它们的路径。proc-macro 里没有 $crate，如果把路径写死成 ::async_stream::...，那么当用户只依赖一个重新导出了 stream! 的 crate、自己没直接依赖 async-stream 时，这个路径就找不到。$crate 是 macro_rules! 特有的：它展开时总是指向定义这个宏的 crate，不管用户从哪条路径调用。外壳把 $crate 这个 token 原封不动地交给 proc-macro，proc-macro 再把它原样拼进输出，路径就总是对的。

cargo expand 把它打印成 ::async_stream，但它不是一个普通的路径。下面这个实验仿照仓库里的 tests/reexporter 和 tests/test-reexport（0.3.1 的 #46 加的）：reexporter 只有一行 pub use async_stream::stream;，user 只依赖 reexporter：

use futures::{executor::block_on, StreamExt};

fn main() {
  // user 不直接依赖 async-stream，`::async_stream` 这个路径在这里不存在
  let s = reexporter::stream! {
    for i in 0..3u8 {
      yield i;
    }
  };
  println!("{:?}", block_on(s.collect::<Vec<_>>()));
}

[0, 1, 2]

在同一个 user crate 里手写 ::async_stream::__private::yielder::pair::<u8>()，rustc 1.99.0 的报错：

error[E0433]: cannot find `async_stream` in the list of imported crates
 --> user/src/bin/direct.rs:3:22
  |
3 |   let _ = unsafe { ::async_stream::__private::yielder::pair::<u8>() };
  |                      ^^^^^^^^^^^^ could not find `async_stream` in the list of imported crates

宏展开出来的那条「看起来一样」的路径能用，手写的不能用，差别就在 $crate。

第二个问题：卫生性。 骨架里的 __yield_tx、__yield_rx 是宏自己生成的变量名。proc-macro 用 quote! 生成的标识符带 Span::call_site()，意思是「当作写在调用 proc-macro 的那个位置」。调用 stream_inner! 的位置在 macro_rules! 的展开里，而 macro_rules! 对局部变量是卫生的：展开里定义的局部变量，调用方传进来的 token 看不见。所以用户在 stream! 里写 __yield_tx，找不到：

let _s = stream! {
  __yield_tx.send(1u32).await;
};

error[E0425]: cannot find value `__yield_tx` in this scope
 --> src/bin/hygiene2.rs:5:5
  |
5 |     __yield_tx.send(1u32).await;
  |     ^^^^^^^^^^ not found in this scope
  |
help: an identifier with the same name is defined here, but is not accessible due to macro hygiene

作为对照，绕过外壳直接调用 __private::stream_inner!((::async_stream) ...)，__yield_tx 和用户代码处在同一个卫生上下文里，就能直接用：

let mut s = pin!(async_stream::__private::stream_inner!((::async_stream)
  __yield_tx.send(1u32).await;
  yield 2u32;
));

got 1
got 2

这只是为了说明外壳的作用，__private 不是公开 API。宏改写出来的 __yield_tx.send(v).await 和骨架里的 let (mut __yield_tx, ..) 都来自 quote!，带同一种 span，所以它们互相能看见；用户写的 token 和它们隔着一层 macro_rules!，看不见。这也意味着用户没法在 stream! 里拿到 Sender 去做宏没预料到的事。

parse_input：取出路径，解析语句

proc-macro 这一侧的入口是 parse_input：

fn parse_input(input: TokenStream) -> syn::Result<(TokenStream2, Vec<syn::Stmt>)> {
    let mut input = TokenStream2::from(input).into_iter();
    let crate_path = match input.next().unwrap() {
        TokenTree::Group(group) => group.stream(),
        _ => panic!(),
    };
    let stmts = syn::Block::parse_within.parse2(replace_for_await(input))?;
    Ok((crate_path, stmts))
}

第一个 token 树必须是外壳塞进来的那个括号组，取出里面的 $crate。不是组就直接 panic!()，没有好好报错，因为按设计只有外壳会调用它：stream_inner 是 #[doc(hidden)]，上面的注释写明「第一个 token 树必须是装着 async-stream crate 路径的组」。
剩下的 token 先经过 replace_for_await，把 for await 换成 #[await_] for（见 for await），再用 Block::parse_within 解析成一串语句。parse_within 解析的是「花括号里面的内容」，所以 stream! 的输入就是一个块体，不需要再套花括号。
解析失败（比如语法错误）时返回 Err，stream_inner 把它转成 compile_error! 交还给 rustc。

stream_inner 与 try_stream_inner

两个 proc-macro 的函数体（stream_inner、try_stream_inner）逐行相同，唯一的差别是 Scrub::new 的第一个参数 is_try：前者 false，后者 true。is_try 只影响 yield 的改写（try_stream! 里 yield v 发出的是 Ok(v)），不影响 ?，这一点下一章细说。

去掉两者共有的部分，stream_inner 是这样：

pub fn stream_inner(input: TokenStream) -> TokenStream {
    let (crate_path, mut stmts) = match parse_input(input) {
        Ok(x) => x,
        Err(e) => return e.to_compile_error().into(),
    };

    let mut scrub = Scrub::new(false, &crate_path);

    for stmt in &mut stmts {
        scrub.visit_stmt_mut(stmt);
    }

    let dummy_yield = if scrub.has_yielded {
        None
    } else {
        Some(quote!(if false {
            __yield_tx.send(()).await;
        }))
    };

    quote!({
        let (mut __yield_tx, __yield_rx) = unsafe { #crate_path::__private::yielder::pair() };
        #crate_path::__private::AsyncStream::new(__yield_rx, async move {
            '__async_stream_private_check_scope: {
                #dummy_yield
                #(#stmts)*
            }
        })
    })
    .into()
}

三步：解析、逐条语句交给 Scrub 改写、拼骨架。骨架的每个部分：

部分	作用
外层 `{ ... }`	让整个展开是一个表达式，`let s = stream! { .. };` 和直接作为函数返回值都行
`unsafe { pair() }`	造一对零大小的 `Sender` / `Receiver`，类型参数 `T` 靠后面的 `send` 推断。`pair` 是 `unsafe fn`，注释说「除了我们的宏，任何人调用它都被视为 unsound」，标 `unsafe` 是为了让人别把它当成可以随便调的函数
`AsyncStream::new(rx, ..)`	运行时的那个类型，见 poll_next
`async move`	把 `__yield_tx` 和块里用到的外部变量都移进 async 块。`stream!` 返回的流因此可以带着这些值离开当前作用域
`'__async_stream_private_check_scope:`	一个带标签的块。它本身什么都不做，只给每个 `yield` 里那句永不执行的 `break` 一个目标，是 0.3.6 修 #106 时加的，见 Scrub
`#dummy_yield`	只在一次 `yield` 都没有时出现，见下一节

骨架里没有 Scrub 的痕迹，也没有任何循环或状态机：状态机由编译器为 async move 块生成，async-stream 只负责在里面放几个 send().await。

一次都没 yield：dummy_yield

T 是从 __yield_tx.send(v) 推断出来的。如果块里一个 yield 都没有，T 就无从推断。手写一个不带这句的骨架（let s = ...，没有返回类型可以参考），rustc 1.99.0 报：

error[E0282]: type annotations needed
 --> src/bin/nodummy.rs:3:47
  |
3 |   let (mut __yield_tx, __yield_rx) = unsafe { async_stream::__private::yielder::pair() };
  |                                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ cannot infer type of the type parameter `T` declared on the function `pair`

宏的办法是插一句永不执行的 if false { __yield_tx.send(()).await; }，把 T 定成 ()。展开结果：

pub fn empty() -> impl Stream<Item = ()> {
    {
        let (mut __yield_tx, __yield_rx) = unsafe {
            ::async_stream::__private::yielder::pair()
        };
        ::async_stream::__private::AsyncStream::new(
            __yield_rx,
            async move {
                '__async_stream_private_check_scope: {
                    if false {
                        __yield_tx.send(()).await;
                    }
                }
            },
        )
    }
}

所以空的 stream! {} 的 Item 是 ()，不能当成任意类型的空流用：

fn nothing() -> impl Stream<Item = u32> {
  stream! {}
}

rustc 1.99.0 的报错：

error[E0271]: type mismatch resolving `<AsyncStream<(), {async block@lib.rs:193:9}> as Stream>::Item == u32`
 --> src/main.rs:4:17
  |
4 | fn nothing() -> impl Stream<Item = u32> {
  |                 ^^^^^^^^^^^^^^^^^^^^^^^ expected `u32`, found `()`

报错里的 {async block@lib.rs:193:9} 也值得看一眼：async 块的位置是 async-stream 自己的 lib.rs 第 193 行第 9 列，正是外壳里 $crate::__private::stream_inner!(...) 那一行。quote! 生成的 token 用的是调用位置，也就是外壳展开的位置。以后在报错里看到 async block@lib.rs:193:9，就知道说的是某个 stream! 的 async 块。想要空的 Stream<Item = u32>，用 futures::stream::empty()。

和 dummy_yield 相反，用户自己写的语句保留了原来的 span。仓库里的 tests/spans_preserved.rs 在 stream! 里写 assert_eq!(line!(), 8) 来守住这一点：报错和 line!() 都指向你写的那一行，不是宏里的某一行。

小结

stream! / try_stream! 是 5 行的 macro_rules!，把 ($crate) 放在最前面，其余 token 原样转给 proc-macro。
这层外壳解决两件事：$crate 让宏被重新导出后路径仍然正确（实测只依赖 reexporter 的 crate 能用 stream!，手写 ::async_stream::... 报 E0433）；macro_rules! 的卫生性让 __yield_tx 对用户代码不可见（E0425，「not accessible due to macro hygiene」）。
stream_inner 和 try_stream_inner 只差 Scrub 的 is_try；输出骨架是 unsafe { pair() } 加 AsyncStream::new(rx, async move { '__async_stream_private_check_scope: { .. } })。
一次都没 yield 时插入 if false { __yield_tx.send(()).await; }，所以 stream! {} 的 Item 是 ()。
报错里的 {async block@lib.rs:193:9} 指的是某个 stream! 展开出的 async 块。