相关疑难解决方法(0)

是否可以循环 a Vec，调用一个返回 a 的方法Future，并构建一个Futures链，由消费者（最终）进行评估？是否执行后Futures就依赖于早期的结果Future在S Vec。

澄清：

我正在开发一个可以从任意一组上游源获取数据的应用程序。

请求数据将依次检查每个来源。如果第一个来源有错误 ( Err)，或没有可用的数据 ( None)，则将尝试第二个来源，依此类推。

每个源应该只尝试一次，并且在之前的所有源都返回其结果之前不应尝试任何源。错误被记录，但否则被忽略，将查询传递到下一个上游数据源。

我有一些用于获取元数据的工作代码：

/// Attempts to read/write data to various external sources. These are
/// nested types, because a data source may exist as both a reader and a writer
struct StoreManager {
    /// Upstream data sources
    readers: Vec<Rc<RefCell<StoreRead>>>,
    /// Downstream data sinks
    writers: Vec<Rc<RefCell<StoreWrite>>>,
}

impl StoreRead for StoreManager {
    fn metadata(self: &Self, id: &Identifier) …

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

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 …

我想采用常规迭代器并将其转换为流,以便我可以进行进一步的流处理.麻烦的是我可能有一个迭代器或错误来处理.我觉得我很接近这个:

#[macro_use]
extern crate log;
extern crate futures; // 0.1.21
extern crate tokio;

use futures::prelude::*;
use futures::{future, stream};
use std::fmt::Debug;
use std::net::{SocketAddr, ToSocketAddrs};

fn resolve(addrs: impl ToSocketAddrs + Debug) -> impl Stream<Item = SocketAddr, Error = ()> {
    match addrs.to_socket_addrs() {
        Ok(iter) => stream::unfold(iter, |iter| match iter.next() {
            Some(a) => Some(future::ok((a, iter))),
            None => None,
        }),
        Err(e) => {
            error!("could not resolve socket addresses {:?}: {:?}", addrs, e);
            stream::empty()
        }
    }
}

fn main() {
    let task = …