那些遇到过的问题

如何在 Rust 中迭代返回 Futures 的函数的 Vec？

squ*_*les 3 iterator future rust

是否可以循环 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) -> Box<Future<Option<Metadata>, Error>> {
       Box::new(ok(self.readers
            .iter()
            .map(|store| {
                executor::block_on(store.borrow().metadata(id)).unwrap_or_else(|err| {
                    error!("Error on metadata(): {:?}", err);
                    None
                })
            })
            .find(Option::is_some)
            .unwrap_or(None)))
    }
}
Run Code Online (Sandbox Code Playgroud)

除了我的不满与所有的Box和Rc/RefCell废话，我真正关心的是与executor::block_on()调用。它阻塞，等待每个Future返回结果，然后继续下一个。

鉴于可以调用fn_returning_future().or_else(|_| other_fn())等等，是否可以建立这样的动态链？或者是否需要Future在移动到下一个之前完全评估迭代器中的每个？

She*_*ter 5

您可以使用stream::unfold将单个值转换为流。在这种情况下，我们可以使用IntoIter迭代器作为单个值。

use futures::{executor, stream, Stream, TryStreamExt}; // 0.3.4

type Error = Box<dyn std::error::Error>;
type Result<T, E = Error> = std::result::Result<T, E>;

async fn network_request(val: i32) -> Result<i32> {
    // Just for demonstration, don't do this in a real program
    use std::{
        thread,
        time::{Duration, Instant},
    };
    thread::sleep(Duration::from_secs(1));
    println!("Resolving {} at {:?}", val, Instant::now());

    Ok(val * 100)
}

fn requests_in_sequence(vals: Vec<i32>) -> impl Stream<Item = Result<i32>> {
    stream::unfold(vals.into_iter(), |mut vals| async {
        let val = vals.next()?;
        let response = network_request(val).await;
        Some((response, vals))
    })
}

fn main() {
    let s = requests_in_sequence(vec![1, 2, 3]);
    executor::block_on(async {
        s.try_for_each(|v| async move {
            println!("-> {}", v);
            Ok(())
        })
        .await
        .expect("An error occurred");
    });
}
Run Code Online (Sandbox Code Playgroud) 
Resolving 1 at Instant { tv_sec: 6223328, tv_nsec: 294631597 }
-> 100
Resolving 2 at Instant { tv_sec: 6223329, tv_nsec: 310839993 }
-> 200
Resolving 3 at Instant { tv_sec: 6223330, tv_nsec: 311005834 }
-> 300
Run Code Online (Sandbox Code Playgroud)

要忽略Errand None，您必须将Errorover穿梭到Item，使Item类型 a Result<Option<T>, Error>：

use futures::{executor, stream, Stream, StreamExt}; // 0.3.4

type Error = Box<dyn std::error::Error>;
type Result<T, E = Error> = std::result::Result<T, E>;

async fn network_request(val: i32) -> Result<Option<i32>> {
    // Just for demonstration, don't do this in a real program
    use std::{
        thread,
        time::{Duration, Instant},
    };
    thread::sleep(Duration::from_secs(1));
    println!("Resolving {} at {:?}", val, Instant::now());

    match val {
        1 => Err("boom".into()),  // An error
        2 => Ok(None),            // No data
        _ => Ok(Some(val * 100)), // Success
    }
}

fn requests_in_sequence(vals: Vec<i32>) -> impl Stream<Item = Result<Option<i32>>> {
    stream::unfold(vals.into_iter(), |mut vals| async {
        let val = vals.next()?;
        let response = network_request(val).await;
        Some((response, vals))
    })
}

fn main() {
    executor::block_on(async {
        let s = requests_in_sequence(vec![1, 2, 3]);

        let s = s.filter_map(|v| async move { v.ok() });
        let s = s.filter_map(|v| async move { v });
        let mut s = s.boxed_local();

        match s.next().await {
            Some(v) => println!("First success: {}", v),
            None => println!("No successful requests"),
        }
    });
}
Run Code Online (Sandbox Code Playgroud) 
use futures::{executor, stream, Stream, StreamExt}; // 0.3.4

