Skip to main content

simple_concurrent_get/ordered/
get_foreach.rs

1use reqwest::{ Client, Response };
2use futures_util::stream::{ StreamExt as _ };
3
4use std::sync::Arc;
5use std::sync::mpsc::{
6    IntoIter,
7    channel,
8};
9
10use crate::{
11    client,
12};
13
14
15
16pub async fn concurrent_get_foreach<I,S,F,R>(fetch_urls: I, concurrent: usize, run_for_each: F) -> IntoIter<R>
17where
18    S: reqwest::IntoUrl,
19    I: IntoIterator<Item=S>,
20    F: Copy + FnOnce(reqwest::Result<Response>) -> R,
21{
22    let client: Arc<Client> = Arc::new(client::build_preset());
23
24    let (sender, receiver) = channel();
25
26    let bodies = futures_util::stream::iter(fetch_urls)
27        .map(|url| {
28            let client = client.clone();
29            async move {
30                run_for_each(
31                    client.get(url).send().await
32                )
33            }
34        })
35        .buffered(concurrent);
36
37    bodies
38        .for_each(|resp| {
39            let sender = sender.clone();
40            async move {
41                sender.send(resp).unwrap();
42            }
43        })
44        .await;
45
46    drop(sender);
47    receiver.into_iter()
48}