Switching higher-order streams to first-order streams

Rust’s async ecosystem relies on the Stream trait to handle asynchronous sequences of values, much like Iterator does for synchronous ones. Developers now push to refactor higher-order combinators—those handling streams of streams, like flatten—into first-order versions. This shift cuts type complexity, eases pinning woes, and sharpens performance in high-throughput scenarios such as network servers or crypto data feeds. Why does it matter? It makes async pipelines more composable without the nested generics that bloat compile times and runtime state.

Start with the basics. An Iterator produces values on demand via next(), returning Option<T>: Some(T) for a value, None at end. Rust defines it as:

trait Iterator {
    type Item;
    fn next(&mut self) -> Option<Self::Item>;
}

Streams extend this to async. They use poll_next(), which returns Poll<Option<Item>>: Pending if not ready, Ready(Some(T)) for a value, or Ready(None) to signal closure. Ignore Pin and Context for now—these enforce self-referential async safety. The trait looks like:

trait Stream {
    type Item;
    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>
    ) -> Poll<Option<Self::Item>>;
}

This parallelism shines in combinators. Iterators chain methods like skip_while().filter().map(), each returning a new iterator. Streams do the same: futures::StreamExt provides map, filter, and more, all returning impl Stream. Take map: it transforms Stream<Item=T> to Stream<Item=U> via a closure.

fn map<U, F>(self, f: F) -> impl Stream<Item = U>
where
    F: FnMut(Self::Item) -> U;

Filter keeps items passing a predicate:

fn filter<F>(self, f: F) -> impl Stream<Item = Self::Item>
where
    F: FnMut(&Self::Item) -> bool;

These work because Rust’s state machines track closure state across polls. Chain them on a TCP stream: filter malformed packets, map to parsed events, then forward. In crypto apps, this processes WebSocket feeds without blocking—essential for real-time trading or oracle data.

The Higher-Order Problem: Flatten

Complications arise with nested structures. Flatten turns Stream<Item = S> where S: Stream<Item = T> into Stream<Item = T>. Picture a stream yielding sub-streams: HTTP/2 multiplexes connections into frame streams; Kafka partitions yield message batches.

fn flatten(self) -> impl Stream<Item = <Self::Item as Stream>::Item>
where
    Self::Item: Stream;

This is “higher-order”: the outer stream’s Item is itself a stream kind. Implementation polls the current inner stream until exhaust, then advances the outer. Futures-util’s Flatten<St> struct manages this, but generics explode: monomorphization per inner type bloats binaries. Pinning propagates inward, risking use-after-free if inners outlive outers. Compile times suffer on deep chains—real projects hit seconds per change.

Switching to First-Order: Simpler, Faster Async

First-order streams fix this by flattening at the type level upfront. Instead of generic Item = impl Stream<Item=T>, use concrete wrappers or boxed trait objects: Stream<Item = Box, then flatten without HKT-like nesting. Libraries experiment with this—tokio-stream's versions since 0.1 erase types via Box, trading some perf for ergonomics.