rust - 如何在Rust中将字节迭代器转换为流

Question

我正在尝试构建一个功能，该功能需要将文件的内容读取为futures::stream::BoxStream，但是我很难确定要做什么。

我已经弄清楚了如何通过字节来逐字节读取文件，该字节实现了迭代器。

use std::fs::File;
use std::io::prelude::*;
use std::io::{BufReader, Bytes};

// TODO: Convert this to a async Stream
fn async_read() -> Box<dyn Iterator<Item = Result<u8, std::io::Error>>> {
    let f = File::open("/dev/random").expect("Could not open file");
    let reader = BufReader::new(f);

    let iter = reader.bytes().into_iter();
    Box::new(iter)
}

fn main() {
    ctrlc::set_handler(move || {
        println!("received Ctrl+C!");
        std::process::exit(0);
    })
    .expect("Error setting Ctrl-C handler");

    for b in async_read().into_iter() {
        println!("{:?}", b);
    }
}

但是，我一直在努力寻找如何将 Box<dyn Iterator<Item = Result<u8, std::io::Error>>>转换为 Stream的方法。

我本以为这样的事情会起作用:

use futures::stream;
use std::fs::File;
use std::io::prelude::*;
use std::io::{BufReader, Bytes};

// TODO: Convert this to a async Stream
fn async_read() -> stream::BoxStream<'static, dyn Iterator<Item = Result<u8, std::io::Error>>> {
    let f = File::open("/dev/random").expect("Could not open file");
    let reader = BufReader::new(f);

    let iter = reader.bytes().into_iter();
    std::pin::Pin::new(Box::new(stream::iter(iter)))
}

fn main() {
    ctrlc::set_handler(move || {
        println!("received Ctrl+C!");
        std::process::exit(0);
    })
    .expect("Error setting Ctrl-C handler");

    while let Some(b) = async_read().poll() {
        println!("{:?}", b);
    }
}

但是我不断收到大量的编译器错误，我尝试了其他排列方式，但通常没有结果。

编译器错误之一:

std::pin::Pin::new
``` --> src/main.rs:14:24
   |
14 |     std::pin::Pin::new(Box::new(stream::iter(iter)))
   |                        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^ expected trait object `dyn std::iter::Iterator`, found enum `std::result::Result`

有人有什么建议吗？

我对Rust还是很陌生，特别是Streams/较低级别的东西，因此，如果我做错了什么，我深表歉意，请随时纠正我。

对于其他背景，我正在尝试这样做，以便您可以 CTRL-C out of a command in nushell

Answer 1

我认为您有点复杂，可以只从impl Stream返回async_read，而无需装箱或固定(基于Iterator的原始版本也是如此)。然后，您需要设置一个异步运行时以轮询流(在本示例中，我仅使用futures::executor::block_on提供的运行时)。然后，您可以在流上调用futures::stream::StreamExt::next()以获得代表下一项的 future 。

这是执行此操作的一种方法:

use futures::prelude::*;
use std::{
    fs::File,
    io::{prelude::*, BufReader},
};

fn async_read() -> impl Stream<Item = Result<u8, std::io::Error>> {
    let f = File::open("/dev/random").expect("Could not open file");
    let reader = BufReader::new(f);
    stream::iter(reader.bytes())
}

async fn async_main() {
    while let Some(b) = async_read().next().await {
        println!("{:?}", b);
    }
}

fn main() {
    ctrlc::set_handler(move || {
        println!("received Ctrl+C!");
        std::process::exit(0);
    })
    .expect("Error setting Ctrl-C handler");

    futures::executor::block_on(async_main());
}