rust - 如何为 actix-web HttpResponse 创建流以逐 block 发送文件？

Question

我想在 Rust 中使用 actix-web 流式传输加密文件。我有一个循环使用氧化钠逐块解密加密的文件块。我想将块发送给客户端。

我的循环看起来像这样:

while stream.is_not_finalized() {
    match in_file.read(&mut buffer) {
        Ok(num_read) if num_read > 0 => {
            let (decrypted, _tag) = stream
                .pull(&buffer[..num_read], None)
                .map_err(|_| error::ErrorInternalServerError("Incorrect password"))
                .unwrap();

            // here I want to send decrypted to HttpResponse
            continue;
        }
        Err(e) => error::ErrorInternalServerError(e),
        _ => error::ErrorInternalServerError("Decryption error"), // reached EOF
    };
}

我找到了一个 streaming 方法，需要一个 Stream作为参数。如何创建一个可以逐块添加的流？

Answer 1

根据您的工作流程(您的数据块有多大，解密的时间有多长等)，您可能对如何制作流有不同的选择。我想到的最合法的方法是使用某种带有 channel 的线程池来在线程和处理程序之间进行通信。东京mpsc可以是这种情况下的一个选项及其 Receiver已经实现了 Stream，您可以使用它的 Sender 从您的线程中提供它的 try_send 来自您的线程，考虑到您使用无界 channel 或具有足够长度的有界 channel ，它应该可以工作。
另一个可能的选择是，当您的解密过程不那么耗时而被视为阻塞时，或者您只想了解如何为 actix 实现 Stream，这是一个示例:

use std::pin::Pin;
use std::task::{Context, Poll};

use actix_web::{get, App, HttpResponse, HttpServer, Responder};
use pin_project::pin_project;
use sodiumoxide::crypto::secretstream::{Pull, Stream};
use tokio::{fs::File, io::AsyncRead};

#[pin_project]
struct Streamer {
    crypto_stream: Stream<Pull>,
    #[pin]
    file: File,
}

impl tokio::stream::Stream for Streamer {
    type Item = Result<actix_web::web::Bytes, actix_web::Error>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.project();

        let mut buffer = [0; BUFFER_LENGTH];

        if this.crypto_stream.is_not_finalized() {
            match this.file.poll_read(cx, &mut buffer) {
                Poll::Ready(res) => match res {
                    Ok(bytes_read) if bytes_read > 0 => {
                        let value = this.crypto_stream.pull(&buffer, None);
                        match value {
                            Ok((decrypted, _tag)) => Poll::Ready(Some(Ok(decrypted.into()))),
                            Err(_) => Poll::Ready(Some(Err(
                                actix_web::error::ErrorInternalServerError("Incorrect password"),
                            ))),
                        }
                    }
                    Err(err) => {
                        Poll::Ready(Some(Err(actix_web::error::ErrorInternalServerError(err))))
                    }
                    _ => Poll::Ready(Some(Err(actix_web::error::ErrorInternalServerError("Decryption error")))),
                },
                Poll::Pending => Poll::Pending,
            }
        } else {
            // Stream finishes when it returns None
            Poll::Ready(None)
        }
    }
}

并从您的处理程序中使用它:

let in_file = File::open(FILE_NAME).await?;
let stream = Stream::init_pull(&header, &key)?;

let stream = Streamer {
    crypto_stream: stream,
    file: in_file,
};
HttpResponse::Ok()
//    .content_type("text/text")
    .streaming(stream)

请注意，您需要 pin_project 和 tokio 以 ["stream", "fs"] 作为依赖项才能工作。