如何从未来的功能中实现流

Question

为了理解流是如何工作的，我试图实现一个使用 random.org 的无限数生成器。我做的第一件事是实现一个版本，我会调用一个名为 get_number 的异步函数，它会填充一个缓冲区和 return 下一个可能的数字：


struct RandomGenerator {
    buffer: Vec<u8>,
    position: usize,
}

impl RandomGenerator {
    pub fn new() -> RandomGenerator {
        Self {
            buffer: Vec::new(),
            position: 0,
        }
    }

    pub async fn get_number(&mut self) -> u8 {
        self.fill_buffer().await;

        let value = self.buffer[self.position];
        self.position += 1;

        value
    }

    async fn fill_buffer(&mut self) {
        if self.buffer.is_empty() || self.is_buffer_depleted() {
            let new_numbers = self.fetch_numbers().await;
            drop(replace(&mut self.buffer, new_numbers));
            self.position = 0;
        }
    }

    fn is_buffer_depleted(&self) -> bool {
        self.buffer.len() >= self.position
    }

    async fn fetch_numbers(&mut self) -> Vec<u8> {
        let response = reqwest::get("https://www.random.org/integers/?num=10&min=1&max=100&col=1&base=10&format=plain&rnd=new").await.unwrap();
        let numbers = response.text().await.unwrap();
        numbers
            .lines()
            .map(|line| line.trim().parse::<u8>().unwrap())
            .collect()
    }
}

通过这个实现，我可以在循环中调用函数 get_number 并获得我想要的任意数量的数字，但我的想法是使用迭代器，这样我就可以调用一堆组合函数，例如 take、take_while，以及其他。

但是当我尝试实现 Stream 时，问题开始出现：我的第一个尝试是拥有一个包含对生成器的引用的结构

struct RandomGeneratorStream<'a> {
    generator: &'a mut RandomGenerator,
}

然后我实现了以下 Stream

impl<'a> Stream for RandomGeneratorStream<'a> {
    type Item = u8;

    fn poll_next(
        self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        let f = self.get_mut().generator.get_number();
        pin_mut!(f);
        f.poll_unpin(cx).map(Some)
    }
}

但是调用它只会挂起进程

generator.into_stream().take(18).collect::<Vec<u8>>().await

在接下来的尝试中，我尝试使用 pin_mut 在流结构上保存未来状态！但最终在无法解决的情况下遇到了许多生命周期错误。在这种情况下可以做什么？这是没有流的工作代码：

use std::mem::replace;
struct RandomGenerator {
    buffer: Vec<u8>,
    position: usize,
}

impl RandomGenerator {
    pub fn new() -> RandomGenerator {
        Self {
            buffer: Vec::new(),
            position: 0,
        }
    }

    pub async fn get_number(&mut self) -> u8 {
        self.fill_buffer().await;

        let value = self.buffer[self.position];
        self.position += 1;

        value
    }

    async fn fill_buffer(&mut self) {
        if self.buffer.is_empty() || self.is_buffer_depleted() {
            let new_numbers = self.fetch_numbers().await;
            drop(replace(&mut self.buffer, new_numbers));
            self.position = 0;
        }
    }

    fn is_buffer_depleted(&self) -> bool {
        self.buffer.len() >= self.position
    }

    async fn fetch_numbers(&mut self) -> Vec<u8> {
        let response = reqwest::get("https://www.random.org/integers/?num=10&min=1&max=100&col=1&base=10&format=plain&rnd=new").await.unwrap();
        let numbers = response.text().await.unwrap();
        numbers
            .lines()
            .map(|line| line.trim().parse::<u8>().unwrap())
            .collect()
    }
}

#[tokio::main]
async fn main() {
    let mut generator = RandomGenerator::new();
    dbg!(generator.get_number().await);
}

在这里您可以找到第一个工作示例的 link（而不是调用 random.org 我使用了游标，因为 dns 解析在操场上不起作用）https://play.rust-lang.org/?version=stable&mode=debug&edition=2018&gist=730eaf1f7db842877d3f3e7ca1c6d2a5

你可以在这里找到我最后一次使用流的尝试 https://play.rust-lang.org/?version=stable&mode=debug&edition=2018&gist=de0b212ee70865f6ac6c19430cd952cd

Answer 1

On the next tries, I tried to hold a state of the future on the stream struct using pin_mut! but ended up having many errors with lifetimes without being able to solve them.

你的方向是正确的，你需要坚持未来才能poll_next正常工作。

不幸的是，您将运行陷入具有可变引用的障碍。你保留一个 &mut RandomGenerator 以便重复使用它，但未来本身也必须保留一个 &mut RandomGenerator 才能完成它的工作。这将违反可变引用的排他性。无论如何切割它都可能会遇到这个问题。

从 Future 到 Stream 的更好方法是遵循建议 and use futures::stream::unfold:

fn as_stream<'a>(&'a mut self) -> impl Stream<Item = u8> + 'a {
    futures::stream::unfold(self, |rng| async {
        let number = rng.get_number().await;
        Some((number, rng))
    })
}

在 playground 上查看。

这不一定能帮助您了解更多有关流的信息，但提供的功能通常比手动滚动更好。这避免了上面的多重可变引用问题的关键原因是因为生成未来的函数 取得了可变引用的所有权 ，然后在完成后将其归还。这样一次只有一个存在。即使您自己实现了 Stream，您也必须使用类似的机制。

如何从未来的功能中实现流

How to implement streams from future functions

rust

rust-tokio

rust-async-std