如何使用Actix客户端将多部分/表单数据流从客户端传递到第三方服务器？

Question

我正在创建自己的前端客户端上传视频到视频托管服务。根据API文档，我将发布一个多部分/表单：

curl -X POST \
     -H "Key: YOUR_KEY" \
     -F "file=@hello_world.mp4" \
     https://muse.ai/api/files/upload

我可以设置我的客户端可以进行跨源请求，但是请求需要一个秘密的API密钥，所以我需要我的服务器代表我的客户机代理请求。

最低限度如下：

use actix_web::{Error, HttpResponse, client::Client, error, web, HttpRequest};
use std::env;

pub async fn upload_video(req: HttpRequest, payload: web::Payload) -> Result<HttpResponse, Error> {
    let muse_api_key = match env::var("MUSE_AI") {
        Ok(token) => token,
        Err(e) => {
            return Err(error::ErrorInternalServerError(e));
        }
    };
    
    let client = Client::default();
    let mut forward_req_resp = client
      .request_from("https://muse.ai/api/files/upload", req.head())
      .header("key", muse_api_key)
      .send_stream(payload)
      .await?;

    let mut client_resp = HttpResponse::build(forward_req_resp.status());

    Ok(client_resp.body(forward_req_resp.body().await?))
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(move || {
        App::new()
        .service(
            web::scope("/video")
                .service(
                    web::resource("/upload").route(web::post().to(upload_video)),
                )
        )
    })
    .bind("127.0.0.1:8001")?
    .run()
    .await
}

在连接我的前端之前，我正在用cURL进行测试：

curl -v -X POST -F "file=@/path/to/sintel_trailer-720p.mp4" 127.0.0.1:8001/video/upload

结果产出：

* TCP_NODELAY set
* Connected to 127.0.0.1 (127.0.0.1) port 8001 (#0)
> POST /video/upload HTTP/1.1
> Host: 127.0.0.1:8001
> User-Agent: curl/7.64.1
> Accept: */*
> Content-Length: 7608419
> Content-Type: multipart/form-data; boundary=------------------------076c45bb618a62c2
> Expect: 100-continue
> 
< HTTP/1.1 100 Continue
< HTTP/1.1 400 Bad Request
< content-length: 40
< content-type: text/plain; charset=utf-8
< set-cookie: actix-session=3JVldgobuv9nv6zE9uV%2F1i0slQMHlV384rXzc9BWwRE%3D%7B%7D; HttpOnly; Path=/
< date: Mon, 01 Feb 2021 07:32:14 GMT
* HTTP error before end of send, stop sending
< 
* Closing connection 0
protocol error: not a result of an error% 

我尝试了几个变体，其中一个值得注意:将以下内容添加到处理程序函数的顶部：

    let mut bytes = web::BytesMut::new();
    while let Some(item) = payload.next().await {
        bytes.extend_from_slice(&item?);
    }

然后将ClientRequestBuilding从.request_from更改为：

.post("https://muse.ai/api/files/upload")

结果是：{ error: "missing_file"}。

Answer 1

我想出了一个技术上可行的解决方案，但这并不是我最初想要的，因为我倾向于沿着多部分/表单数据流转发，而不是将流的文件部分写入本地文件，这样就可以轻松地占用可用的内存。

虽然我正在回答我自己的问题，但我希望能有更多的投入来填补我所缺少的空白。

解决方案：

use reqwest::blocking;
use actix_multipart::Multipart;

pub async fn upload_video(mut payload: Multipart) -> Result<HttpResponse, Error> {
    let muse_api_key = match env::var("MUSE_AI") {
        Ok(token) => token,
        Err(e) => {
            return Err(error::ErrorInternalServerError(e));
        },
    };

    let mut filepath_copy = String::from("");

    while let Ok(Some(mut field)) = payload.try_next().await {
        let content_type = field.content_disposition().unwrap();
        let filename = content_type.get_filename().unwrap();
        let filepath = format!("./{}", sanitize_filename::sanitize(&filename));
        filepath_copy = format!("./{}", sanitize_filename::sanitize(&filename));

        let mut f = web::block(|| std::fs::File::create(filepath))
            .await?;

        while let Some(chunk) = field.next().await {
            let data = chunk.unwrap();
            f = web::block(move || f.write_all(&data).map(|_| f)).await?;
        }
    }

    let client = blocking::Client::new();

    let form = blocking::multipart::Form::new()
      .file("file", filepath_copy.clone())?;

    let forward_req_resp = web::block(move ||
      client
        .post("https://muse.ai/api/files/upload")
        .header("key", muse_api_key)
        .multipart(form)
        .send()
    ).await?;
    
    web::block(|| fs::remove_file(filepath_copy)).await?;

    let status = forward_req_resp.status();

    Ok(HttpResponse::build(status).finish())
}

依赖性(用于此处理程序，而不是整个应用程序)：

futures = "0.3.12"
sanitize-filename = "0.3.0"
reqwest = { version = "0.11.0", features = ["multipart", "blocking", "json"] }
actix-multipart = "0.3.0"