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

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方法，需要 aStream作为参数。如何创建一个可以逐块添加块的流？

Answer 1

根据您的工作流程（数据块有多大、解密耗时等），您可能对如何制作流有不同的选择。我想到的最合法的方法是使用某种带有通道的线程池在线程和处理程序之间进行通信。Tokio 的mpsc可以是这种情况下的一个选项，它的Receiver已经实现了 Stream，您可以通过使用线程中的Sender的 try_send 从线程中提供它，考虑到您使用无界通道或具有足够长度的有界通道，它应该工作。

如果您的解密过程不那么耗时而被视为阻塞，或者您只是想了解如何为 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"] 作为依赖项才能工作。