Rust中Hyper从0.14迁移至1.x的流式响应实现问题

问题分析 #

你遇到的类型不匹配错误，核心原因是错误使用了Stream的装箱方法：

• 你调用的body.boxed()是Stream trait的方法，返回Pin<Box<dyn Stream<Item = Bytes> + Send>>，但Hyper的Response要求的是实现Body trait的类型
• 目标返回类型BoxBody<Bytes, Infallible>是Hyper定义的Body trait的装箱实现，需要从StreamBody通过Body trait的方法转换而来

修正后的实现代码 #

方案1：忽略序列化错误（保持Infallible） #

适合确定序列化不会失败的场景，直接panic处理错误：

use hyper::{Response, StatusCode, header::CONTENT_TYPE, body::{BoxBody, StreamBody}};
use futures::StreamExt;
use bytes::Bytes;
use serde::Serialize;

fn response_stream<T, P>(stream: T) -> Response<BoxBody<Bytes, std::convert::Infallible>>
where
    T: futures::Stream<Item = P> + Send + 'static,
    P: Serialize + 'static,
{
    // 序列化每个元素并添加换行符，转换为Bytes
    let stream = stream.map(|item| {
        let json_str = serde_json::to_string(&item).expect("Failed to serialize item");
        Bytes::from(format!("{json_str}\n"))
    });

    // 将Stream包装为Body，再转换为BoxBody
    let body = StreamBody::new(stream).into_boxed_body();

    Response::builder()
        .status(StatusCode::OK)
        .header(CONTENT_TYPE, "application/jsonlines")
        .body(body)
        .expect("Failed to build Response")
}

方案2：优雅处理序列化错误（返回Hyper错误） #

更健壮的实现，将序列化错误转换为Hyper的错误类型，避免panic：

use hyper::{Response, StatusCode, header::CONTENT_TYPE, body::{BoxBody, StreamBody}, Error as HyperError};
use futures::StreamExt;
use bytes::Bytes;
use serde::Serialize;

fn response_stream<T, P>(stream: T) -> Response<BoxBody<Bytes, HyperError>>
where
    T: futures::Stream<Item = P> + Send + 'static,
    P: Serialize + 'static,
{
    // 处理序列化错误，转换为HyperError
    let stream = stream.map(|item| {
        serde_json::to_string(&item)
            .map(|s| Bytes::from(format!("{s}\n")))
            .map_err(|serde_err| HyperError::from(serde_err))
    });

    let body = StreamBody::new(stream).into_boxed_body();

    Response::builder()
        .status(StatusCode::OK)
        .header(CONTENT_TYPE, "application/jsonlines")
        .body(body)
        .expect("Failed to build Response")
}

关键迁移点说明 #

1. 替代Body::wrap_stream：Hyper 1.x中用StreamBody::new(stream)将Stream转换为Body实现
2. 正确装箱Body：使用into_boxed_body()（Body trait的方法）得到BoxBody，而非Stream的boxed()方法
3. 错误类型适配：
   • 若要返回Infallible错误类型，需确保序列化过程无错误（用expect/unwrap）
   • 若要处理错误，需将自定义错误（如serde_json::Error）转换为Hyper兼容的错误类型（如HyperError）

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