How to execute a method on a struct for each instance in a collection of structs until an error is encountered in Rust?

Question

I am trying to run the execute method for each instance of the struct LogCommand that I have in a collection of structs, stopping when an error is encountered.

use std::pin::Pin;
use std::future::Future;

use anyhow::Error;
use futures::future::ok;
use futures::stream::{TryStream, TryStreamExt};

pub struct LogCommand {
    message: String,
}

impl LogCommand {
    fn new(message: String) -> LogCommand {
        LogCommand { message }
    }

    fn execute(&self) -> Pin<Box<dyn Future<Output = Result<bool, Error>>>> {
        println!("{}", self.message);
        Box::pin(ok(true))
    }
}

fn main() {
    let mut structs: Vec<LogCommand> = Vec::new();
    structs.push(LogCommand::new("Test1".to_string()));
    structs.push(LogCommand::new("Test2".to_string()));
    structs.push(LogCommand::new("Test3".to_string()));

    run_all_methods(structs);
}

async fn run_all_methods(logCommands: Vec<LogCommand>) -> Result<bool, Error> {
    futures::stream::iter(logCommands).try_for_each(|i| i.execute())
}

The error I am getting is as follows:

the method `try_for_each` exists for struct `Iter<IntoIter<LogCommand>>`, but its trait bounds were not satisfied
the following trait bounds were not satisfied:
`futures::stream::Iter<std::vec::IntoIter<LogCommand>>: TryStream`
which is required by `futures::stream::Iter<std::vec::IntoIter<LogCommand>>: TryStreamExt`
`&futures::stream::Iter<std::vec::IntoIter<LogCommand>>: TryStream`
which is required by `&futures::stream::Iter<std::vec::IntoIter<LogCommand>>: TryStreamExt`
`&mut futures::stream::Iter<std::vec::IntoIter<LogCommand>>: TryStream`
which is required by `&mut futures::stream::Iter<std::vec::IntoIter<LogCommand>>: TryStreamExt`rustcClick for full compiler diagnostic
iter.rs(9, 1): doesn't satisfy `_: TryStreamExt`
iter.rs(9, 1): doesn't satisfy `_: TryStream`

Any idea on how I might go about running each of these methods and stopping when an error is encountered, without directly using a for loop? (I have been asked to write this in a more functional manner.)

Answer 1

The reason for the error is that try_for_each() is a method for TryStreams — streams of Results — but futures::stream::iter(logCommands) is not a stream of Results but of LogCommands. The Results haven't appeared yet because execute() hasn't been done yet.

To get the stream of Results, you can use StreamExt::then(), and then to get the completion or error you can use one of the TryStream operations.

But you haven't said what you want to do with the bools. If you don't actually want any data there, you can replace it with () and collect:

use futures::StreamExt;

...

impl LogCommand {
    fn new(message: String) -> LogCommand {
        LogCommand { message }
    }

    fn execute(&self) -> Pin<Box<dyn Future<Output = Result<(), Error>>>> {
        println!("{}", self.message);
        Box::pin(ok(()))
    }
}

async fn run_all_methods(log_commands: Vec<LogCommand>) -> Result<(), Error> {
    futures::stream::iter(log_commands)
        .then(|i| i.execute())
        .try_collect()
        .await
}

Playground

If you want to combine the bools in some way (logical AND or OR), then maybe try_fold() will suit. If you want to exit early if the bool has a particular value, convert it to an error.