# Rust Async

> When to activate: async Rust, tokio, async/await, futures, streams, channels, spawn, select, join

- Skill: `mattakushi432/rust-async` (Agent Skill)
- Install (CLI): `npx skillmds@latest add mattakushi432/rust-async`
- Raw SKILL.md: https://api.skillmd.com/api/skills/mattakushi432/rust-async/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: Mattakushi432 (https://skillmd.com/u/mattakushi432)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/mattakushi432/rust-async

---


# Rust Async Patterns

## async/await Basics

```toml
[dependencies]
tokio = { version = "1", features = ["full"] }
```

```rust
#[tokio::main]
async fn main() {
    let result = fetch_data("https://api.example.com/data").await;
    match result {
        Ok(data) => println!("Got: {data}"),
        Err(e) => eprintln!("Error: {e}"),
    }
}

async fn fetch_data(url: &str) -> Result<String, reqwest::Error> {
    let response = reqwest::get(url).await?;
    response.text().await
}
```

## Spawning Tasks

`tokio::spawn` creates a new async task that runs concurrently.

```rust
use tokio::task::JoinHandle;

async fn process_items(items: Vec<String>) -> Vec<String> {
    let handles: Vec<JoinHandle<String>> = items
        .into_iter()
        .map(|item| tokio::spawn(async move { expensive_transform(item).await }))
        .collect();

    let mut results = Vec::new();
    for handle in handles {
        results.push(handle.await.expect("task panicked"));
    }
    results
}

// Background worker with shutdown signal
async fn start_worker(mut shutdown: tokio::sync::broadcast::Receiver<()>) {
    loop {
        tokio::select! {
            _ = shutdown.recv() => break,
            _ = do_work() => {}
        }
    }
}
```

## Joining Multiple Futures

```rust
use tokio::try_join;

async fn parallel_fetch() -> Result<(User, Posts), AppError> {
    // Both run concurrently; fails fast if either errors
    let (user, posts) = try_join!(fetch_user(42), fetch_posts(42))?;
    Ok((user, posts))
}

// FuturesUnordered for dynamic sets
use futures::stream::{FuturesUnordered, StreamExt};

async fn process_dynamic(ids: Vec<u64>) {
    let mut futs: FuturesUnordered<_> = ids.into_iter()
        .map(|id| fetch_item(id))
        .collect();

    while let Some(result) = futs.next().await {
        println!("got: {:?}", result);
    }
}
```

## select! for Racing Futures

```rust
use tokio::time::{sleep, Duration};

async fn with_timeout<F, T>(fut: F, dur: Duration) -> Option<T>
where
    F: std::future::Future<Output = T>,
{
    tokio::select! {
        result = fut => Some(result),
        _ = sleep(dur) => None,
    }
}

// Cancellation token
use tokio_util::sync::CancellationToken;

async fn cancellable_work(token: CancellationToken) {
    loop {
        tokio::select! {
            biased;
            _ = token.cancelled() => break,
            _ = do_unit_of_work() => {}
        }
    }
}
```

## Channels

```rust
// mpsc: multiple producer, single consumer
use tokio::sync::mpsc;

async fn producer_consumer() {
    let (tx, mut rx) = mpsc::channel::<String>(32);

    for i in 0..5 {
        let tx = tx.clone();
        tokio::spawn(async move {
            tx.send(format!("message from {i}")).await.unwrap();
        });
    }
    drop(tx); // close when all producers done

    while let Some(msg) = rx.recv().await {
        println!("received: {msg}");
    }
}

// broadcast: single producer, multiple consumers
use tokio::sync::broadcast;

async fn broadcast_example() {
    let (tx, _) = broadcast::channel::<String>(16);
    let mut rx1 = tx.subscribe();
    let mut rx2 = tx.subscribe();

    tx.send("hello everyone".into()).unwrap();
    println!("{}", rx1.recv().await.unwrap());
    println!("{}", rx2.recv().await.unwrap());
}
```

## Streams

```rust
use futures::stream::{self, StreamExt};

async fn process_stream() {
    let doubled: Vec<u32> = stream::iter(0u32..10)
        .filter(|&x| async move { x % 2 == 0 })
        .map(|x| x * 2)
        .collect()
        .await;

    // Wrap mpsc receiver as a Stream
    use tokio_stream::wrappers::ReceiverStream;
    let (tx, rx) = tokio::sync::mpsc::channel::<i32>(16);
    let mut stream = ReceiverStream::new(rx);
    while let Some(item) = stream.next().await {
        println!("{item}");
    }
}
```

## Shared State in Async Code

```rust
use std::sync::Arc;
use tokio::sync::{Mutex, RwLock};

#[derive(Clone)]
struct AppState {
    counter: Arc<Mutex<u64>>,
    config: Arc<RwLock<Config>>,
}

impl AppState {
    async fn increment(&self) {
        let mut guard = self.counter.lock().await;
        *guard += 1;
    }

    async fn get_config(&self) -> Config {
        self.config.read().await.clone()
    }
}
```

## Common Anti-Patterns

- **Blocking in async context** — use `tokio::task::spawn_blocking` for CPU-heavy or blocking I/O
- **Holding `MutexGuard` across `.await`** — use `tokio::sync::Mutex`, not `std::sync::Mutex`
- **`std::thread::sleep` in async code** — use `tokio::time::sleep` instead
- **Spawning unlimited tasks** — use a semaphore or bounded channel to limit concurrency
- **Forgetting cancellation propagation** — check `CancellationToken` or `select!` on shutdown signals