Iterate over Stream processes concurrently

[depombo]

Hi I am unfamiliar with the Stream API and I am struggling to be able run concurrent operations on each process from heim_process::processes. This is more of an async Rust question, but could you guys please help me figure out how to idiomatically and concurrently run the following operation on each process? The code provided below runs sequentially and I have not been able to figure out how to get it to work given some of the syntax limitations of iterating over Streams. It would also be great for newcomers like me to have an example of something like this in https://github.com/heim-rs/heim/blob/master/examples/process.rs

Thanks in advance!

use heim_process::processes;

...

async fn get_procs_cpu_usage() -> Vec<f32> {
  let mut usages = Vec::new();
  let mut processes = processes();
  while let Some(process) = processes.next().await {
    if let Ok(proc) = process {
      let measurement_1 = proc.cpu_usage().await.expect("Failed to get CPU usage for process");
      // CpuUsage struct represents instantaneous CPU usage and
      // does not represent any reasonable value by itself
      sleep(Duration::from_secs(1)).await;
      let measurement_2 = proc.cpu_usage().await.expect("Failed to get CPU usage for process");
      let usage = (measurement_2 - measurement_1).get::<ratio>();
      usages.push(usage);
    }
  }
  usages
}

[svartalf]

Depending on your use-case you might want to use one of the TryStreamExt trait methods, namely try_for_each_concurrent or try_buffer_unordered, for example.

While I prefer not to spend time on writing more newcomers-friendly documentation here (it is mostly out of scope and I do not have enough free time to do that), any PRs with more examples are more then welcomed!

[depombo]

Totally get it, here is how I did it in case it saves another newcomer in the internet some time:

async fn get_procs_cpu_usage() -> Vec<f32> {
  let processes = processes();
  let futures = processes
    .map(|process| async {
      match process {
        Ok(proc) => {
          let measurement_1 = proc.cpu_usage().await.expect("Failed to get CPU usage for process");
          // CpuUsage struct represents instantaneous CPU usage and
          // does not represent any reasonable value by itself
          sleep(Duration::from_secs(1)).await;
          let measurement_2 = proc.cpu_usage().await.expect("Failed to get CPU usage for process");
          let usage = (measurement_2 - measurement_1).get::<ratio>();
          usage
        },
        Err(_) => 0.0
      }
    })
    .collect::<Vec<_>>()
    .await;
  join_all(futures).await
}