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
}
}
}