在reqwest创建的流中使用try_for_each_concurrent无法返回自定义错误

处理这类跨crate错误不兼容的问题，Rust里最惯用的方式是定义自己的错误枚举类型，把所有可能出现的错误（包括reqwest的错误和你的自定义错误）都包含进去，通过实现From trait让不同错误类型能自动转换为你的自定义错误。这样就能统一流的错误类型，让try_for_each_concurrent正常工作。

步骤示例 #

1. 定义自定义错误（推荐用thiserror简化实现） #

首先在Cargo.toml里添加依赖：

[dependencies]
reqwest = { version = "0.11", features = ["stream"] }
tokio = { version = "1.0", features = ["full"] }
thiserror = "1.0"

然后定义错误枚举：

use thiserror::Error;

#[derive(Error, Debug)]
enum MyError {
    #[error("Reqwest error: {0}")]
    Reqwest(#[from] reqwest::Error),
    
    #[error("Transformation error: {0}")]
    Transform(#[from] MyTransformError),
}

// 你的自定义错误类型
#[derive(Error, Debug)]
#[error("Failed to transform bytes: {0}")]
struct MyTransformError(String);

2. 转换流的错误类型并处理 #

把原流的错误转换成自定义错误，之后闭包里的?就能自动把MyTransformError转换成MyError：

use reqwest::Body;
use tokio_stream::StreamExt; // try_for_each_concurrent 依赖 tokio_stream

#[tokio::main]
async fn main() -> Result<(), MyError> {
    let stream = reqwest::get("http://httpbin.org/ip")
        .await?
        .bytes_stream()
        .map_err(MyError::from); // 将 reqwest::Error 转换为自定义错误

    let result = stream.try_for_each_concurrent(2, |chunk| async move {
        // 调用可能返回自定义错误的函数
        let transformed = transform_chunk(chunk)?;
        // 处理转换后的内容
        println!("Processed chunk: {:?}", transformed);
        Ok(())
    }).await;

    result
}

fn transform_chunk(chunk: bytes::Bytes) -> Result<String, MyTransformError> {
    // 模拟转换失败场景
    if chunk.is_empty() {
        return Err(MyTransformError("Empty chunk".into()));
    }
    String::from_utf8(chunk.to_vec()).map_err(|e| MyTransformError(e.to_string()))
}

替代方案：直接包装错误到reqwest::Error #

如果你不想定义自定义错误，也可以用reqwest::Error的构造函数把自定义错误包装进去：

let stream = reqwest::get("http://httpbin.org/ip")
    .await?
    .bytes_stream();

let result = stream.try_for_each_concurrent(2, |chunk| async move {
    let transformed = transform_chunk(chunk)
        .map_err(|e| reqwest::Error::new(reqwest::ErrorKind::Other, e.to_string()))?;
    println!("Processed chunk: {:?}", transformed);
    Ok(())
}).await;

不过这种方法不够规范，会把非reqwest相关的错误塞进reqwest的错误类型里，长期维护会增加混乱，更推荐自定义错误的方式。

为什么不用重新实现流？ #

你提到的重新实现流的方法确实可行，但完全没必要——Rust的流处理已经提供了map_err这类适配器，可以轻松转换错误类型，不需要从头实现新流。这也是异步流处理的惯用技巧：通过适配器修改流的项或错误类型。

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