I see this happen almost every time someone moves from synchronous Rust to async. They've mastered Iterator, they're comfortable with for loops, and then they encounter their first Stream. They try to treat it like a collection, and the compiler starts screaming.

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

async fn process_events() {
    let events = stream::iter(vec!["event1", "event2", "event3"]);

    // This looks right, but it won't compile.
    for event in events { 
        println!("Processing {}", event);
    }
}

The "For Loop" Temptation

The error you'll get is a wall of text telling you that Stream doesn't implement IntoIterator. It makes sense if you think about it: a for loop is synchronous. It expects the next item to be available immediately. But a Stream is fundamentally asynchronous. The next item might not exist yet—it might be waiting on a network packet or a timer.

If Rust allowed a for loop here, the loop would have to block the entire thread until the next item arrived, which completely defeats the purpose of using async in the first place. You can't "block" inside an async function without risking a deadlock or killing your performance.

Polling with while let and StreamExt

To fix this, we have to change how we request items. Instead of the loop driving the stream, we manually poll the stream for the next item and .await the result. This allows the executor to pause this specific task and do other work while the stream is waiting for data.

First, you need the StreamExt trait. This is the "extension" trait that adds the most useful methods to streams—similar to how Iterator provides map and filter. Without it, the Stream trait itself is pretty bare.

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

async fn process_events() {
    let mut events = stream::iter(vec!["event1", "event2", "event3"]);

    // We use while let because .next() returns Option<Item>
    while let Some(event) = events.next().await {
        println!("Processing {}", event);
    }
}

Notice two key changes: I made events mutable (because .next() modifies the stream's internal state) and I used while let Some(...) = stream.next().await. This is the standard "idiom" for consuming a stream sequentially.

Moving Beyond Sequential Processing

Processing one item at a time is fine, but it's often not why we use streams. If each event required a database call, the while let loop would be a bottleneck. You'd be waiting for event 1 to finish before even starting to fetch event 2.

This is where buffer_unordered comes in. It's one of my favorite tools in the futures crate. It allows you to kick off multiple futures from a stream and handle them as they complete, regardless of the order they started in.

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

async fn handle_event(id: &str) -> String {
    // Simulate some async work (like an API call)
    format!("Result for {}", id)
}

async fn process_events_concurrently() {
    let events = stream::iter(vec!["event1", "event2", "event3"]);

    let mut results = events
        .map(|id| handle_event(id)) // Transform stream of IDs into stream of Futures
        .buffer_unordered(5);      // Allow up to 5 futures to run concurrently

    while let Some(result) = results.next().await {
        println!("{}", result);
    }
}

By using buffer_unordered, you're telling Rust: "Keep up to 5 requests in flight at once." If "event2" finishes before "event1", it will be yielded first. This is how you build high-throughput async systems without manually managing a vector of join handles.

📋 Practical Task

Building an Async Log Filtering Stream

You are building a system that monitors a stream of log entries. Some logs are INFO, some are WARN, and some are ERROR. You need to create a processor that filters out everything except ERROR logs and converts them into uppercase alerts.

Your Task:

• Start with a stream of strings: vec!["INFO: System boot", "ERROR: Disk full", "INFO: User login", "ERROR: Network timeout", "WARN: High memory"].
• Use StreamExt::filter to keep only the strings that start with "ERROR".
• Use StreamExt::map to transform the remaining strings to uppercase.
• Consume the resulting stream using a while let loop and print each alert to the console.

Hint: Remember that filter requires the closure to return a future (or a bool wrapped in a way that works with async), but since this is a simple check, you can use filter(|x| async move { ... }) or similar patterns depending on your crate version.