? 之后流就结束了
主线位置：宏展开一段里
?的那条支线。Scrub 把每个?改写成match：Ok取出值；Err就send(Err(e.into())).await，然后return。这一步不看是哪个宏，下面两个坑都从这里来。
flowchart TB
q["expr?"] --> m{"Ok 还是 Err"}
m -->|"Ok(v)"| v["取出 v，接着往下"]
m -->|"Err(e)"| s["send(Err(e.into())).await"]
s --> r["return：async 块结束"]
r --> d["poll_next：先交出 Err，下一次 None"]
s -.-> t["所以流的 Item 必须是 Result"]
依赖与 编译报错 相同。
expected Result<_, _>, found i32（E0308）
stream! 里写了 ?，yield 的却是普通的值：
use async_stream::stream;
use futures::StreamExt;
#[tokio::main]
async fn main() {
// Item 是 i32，但块里用了 `?`
let s = stream! {
for t in ["1", "2"] {
let v: i32 = t.parse()?;
yield v;
}
};
let items: Vec<i32> = s.collect().await;
println!("{items:?}");
}
error[E0308]: mismatched types
--> src/bin/q_item.rs:10:13
|
7 | let s = stream! {
| ___________-
8 | | for t in ["1", "2"] {
9 | | let v: i32 = t.parse()?;
10 | | yield v;
| | ^ expected `Result<_, _>`, found `i32`
11 | | }
12 | | };
| |___- arguments to this method are incorrect
|
= note: expected enum `Result<_, _>`
found type `i32`
help: the return type of this call is `i32` due to the type of the argument passed
...
note: method defined here
--> .../async-stream-0.3.6/src/yielder.rs:42:12
|
42 | pub fn send(&mut self, value: T) -> impl Future<Output = ()> {
| ^^^^
...
help: try wrapping the expression in a variant of `Result`
|
10 | yield Ok(v);
| +++ +
报错指着 yield v，说期望的是 Result，原因却在上一行的 ?。改写之后，块里有两处调用同一个 __yield_tx.send(..)：? 改出来的 send(Err(e.into()))，和 yield v 改出来的 send(v)。Sender<T> 只有一个 T，前者把它定成了 Result<_, _>，后者传进来一个 i32，于是类型不匹配。报错的 note 也指到了 yielder.rs 第 42 行的 send，这是改写痕迹露出来的地方。
改法有两个，结果一样：
- 照编译器的建议写
yield Ok(v);，流的Item是Result<i32, ParseIntError>； - 换成
try_stream!，yield v;由宏包成Ok(v)。
Scrub 里实测过两种写法对同一输入产出相同的 [Ok(1), Err(..)]。stream! 和 try_stream! 的区别只在 yield 要不要你自己写 Ok，? 的行为没有区别。同理，? 用在 Option 上也不行，改写写死了 Result::Ok / Result::Err，报错原文同样在 Scrub 那一章。
第一个 Err 之后流就结束了
这个坑没有报错，只有现象：流比预期的短。
use std::num::ParseIntError;
use async_stream::{stream, try_stream};
use futures::{Stream, StreamExt};
const INPUT: [&str; 4] = ["1", "x", "3", "y"];
// `?`：第一个 Err 作为最后一项发出，流随即结束
fn with_q() -> impl Stream<Item = Result<i32, ParseIntError>> {
try_stream! {
for t in INPUT {
println!(" [with_q] parsing {t:?}");
let v: i32 = t.parse()?;
yield v;
}
println!(" [with_q] after loop");
}
}
// 自己 match：错误也作为一项发出，然后接着处理下一项
fn keep_going() -> impl Stream<Item = Result<i32, ParseIntError>> {
stream! {
for t in INPUT {
match t.parse::<i32>() {
Ok(v) => yield Ok(v),
Err(e) => yield Err(e),
}
}
}
}
#[tokio::main]
async fn main() {
let a: Vec<_> = with_q().collect().await;
println!("with_q: {a:?}");
let b: Vec<_> = keep_going().collect().await;
println!("keep_going: {b:?}");
}
输出（跑 3 次相同）：
[with_q] parsing "1"
[with_q] parsing "x"
with_q: [Ok(1), Err(ParseIntError { kind: InvalidDigit })]
keep_going: [Ok(1), Err(ParseIntError { kind: InvalidDigit }), Ok(3), Err(ParseIntError { kind: InvalidDigit })]
with_q 解析到 "x" 就停了："3"、"y" 根本没被解析，循环后面的 after loop 也没打印。? 改写里的 return 退出的是整个 async 块，不是这一轮循环。poll_next 先把槽里的 Err 交出去，下一次 poll 时块已经完成，返回 None（poll_next）。消费者看到的就是「一个 Err，然后流结束」。
keep_going 把每个错误也作为一项发出，然后接着处理下一项，四个输入产出四项。
? 在流里等于「这个错误是致命的」。这在有些场景正是想要的，有些场景则会悄悄丢数据：
| 场景 | ? 合不合适 |
|---|---|
sea-orm 的 into_stream 分页读：一页查询失败 | 合适。后面的页也查不了，结束是对的（这段代码见 05 在本站中） |
| 逐行解析一个文件或一段响应：某一行格式不对 | 看需求。要整体失败就用 ?；要跳过坏行，就自己 match，yield Err(e) 或只记日志 |
accept 循环：某一次 accept 失败 | 要小心。async-stream 自己的文档示例就是 listener.accept().await? 后 yield stream，照这个写，一次 accept 出错就结束了整个监听流。accept 的哪些错误是暂时的（比如文件描述符用完），本章没有实测 |
| SSE 推送：某一步失败 | Err 会让 axum 中断这个响应，客户端只看到连接断开，看不到错误内容。实测和改法见 哪里可以用 |
用户自己写的 return; 效果相同，也会提前结束流。try_stream! 的名字容易让人以为它对错误有特别的处理，其实只多了「yield 自动包 Ok」这一点。
小结
- 两个宏里的
?都被改写成「Err时send(Err(e.into())).await然后return」，和是stream!还是try_stream!无关。 stream!里写了?，Sender<T>的T就被定成Result，yield v传i32会报expected `Result<_, _>`, found `i32`，报错指着yield而不是?。写yield Ok(v)或换成try_stream!。?之后是return，退出整个 async 块：实测输入["1", "x", "3", "y"]只产出[Ok(1), Err(..)]。错误不致命时自己match，把错误也作为一项发出，然后接着处理。accept循环、SSE 推送这类长期运行的流，用?前先想清楚「这个错误该不该结束整个流」。