【问题标题】:How to create a Stream from reading and transforming a file?如何通过读取和转换文件来创建流?
【发布时间】:2020-03-26 21:30:20
【问题描述】:

我正在尝试读取文件、解密文件并返回数据。由于该文件可能非常大,我想在流中执行此操作。

我找不到实现流的好模式。我正在尝试做这样的事情:

let stream = stream::unfold(decrypted_init_length, |decrypted_length| async move {
    if decrypted_length < start + length {
        let mut encrypted_chunk = vec![0u8; encrypted_block_size];
        match f.read(&mut encrypted_chunk[..]) {
            Ok(size) => {
                if size > 0 {
                    let decrypted = my_decrypt_fn(&encrypted_chunk[..]);
                    let updated_decrypted_length = decrypted_length + decrypted.len();
                    Some((decrypted, updated_decrypted_length))
                } else {
                    None
                }
            }
            Err(e) => {
                println!("Error {}", e);
                None
            }
        }
    } else {
        None
    }
});

问题是上述异步闭包中不允许f.read出现以下错误:

89  | |             match f.read(&mut encrypted_chunk[..]) {
    | |                   -
    | |                   |
    | |                   move occurs because `f` has type `std::fs::File`, which does not implement the `Copy` trait
    | |                   move occurs due to use in generator

我不想在闭包内部打开f。有没有更好的方法来解决这个问题?我可以使用不同的 crate 或 trait 或方法(即不是 stream::unfold)。

【问题讨论】:

  • 您使用的是异步文件类型吗? (例如async-std 或 tokio)。
  • @squiguy 不,我使用的是常规文件类型,即std::fs::File。
  • 很难回答您的问题,因为它不包含minimal reproducible example。我们无法分辨代码中存在哪些 crate(及其版本)、类型、特征、字段等。如果您尝试在Rust Playground 上重现您的错误,如果可能的话,这将使我们更容易为您提供帮助,否则在一个全新的 Cargo 项目中,然后在edit 您的问题中包含附加信息。您可以使用Rust-specific MRE tips 来减少您在此处发布的原始代码。谢谢!

标签: rust rust-tokio hyper


【解决方案1】:

我找到了解决方案:在here 使用async-stream crate。

stream::unfold 对我不起作用的原因之一是 async move 闭包不允许访问外部的 mut 变量,例如 f 文件句柄。

现在使用async-stream,我将代码更改为以下内容,并且它可以工作:(注意此板条箱添加的yield)。

use async_stream::try_stream;

<snip>

    try_stream! {
        while decrypted_length < start + length {
            match f.read(&mut encrypted_chunk[..]) {
                Ok(size) => 
                    if size > 0 {
                        println!("read {} bytes", size);
                        let decrypted = my_decrypt_fn(&encrypted_chunk[..size], ..);
                        decrypted_length = decrypted_length + decrypted.len();
                        yield decrypted;
                    } else {
                        break
                    }
                Err(e) => {
                    println!("Error {}", e);
                    break
                }
            }
        }
    }

更新:

我发现async-stream 有一些我不能忽视的限制。我最终直接实现了Stream,不再使用async-stream。现在我的代码如下所示:

pub struct DecryptFileStream {
    f: File,
    <other_fields>,
}

impl Stream for DecryptFileStream {
    type Item = io::Result<Vec<u8>>;

    fn poll_next(self: Pin<&mut Self>,
                  _cx: &mut Context<'_>) -> Poll<Option<io::Result<Vec<u8>>>> {
         // read the file `f` of self and business_logic
         // 
         if decrypted.len() > 0 {
             Poll::Ready(Some(Ok(decrypted)))
         } else {
             Poll::Ready(None)
         }
    }
}

//. then use the above stream: 

    let stream = DecryptFileStream::new(...);
    Response::new(Body::wrap_stream(stream))

【讨论】:

    【解决方案2】:

    stream::unfold 仅适用于实现 Stream 的类型,在 Rust 中专门用于异步编程。如果你想做同步读取,你所谓的“流”被标记为在 Rust 中实现 Read。因此,您可以调用Read::read() 从File 的当前位置读取一些数据(受您传入的缓冲区长度的限制),然后解密该数据。

    【讨论】:

    • 虽然我是用std::fs::File来读取chunk的,但是更高级的逻辑是异步编程。在这种情况下,我使用hyper crate,我想为hyper 响应正文创建一个流。
    • 所以您想获取一个加密文件并对其应用Body::wrap_stream()?在这种情况下,您可能想使用tokio::fs::File 及其AsyncRead impl。不过,如果您按照 Shepmaster 的建议澄清您的问题,这可能会有所帮助。
    • 是的,一旦我从文件中得到一个流,我将使用Body::wrap_stream()。我现在找到了一个很好的板条箱。将详细信息作为单独的答案发布。
    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2017-01-17
    • 1970-01-01
    • 1970-01-01
    • 2021-12-14
    • 1970-01-01
    相关资源
    最近更新 更多