如何使用 Actix 客户端将 multipart/form-data 流从客户端传递到第三方服务器？

如何解决如何使用 Actix 客户端将 multipart/form-data 流从客户端传递到第三方服务器？

我正在创建自己的前端客户端以将视频上传到视频托管服务。 根据 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"}。

解决方法

我想出了一个技术上可行的解决方案，但这并不是我最初想要的，因为我更喜欢转发 multipart/form-data 流，而不是将流的文件部分写入本地文件，这可以轻松地大规模占用可用内存。

虽然我是在回答我自己的问题，但我希望得到更多的意见来填补我遗漏的空白。

解决方案：

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"