use std::sync::Arc;
use std::time::Duration;
use futures::stream::{FuturesUnordered, StreamExt};
use reqwest::Client;
use tokio::sync::Semaphore;
use crate::models::FetchResult;
pub struct Fetcher {
client: Client,
semaphore: Arc<Semaphore>,
}
impl Fetcher {
pub fn new(max_concurrent: usize) -> Self {
let client = Client::builder()
.timeout(Duration::from_secs(10))
.user_agent("fanout-fetcher/1.0")
.build()
.expect("failed to build reqwest client");
Fetcher {
client,
semaphore: Arc::new(Semaphore::new(max_concurrent)),
}
}
pub async fn fetch_all(&self, urls: Vec<String>) -> Vec<FetchResult> {
let mut in_flight = FuturesUnordered::new();
for url in urls {
in_flight.push(self.fetch_one(url));
}
let mut results = Vec::with_capacity(in_flight.len());
while let Some(result) = in_flight.next().await {
results.push(result);
}
results
}
async fn fetch_one(&self, url: String) -> FetchResult {
// Hold a permit for the whole request so only N run at once.
let _permit = self.semaphore.acquire().await.expect("semaphore closed");
let body = match self.client.get(&url).send().await {
Ok(resp) => match resp.error_for_status() {
Ok(ok) => ok.text().await.map_err(|e| e.to_string()),
Err(status_err) => Err(status_err.to_string()),
},
Err(net_err) => Err(net_err.to_string()),
};
FetchResult { url, body }
}
}#[derive(Debug)]
pub struct FetchResult {
pub url: String,
pub body: Result<String, String>,
}
impl FetchResult {
pub fn is_ok(&self) -> bool {
self.body.is_ok()
}
pub fn byte_len(&self) -> usize {
match &self.body {
Ok(text) => text.len(),
Err(_) => 0,
}
}
}
#[derive(Debug, Default)]
pub struct Summary {
pub succeeded: usize,
pub failed: usize,
pub total_bytes: usize,
}
impl Summary {
pub fn from_results(results: &[FetchResult]) -> Self {
let mut summary = Summary::default();
for r in results {
if r.is_ok() {
summary.succeeded += 1;
summary.total_bytes += r.byte_len();
} else {
summary.failed += 1;
}
}
summary
}
}mod fetcher;
mod models;
use fetcher::Fetcher;
use models::Summary;
#[tokio::main]
async fn main() {
let urls: Vec<String> = vec![
"https://httpbin.org/get",
"https://httpbin.org/status/200",
"https://httpbin.org/status/503",
"https://example.com",
"https://httpbin.org/delay/1",
]
.into_iter()
.map(String::from)
.collect();
let fetcher = Fetcher::new(8);
let results = fetcher.fetch_all(urls).await;
for r in &results {
match &r.body {
Ok(text) => println!("OK {:>6} bytes {}", text.len(), r.url),
Err(err) => println!("FAIL {:>12} {}", err, r.url),
}
}
let summary = Summary::from_results(&results);
println!(
"\ndone: {} ok, {} failed, {} bytes total",
summary.succeeded, summary.failed, summary.total_bytes
);
}This snippet shows how to fan out a batch of HTTP GET requests concurrently and collect the results without letting the process open thousands of sockets at once. The naive form of this pattern is join_all(urls.map(fetch)), which spawns every request immediately; that works for a handful of URLs but melts down under a large list because it exhausts file descriptors and triggers server-side throttling. The code here keeps the same collect-all ergonomics while capping in-flight work.
In fetcher.rs, Fetcher wraps a single reqwest::Client (which owns a connection pool, so it must be reused, not recreated per call) and an Arc<Semaphore>. The fetch_all method builds a FuturesUnordered of individual fetch_one futures. Each future first awaits semaphore.acquire(), so only max_concurrent permits are ever held at once; the rest of the futures park until a permit frees up. This is real backpressure: the stream drives the requests, but the semaphore is the throttle. Results come back in completion order rather than submission order, which is why each FetchResult carries its own url for correlation.
Errors are modeled as values, not panics. fetch_one returns a FetchResult whose body is a Result<String, String>, so one failing host never aborts the whole batch — the .awaited future simply records the error and the stream keeps draining. The _permit binding holds the semaphore guard alive for the duration of the request and releases it on drop.
In models.rs, FetchResult and Summary are plain data types; Summary::from_results folds the collected vector into success/failure counts, illustrating that once the fan-out completes the data is ordinary and easy to reason about.
In main.rs, the #[tokio::main] entry point constructs one Fetcher with a concurrency limit of eight and awaits fetch_all. The key trade-off is throughput versus resource safety: a higher limit finishes faster but risks rate limits and memory pressure from buffered bodies. FuturesUnordered is preferred over join_all here because it polls a bounded working set and yields as soon as any request settles, keeping latency low. A common pitfall is forgetting to move the permit into the async block, which would release it early and defeat the limit.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
struct Config<'a> {
name: &'a str,
value: &'a str,
}
fn parse_config(line: &str) -> Config {Lifetime annotations for flexible borrowing in structs
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use tracing::{info, instrument};
#[instrument]
fn process_request(user_id: u64) {
info!(user_id, "Processing request");
// Work happens heretracing for structured logging and distributed tracing
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
Share this code
Here's the card — post it anywhere.
