1# rust-async-patterns — detailed patterns and worked examples
2 
3## Patterns
4 
5### Pattern 1: Concurrent Task Execution
6 
7```rust
8use tokio::task::JoinSet;
9use anyhow::Result;
10 
11// Spawn multiple concurrent tasks
12async fn fetch_all_concurrent(urls: Vec<String>) -> Result<Vec<String>> {
13    let mut set = JoinSet::new();
14 
15    for url in urls {
16        set.spawn(async move {
17            fetch_data(&url).await
18        });
19    }
20 
21    let mut results = Vec::new();
22    while let Some(res) = set.join_next().await {
23        match res {
24            Ok(Ok(data)) => results.push(data),
25            Ok(Err(e)) => tracing::error!("Task failed: {}", e),
26            Err(e) => tracing::error!("Join error: {}", e),
27        }
28    }
29 
30    Ok(results)
31}
32 
33// With concurrency limit
34use futures::stream::{self, StreamExt};
35 
36async fn fetch_with_limit(urls: Vec<String>, limit: usize) -> Vec<Result<String>> {
37    stream::iter(urls)
38        .map(|url| async move { fetch_data(&url).await })
39        .buffer_unordered(limit) // Max concurrent tasks
40        .collect()
41        .await
42}
43 
44// Select first to complete
45use tokio::select;
46 
47async fn race_requests(url1: &str, url2: &str) -> Result<String> {
48    select! {
49        result = fetch_data(url1) => result,
50        result = fetch_data(url2) => result,
51    }
52}
53```
54 
55### Pattern 2: Channels for Communication
56 
57```rust
58use tokio::sync::{mpsc, broadcast, oneshot, watch};
59 
60// Multi-producer, single-consumer
61async fn mpsc_example() {
62    let (tx, mut rx) = mpsc::channel::<String>(100);
63 
64    // Spawn producer
65    let tx2 = tx.clone();
66    tokio::spawn(async move {
67        tx2.send("Hello".to_string()).await.unwrap();
68    });
69 
70    // Consume
71    while let Some(msg) = rx.recv().await {
72        println!("Got: {}", msg);
73    }
74}
75 
76// Broadcast: multi-producer, multi-consumer
77async fn broadcast_example() {
78    let (tx, _) = broadcast::channel::<String>(100);
79 
80    let mut rx1 = tx.subscribe();
81    let mut rx2 = tx.subscribe();
82 
83    tx.send("Event".to_string()).unwrap();
84 
85    // Both receivers get the message
86    let _ = rx1.recv().await;
87    let _ = rx2.recv().await;
88}
89 
90// Oneshot: single value, single use
91async fn oneshot_example() -> String {
92    let (tx, rx) = oneshot::channel::<String>();
93 
94    tokio::spawn(async move {
95        tx.send("Result".to_string()).unwrap();
96    });
97 
98    rx.await.unwrap()
99}
100 
101// Watch: single producer, multi-consumer, latest value
102async fn watch_example() {
103    let (tx, mut rx) = watch::channel("initial".to_string());
104 
105    tokio::spawn(async move {
106        loop {
107            // Wait for changes
108            rx.changed().await.unwrap();
109            println!("New value: {}", *rx.borrow());
110        }
111    });
112 
113    tx.send("updated".to_string()).unwrap();
114}
115```
116 
117### Pattern 3: Async Error Handling
118 
119```rust
120use anyhow::{Context, Result, bail};
121use thiserror::Error;
122 
123#[derive(Error, Debug)]
124pub enum ServiceError {
125    #[error("Network error: {0}")]
126    Network(#[from] reqwest::Error),
127 
128    #[error("Database error: {0}")]
129    Database(#[from] sqlx::Error),
130 
131    #[error("Not found: {0}")]
132    NotFound(String),
133 
134    #[error("Timeout after {0:?}")]
135    Timeout(std::time::Duration),
136}
137 
138// Using anyhow for application errors
139async fn process_request(id: &str) -> Result<Response> {
140    let data = fetch_data(id)
141        .await
142        .context("Failed to fetch data")?;
143 
144    let parsed = parse_response(&data)
145        .context("Failed to parse response")?;
146 
147    Ok(parsed)
148}
149 
150// Using custom errors for library code
151async fn get_user(id: &str) -> Result<User, ServiceError> {
152    let result = db.query(id).await?;
153 
154    match result {
155        Some(user) => Ok(user),
156        None => Err(ServiceError::NotFound(id.to_string())),
157    }
158}
159 
160// Timeout wrapper
161use tokio::time::timeout;
162 
163async fn with_timeout<T, F>(duration: Duration, future: F) -> Result<T, ServiceError>
164where
165    F: std::future::Future<Output = Result<T, ServiceError>>,
166{
167    timeout(duration, future)
168        .await
169        .map_err(|_| ServiceError::Timeout(duration))?
170}
171```
172 
173### Pattern 4: Graceful Shutdown
174 
175```rust
176use tokio::signal;
177use tokio::sync::broadcast;
178use tokio_util::sync::CancellationToken;
179 
180async fn run_server() -> Result<()> {
181    // Method 1: CancellationToken
182    let token = CancellationToken::new();
183    let token_clone = token.clone();
184 
185    // Spawn task that respects cancellation
186    tokio::spawn(async move {
187        loop {
188            tokio::select! {
189                _ = token_clone.cancelled() => {
190                    tracing::info!("Task shutting down");
191                    break;
192                }
193                _ = do_work() => {}
194            }
195        }
196    });
197 
198    // Wait for shutdown signal
199    signal::ctrl_c().await?;
200    tracing::info!("Shutdown signal received");
201 
202    // Cancel all tasks
203    token.cancel();
204 
205    // Give tasks time to cleanup
206    tokio::time::sleep(Duration::from_secs(5)).await;
207 
208    Ok(())
209}
210 
211// Method 2: Broadcast channel for shutdown
212async fn run_with_broadcast() -> Result<()> {
213    let (shutdown_tx, _) = broadcast::channel::<()>(1);
214 
215    let mut rx = shutdown_tx.subscribe();
216    tokio::spawn(async move {
217        tokio::select! {
218            _ = rx.recv() => {
219                tracing::info!("Received shutdown");
220            }
221            _ = async { loop { do_work().await } } => {}
222        }
223    });
224 
225    signal::ctrl_c().await?;
226    let _ = shutdown_tx.send(());
227 
228    Ok(())
229}
230```
231 
232### Pattern 5: Async Traits
233 
234```rust
235use async_trait::async_trait;
236 
237#[async_trait]
238pub trait Repository {
239    async fn get(&self, id: &str) -> Result<Entity>;
240    async fn save(&self, entity: &Entity) -> Result<()>;
241    async fn delete(&self, id: &str) -> Result<()>;
242}
243 
244pub struct PostgresRepository {
245    pool: sqlx::PgPool,
246}
247 
248#[async_trait]
249impl Repository for PostgresRepository {
250    async fn get(&self, id: &str) -> Result<Entity> {
251        sqlx::query_as!(Entity, "SELECT * FROM entities WHERE id = $1", id)
252            .fetch_one(&self.pool)
253            .await
254            .map_err(Into::into)
255    }
256 
257    async fn save(&self, entity: &Entity) -> Result<()> {
258        sqlx::query!(
259            "INSERT INTO entities (id, data) VALUES ($1, $2)
260             ON CONFLICT (id) DO UPDATE SET data = $2",
261            entity.id,
262            entity.data
263        )
264        .execute(&self.pool)
265        .await?;
266        Ok(())
267    }
268 
269    async fn delete(&self, id: &str) -> Result<()> {
270        sqlx::query!("DELETE FROM entities WHERE id = $1", id)
271            .execute(&self.pool)
272            .await?;
273        Ok(())
274    }
275}
276 
277// Trait object usage
278async fn process(repo: &dyn Repository, id: &str) -> Result<()> {
279    let entity = repo.get(id).await?;
280    // Process...
281    repo.save(&entity).await
282}
283```
284 
285### Pattern 6: Streams and Async Iteration
286 
287```rust
288use futures::stream::{self, Stream, StreamExt};
289use async_stream::stream;
290 
291// Create stream from async iterator
292fn numbers_stream() -> impl Stream<Item = i32> {
293    stream! {
294        for i in 0..10 {
295            tokio::time::sleep(Duration::from_millis(100)).await;
296            yield i;
297        }
298    }
299}
300 
301// Process stream
302async fn process_stream() {
303    let stream = numbers_stream();
304 
305    // Map and filter
306    let processed: Vec<_> = stream
307        .filter(|n| futures::future::ready(*n % 2 == 0))
308        .map(|n| n * 2)
309        .collect()
310        .await;
311 
312    println!("{:?}", processed);
313}
314 
315// Chunked processing
316async fn process_in_chunks() {
317    let stream = numbers_stream();
318 
319    let mut chunks = stream.chunks(3);
320 
321    while let Some(chunk) = chunks.next().await {
322        println!("Processing chunk: {:?}", chunk);
323    }
324}
325 
326// Merge multiple streams
327async fn merge_streams() {
328    let stream1 = numbers_stream();
329    let stream2 = numbers_stream();
330 
331    let merged = stream::select(stream1, stream2);
332 
333    merged
334        .for_each(|n| async move {
335            println!("Got: {}", n);
336        })
337        .await;
338}
339```
340 
341### Pattern 7: Resource Management
342 
343```rust
344use std::sync::Arc;
345use tokio::sync::{Mutex, RwLock, Semaphore};
346 
347// Shared state with RwLock (prefer for read-heavy)
348struct Cache {
349    data: RwLock<HashMap<String, String>>,
350}
351 
352impl Cache {
353    async fn get(&self, key: &str) -> Option<String> {
354        self.data.read().await.get(key).cloned()
355    }
356 
357    async fn set(&self, key: String, value: String) {
358        self.data.write().await.insert(key, value);
359    }
360}
361 
362// Connection pool with semaphore
363struct Pool {
364    semaphore: Semaphore,
365    connections: Mutex<Vec<Connection>>,
366}
367 
368impl Pool {
369    fn new(size: usize) -> Self {
370        Self {
371            semaphore: Semaphore::new(size),
372            connections: Mutex::new((0..size).map(|_| Connection::new()).collect()),
373        }
374    }
375 
376    async fn acquire(&self) -> PooledConnection<'_> {
377        let permit = self.semaphore.acquire().await.unwrap();
378        let conn = self.connections.lock().await.pop().unwrap();
379        PooledConnection { pool: self, conn: Some(conn), _permit: permit }
380    }
381}
382 
383struct PooledConnection<'a> {
384    pool: &'a Pool,
385    conn: Option<Connection>,
386    _permit: tokio::sync::SemaphorePermit<'a>,
387}
388 
389impl Drop for PooledConnection<'_> {
390    fn drop(&mut self) {
391        if let Some(conn) = self.conn.take() {
392            let pool = self.pool;
393            tokio::spawn(async move {
394                pool.connections.lock().await.push(conn);
395            });
396        }
397    }
398}
399```
400 
401## Debugging Tips
402 
403```rust
404// Enable tokio-console for runtime debugging
405// Cargo.toml: tokio = { features = ["tracing"] }
406// Run: RUSTFLAGS="--cfg tokio_unstable" cargo run
407// Then: tokio-console
408 
409// Instrument async functions
410use tracing::instrument;
411 
412#[instrument(skip(pool))]
413async fn fetch_user(pool: &PgPool, id: &str) -> Result<User> {
414    tracing::debug!("Fetching user");
415    // ...
416}
417 
418// Track task spawning
419let span = tracing::info_span!("worker", id = %worker_id);
420tokio::spawn(async move {
421    // Enters span when polled
422}.instrument(span));
423```
424