title: How to integrate async data collection with threadpool data processing in Rust
body: <p>I'd like to improve the integration of my async data collection with my rayon data processing by overlapping the retrieval and the processing.  Currently, I pull lots of pages from a web site using normal async code.  Once that is complete, I do the cpu-intensive work using rayon's par_iter.</p>
  <p>It seems like I should be able to easily overlap the processing, so that I'm not waiting for every last page before I begin the grunt work.  Every page that I retrieve is independent of the others, so there is no need to wait before the conversion.</p>
  <p>Here's what I have working currently (simplified just a bit):</p>
  <pre><code>use rayon::prelude::*;
  use futures::{stream, StreamExt};
  use reqwest::{Client, Result};
  
  
  const CONCURRENT_REQUESTS: usize = usize::MAX;
  const MAX_PAGE: usize = 1000;
  
  #[tokio::main]
  async fn main() {
      // get data from server
      let client = Client::new();
      let bodies: Vec&lt;Result&lt;String&gt;&gt; = stream::iter(1..MAX_PAGE+1)
          .map(|page_number| {
              let client = &amp;client;
              async move {
                  client
                      .get(format!(&quot;https://someurl?{page_number}&quot;))
                      .send()
                      .await?
                      .text()
                      .await
              }
          })
          .buffer_unordered(CONCURRENT_REQUESTS)
          .collect()
          .await;
  
      // transform the data
      let mut rows: Vec&lt;MyRow&gt; = bodies
          .par_iter()
          .filter_map(|body| body.as_ref().ok())
          .map(|data| {
              let page = serde_json::from_str::&lt;MyPage&gt;(data).unwrap();
              page.rows
                  .iter()
                  .map(|x| Row::new(x))
                  .collect::&lt;Vec&lt;MyRow&gt;&gt;()
          })
          .flatten()
          .collect();
  
      // do something with rows
  }
  </code></pre>