What I'm trying to do is save all JobHandles in a vector and after iterating an X amount of time I want to await them all.

I'm doing this because the endpoint I'm sending requests too will return a 429 error if I send too many requests within a certain time frame.

#[tokio::main]
pub async fn get_collection_stats(city_name: String) -> Result<(serde_json::Value, surf::StatusCode), Box<dyn Error>> {
    let endpoint = format!("https://some-city-api.org/{}", city_name);

    let mut city_res = surf::get(&endpoint).await;

    let mut res: surf::Response= match city_res {
        Ok(value) => value,
        Err(e) => { panic!("Error: {}", e) }
    };

    let stats: serde_json::Value  = match res.body_json().await.ok() {
        Some(val) => val,
        None => serde_json::from_str("{}").unwrap()
    };

    Ok((stats, res.status()))
}

let mut count = 0;
let mut requests: Vec<_> = Vec::new();
for name in city_names {
    if count < 5 {
        let mut stats = tokio::task::spawn_blocking(|| {
            match get_data_about_city(String::from(name)) {
                Ok(value) => value,
                Err(_) => serde_json::from_str("{}").unwrap()
            }
        });

        requests.push(stats);
        count  = 1;
    } else {
        for task in requests {
            dbg!(task.await);
        }
        count = 0;
        break
    }
}

So far I have this. This works fine, but it only works when I have the break in the else. I want to be able to do batches of 5 requests without that break. Without the break I get an error like this:

error[E0382]: borrow of moved value: `requests`
   --> src\main.rs:109:13
    |
87  |     let mut requests: Vec<_> = Vec::new();
    |         ------------ move occurs because `requests` has type `Vec<tokio::task::JoinHandle<(serde_json::Value, StatusCode)>>`, which does not implement the `Copy` trait
...
109 |             requests.push(stats);
    |             ^^^^^^^^^^^^^^^^^^^^ value borrowed here after move
...
112 |             for task in requests {
    |                         --------
    |                         |
    |                         `requests` moved due to this implicit call to `.into_iter()`, in previous iteration of loop
    |                         help: consider borrowing to avoid moving into the for loop: `&requests`
    |
note: this function takes ownership of the receiver `self`, which moves `requests`
   --> C:\Users\Zed\.rustup\toolchains\stable-x86_64-pc-windows-msvc\lib/rustlib/src/rust\library\core\src\iter\traits\collect.rs:234:18
    |
234 |     fn into_iter(self) -> Self::IntoIter;
    |                  ^^^^

Okay, I fixed the move issue. Now I have this problem.

   |
113 |                 dbg!(task.await);
    |                      ^^^^^^^^^^ `&tokio::task::JoinHandle<(serde_json::Value, StatusCode)>` is not a future
    |
    = help: the trait `Future` is not implemented for `&tokio::task::JoinHandle<(serde_json::Value, StatusCode)>`
    = note: `Future` is implemented for `&mut tokio::task::JoinHandle<(serde_json::Value, surf::StatusCode)>`, but not for `&tokio::task::JoinHandle<(serde_json::Value, surf::StatusCode)>`

How should I proceed with what I want to do?

CodePudding user response：

I figured it out. I ended up using the futures crates. The flow would look something like this.

for name in city_names {
    let url = format!("https://some-city-api.org/{}", name);
        urls.push(url.clone());
}

let mut futs = FuturesUnordered::new();

for url in urls {
    futs.push(surf::get(url));
    
    // 50 requests reached, await everything in buffer
    if futs.len() == 50 {
        while let Some(res) = futs.next().await {
            // Do something with requests
        }
    }
}