依赖树里的 async-stream
主线位置：主线在别人代码里的两个实例。sea-orm 用
stream!把「一页一页查」变成流，dioxus-fullstack 用try_stream!把「一行一行读」变成流，正好对应宏展开一段的?改写和运行时一段的惰性执行。
flowchart TB app["本站 app"] --> sea["sea-orm 1.1.20"] app --> dx["dioxus-fullstack 0.7.9"] sea --> as["async-stream 0.3.6"] dx --> as as --> imp["async-stream-impl 0.3.6"] sea --> p["paginator.rs：into_stream"] dx --> s["payloads/sse.rs：FromResponse for ServerEvents"]
框架自己：一处 Stream，没有 stream!
在 ceffc19 的 crates/ 下搜 stream!、impl Stream、Sse、Body::from_stream、text/event-stream，都没有结果。用到 futures::stream 的只有一处，后台的批量拒绝审核：
// Phase 8.8：8 路并发处理。DB pool 上限 32，留一半给前台读路径，
// 仍把批量耗时压缩到 ~1/8。单条失败只告警，继续处理其余。
use futures::stream::{self, StreamExt};
// ...
stream::iter(rows)
.for_each_concurrent(8, |row| {
let done_inc = done_for_stream.clone();
async move {
match reject_one(db_ref, admin_id, row).await {
Ok(()) => {
done_inc.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
Err(e) => tracing::warn!(error = %e, "bulk reject: one entry failed"),
}
}
})
.await;
这里的流只是「把一个 Vec 变成可以限并发消费的东西」，数据早就在内存里了，用 stream::iter 正合适，没有要 yield 的地方。注意它的错误处理：单条失败只记日志、继续处理其余，这正是 ? 之后流就结束了里说的「错误不致命」的写法。
LLM 调用也不是流式的：anthropic.rs 的 complete 发一次请求，.json(&body).send() 之后整体 .text()，openai.rs 也一样，不需要逐段产出。
Cargo.lock：两条来路
Cargo.lock 第 312 行是 async-stream 0.3.6，和本卷的版本一致。依赖它的有两个包：
两个包都把 async-stream 列为必选依赖，不在任何 feature 后面，所以不管本站用不用这两段代码，async-stream 和 async-stream-impl（连同它要的 syn）都会被编译。
sea-orm：into_stream 是 stream! 加 ?
pub fn into_stream(mut self) -> PinBoxStream<'db, Result<Vec<S::Item>, DbErr>> {
Box::pin(stream! {
while let Some(vec) = self.fetch_and_next().await? {
yield Ok(vec);
}
})
}
五行代码，前面几章的规则都在里面：
stream!里写?：fetch_and_next().await?写在while let的条件里，也是块的字面位置，Scrub 照样改写。数据库出错时，Err(DbErr)作为最后一项发出，然后return，流结束。分页读到一半数据库出错，后面的页也查不了，这里用?是对的（? 之后流就结束了）。yield Ok(vec)：因为用的是stream!而不是try_stream!，Ok要自己写。写成yield vec就是 E0308 那条报错。- 返回类型钉死了错误类型：
PinBoxStream<'db, Result<Vec<_>, DbErr>>，?里的e.into()有了目标，不会碰到 E0282。 Box::pin：返回的流已经 pin 好，调用方可以直接try_next()，不会碰到cannot be unpinned。代价是每个流一次堆分配。- 惰性：
stream!只在消费者 poll 时执行，调用方不取下一项，下一页就不查。生命周期'db能出现在返回类型里，也是因为它不spawn。
fetch_and_next 在拿到空页时返回 None，循环结束，这一点在 sea-orm 卷的分页里讲过。
本站用的是分页的另一半：后台用户列表和论坛主题列表都是 paginate 后 fetch_page(page)，按页码取一页给前端，没有人需要「一页接一页地读完」，所以 into_stream 不在本站的运行路径上。
dioxus-fullstack：客户端用 try_stream!，服务端用通道
dioxus 的 server 函数可以返回 ServerEvents<T>，做 SSE 推送。这个类型在两端的实现方式不同，正好是为什么需要它里比较的两条路。
客户端：把 HTTP 响应体变成事件流，payloads/sse.rs 第 91 行 起的 FromResponse for ServerEvents<T>，核心是第 117–157 行：
let stream: Pin<Box<dyn Stream<Item = Result<ServerSentEvent, ServerFnError>>>> = Box::pin(
async_stream::try_stream! {
loop {
line_buffer.clear();
if stream.read_line(&mut line_buffer).await.map_err(|err| ServerFnError::StreamError(err.to_string()))? == 0 {
break;
}
// ...
// dispatch
if line.is_empty() {
if let Some(event) = event_buffer.produce_event() {
yield event;
}
continue;
}
// ...
match field {
"event" => event_buffer.set_event_type(value),
"data" => event_buffer.push_data(value),
// ...
}
}
},
);
- 状态（
line_buffer、event_buffer、读到哪了）全是块里的局部变量，try_stream!替它管状态机；同样的逻辑手写poll_next，要自己保存「正在读的那一行」的 future。 - 读失败用
?：map_err成ServerFnError再?，错误作为最后一项交出，然后流结束。连接断了，后面也读不到东西，结束是对的。 - 读到 EOF（
read_line返回 0）用break跳出循环，块自然结束，流返回None。 - 变量标注了
Pin<Box<dyn Stream<..>>>，同时解决了错误类型推断和 pin 两个问题，和 sea-orm 的做法一样。
服务端：ServerEvents::new 给用户函数一个发送端，第 279 行建的是 futures_channel::mpsc::unbounded()，第 287 行把用户函数 spawn_platform 出去。这是「通道 + spawn」那条路，而且通道是无界的：从代码看，用户函数可以一直发，不会因为客户端读得慢而被挡住；客户端断开后，用户函数要在下一次发送失败时才知道。这一段本卷没有实测。另一个入口 from_stream 接受任意 TryStream<Ok = T, Error = BoxError>，在那里交一个 try_stream! 进去，就是惰性的那条路。
本站没有返回 ServerEvents 的 server 函数，客户端那段 try_stream! 会被编译，但不会被执行。
小结
- 本站框架没有
stream!、没有手写Stream；唯一的流是后台批量审核的stream::iter(..).for_each_concurrent(8, ..)，数据已在内存里，用不上yield。LLM 调用是一次性请求。 Cargo.lock里的 async-stream 0.3.6 是 sea-orm 1.1.20 和 dioxus-fullstack 0.7.9 的必选依赖带进来的。- sea-orm 的
into_stream是stream!+?+yield Ok(vec)+Box::pin，前几章的规则一行不落；本站用fetch_page，没走它。 - dioxus-fullstack 的 SSE 客户端用
try_stream!逐行解析，服务端的ServerEvents::new用无界通道加spawn，from_stream可以接try_stream!。本站没有 SSE，这两段都不在运行路径上。