type Error = Box<dyn std::error::Error>;
type Result<T, E = Error> = std::result::Result<T, E>;

async fn network_request(val: i32) -> Result<Option<i32>> {
    // Just for demonstration, don't do this in a real program
    use std::{
        thread,
        time::{Duration, Instant},
    };
    thread::sleep(Duration::from_secs(1));
    println!("Resolving {} at {:?}", val, Instant::now());

    match val {
        1 => Err("boom".into()),  // An error
        2 => Ok(None),            // No data
        _ => Ok(Some(val * 100)), // Success
    }
}

fn requests_in_sequence(vals: Vec<i32>) -> impl Stream<Item = Result<Option<i32>>> {
    stream::unfold(vals.into_iter(), |mut vals| async {
        let val = vals.next()?;
        let response = network_request(val).await;
        Some((response, vals))
    })
}

fn main() {
    executor::block_on(async {
        let s = requests_in_sequence(vec![1, 2, 3]);

        let s = s.filter_map(|v| async move { v.ok() });
        let s = s.filter_map(|v| async move { v });
        let mut s = s.boxed_local();

        match s.next().await {
            Some(v) => println!("First success: {}", v),
            None => println!("No successful requests"),
        }
    });
}
Run Code Online (Sandbox Code Playgroud)

有没有可能建立一个像这样的动态链

是的，通过利用async功能：

use futures::executor; // 0.3.4

type Error = Box<dyn std::error::Error>;
type Result<T, E = Error> = std::result::Result<T, E>;

async fn network_request(val: i32) -> Result<Option<i32>> {
    // Just for demonstration, don't do this in a real program
    use std::{
        thread,
        time::{Duration, Instant},
    };
    thread::sleep(Duration::from_secs(1));
    println!("Resolving {} at {:?}", val, Instant::now());

    match val {
        1 => Err("boom".into()),  // An error
        2 => Ok(None),            // No data
        _ => Ok(Some(val * 100)), // Success
    }
}

async fn requests_in_sequence(vals: Vec<i32>) -> Result<i32> {
    let mut vals = vals.into_iter().peekable();

    while let Some(v) = vals.next() {
        match network_request(v).await {
            Ok(Some(v)) => return Ok(v),
            Err(e) if vals.peek().is_none() => return Err(e),
            Ok(None) | Err(_) => { /* Do nothing and try the next source */ }
        }
    }

    Err("Ran out of sources".into())
}

fn main() {
    executor::block_on(async {
        match requests_in_sequence(vec![1, 2, 3]).await {
            Ok(v) => println!("First success: {}", v),
            Err(e) => println!("No successful requests: {}", e),
        }
    });
}
Run Code Online (Sandbox Code Playgroud)

也可以看看：

是否需要Future在移动到下一个之前完全评估迭代器中的每个

这不是你自己要求的一部分吗？强调我的：

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

归档时间：

查看次数：

3187 次

最近记录：

5 年，7 月 前
使用动态数量的.and()创建Diesel.rs查询 6
更多相关链接
相关归档
如何在Rust中生成一个范围内的随机数？ 18
切片到固定大小的数组 13
如何子类CompletableFuture？ 10
"溢出评估需求"但这种递归根本不应该发生 6
如何检查给定路径是文件还是目录？ 5
Rust中有类似nodemon的东西吗？ 5
如何匹配字符串的开头？ 5
迭代列表时借用的 RefCell 持续时间不够长 3
如何让 Visual Studio Code 重置 Rust 的 IntelliSense 引擎？ 3
Rust 中的堆栈分配与堆分配到底是如何工作的？ 1
难疑归档
使用Git将最近的提交移动到新分支 4647
带请求正文的HTTP GET 1896
JavaScript中==和===之间的区别 1592
如何计算列表项的出现次数？ 1417
在JavaScript中删除数组元素 - 删除vs splice 1304
如何在Android应用程序中的活动之间传递数据？ 1293
如何在JavaScript正则表达式中访问匹配的组？ 1277
生成0到9之间的随机整数 1224
为特定提交生成git补丁 1144
使用node.js作为简单的Web服务器 1068