Fastest way to send many groups of HTTP requests using new async/await syntax and control the amount of workers

Question

Most recent threads I have read are saying async is the better way to perform lots of I/O bound work such as sending HTTP requests and the like. I have tried to pick up async recently but am struggling with understanding how to send many groups of requests in parallel, for example:

let client = reqwest::Client::new();
let mut requests = 0;

let get = client.get("https://somesite.com").send().await?;
let response = get.text().await?;

if response.contains("some stuff") {
    let get = client.get("https://somesite.com/something").send().await?;
    let response = get.text().await?;

    if response.contains("some new stuff") {
        requests += 1;
        println!("Got response {}", requests)

This does what I want, but how can I run it in parallel and control the amount of "worker threads" or whatever the equivalent is to a thread pool in async?

I understand it is similar to this question, but mine is strictly talking about the nightly Rust async/await syntax and a more specific use case where groups of requests/tasks need to be done. I also find using combinators for these situations a bit confusing, was hoping the newer style would help make it a bit more readable.

score 0 · Answer 1 · answered Oct 24 '19 at 03:23

Not sure if this is the fastest way, as I am just experimenting myself, but here is my solution:

let client = reqwest::Client::new();

let links = vec![ // A vec of strings representing links
    "example.net/a".to_owned(), 
    "example.net/b".to_owned(),
    "example.net/c".to_owned(),
    "example.net/d".to_owned(),
    ];

let ref_client = &client; // Need this to prevent client from being moved into the first map
futures::stream::iter(links)
    .map(async move |link: String| {
        let res = ref_client.get(&link).send().await;

        // res.map(|res| res.text().await.unwrap().to_vec())
        match res { // This is where I would usually use `map`, but not sure how to await for a future inside a result 
            Ok(res) => Ok(res.text().await.unwrap()),
            Err(err) => Err(err), 
        }
    })
    .buffer_unordered(10) // Number of connection at the same time
    .filter_map(|c| future::ready(c.ok())) // Throw errors out, do your own error handling here
    .filter_map(|item| {
        if item.contains("abc") {
            future::ready(Some(item))
        } else {
            future::ready(None)
        }
    })
    .map(async move |sec_link| {
        let res = ref_client.get(&sec_link).send().await;
        match res {
            Ok(res) => Ok(res.text().await.unwrap()),
            Err(err) => Err(err),
        }
    })
    .buffer_unordered(10) // Number of connections for the secondary requests (so max 20 connections concurrently)
    .filter_map(|c| future::ready(c.ok()))
    .for_each(|item| {
        println!("File received: {}", item);
        future::ready(())
    })
    .await;

This requires the #![feature(async_closure)] feature.

Fastest way to send many groups of HTTP requests using new async/await syntax and control the amount of workers

1 Answers1