I remember a developer on my team a couple of years ago who spent nearly a week fighting a "callback hell" nightmare in a crypto price tracker we were building. He had a delegate method firing every 200ms with new price updates, and he was trying to sync that data with the UI using a mess of manual locks and `@Published` properties. The code was brittle, and he was seeing weird race conditions where the UI would flicker between old and new prices. When I showed him how to wrap that delegate in an AsyncStream, he actually looked relieved. We collapsed fifty lines of state management into a single for await loop. That is the real power of AsyncSequence: it lets you treat a continuous flow of events as if it were just a standard collection.

Bridging the Gap with Continuation

You’ve likely encountered AsyncSequence in the wild—think of URLSession.bytes(from:)—but the real magic happens when you need to create your own. Most of the time, you're dealing with legacy APIs or system frameworks that still rely on closures or delegates. AsyncStream is your bridge here.

The core of an AsyncStream is the continuation. This object is the handle you use to "yield" values into the stream from a non-async context. I always tell people to think of it as a pipe: the continuation is the mouth of the pipe where you pour data in, and the for await loop is where the data comes out.

struct PriceMonitor {
    func streamPrices() -> AsyncStream<Double> {
        AsyncStream { continuation in
            // Imagine this is a legacy SDK callback
            PriceSDK.shared.startUpdating { newPrice in
                continuation.yield(newPrice)
            }
            
            // We handle the shutdown later
        }
    }
}

Notice how the AsyncStream closure gives us that continuation immediately. This allows us to set up our listener and start pushing values into the stream whenever the SDK decides they're ready.

Handling the Cleanup

One thing that often trips people up is the lifecycle of the stream. If you just yield values and forget about the producer, you've got a memory leak waiting to happen. If the task consuming the stream is cancelled, you need a way to tell the underlying SDK to stop sending updates.

The continuation.onTermination handler is where you do this. It's not optional if you're managing resources. I've seen too many apps crash because a background socket remained open long after the view controller that requested the data was dismissed.

AsyncStream { continuation in
    let observer = PriceSDK.shared.observe { price in
        continuation.yield(price)
    }
    
    continuation.onTermination = { termination in
        switch termination {
        case .cancelled:
            print("Consumer stopped listening, killing the SDK observer.")
        case .finished:
            print("Stream completed naturally.")
        }
        observer.stop()
    }
}

Processing the Stream in Real-time

Since AsyncStream conforms to AsyncSequence, you get all the functional tools you're used to, like filter and map. The crucial difference here is that these operations are lazy. They don't process the whole collection (because there is no "whole" collection); they process each element as it arrives from the continuation.

If you only care about price jumps greater than 5%, you don't need to put an if statement inside your loop. You can chain the logic directly onto the stream. It makes the calling code incredibly clean, almost like you're writing a declarative query for live data.

let priceStream = monitor.streamPrices()

// We filter and transform before we even start the loop
let significantChanges = priceStream
    .filter { $0 > 100.0 }
    .map { "Price alert: \($0)" }

for await alert in significantChanges {
    print(alert)
}

Just keep in mind: once you start iterating with for await, the loop will suspend your task until the next value is yielded or the stream is closed. If you need to do other things while waiting for prices, you'll need to wrap that loop in its own Task.

📋 Practical Task

Exercise: Building a System Memory Pressure Monitor

Your goal is to create a utility that monitors "Memory Pressure" events. Since we don't have a real hardware sensor in this environment, you will simulate one using a Timer, but you must wrap it in an AsyncStream so the consumer can handle it using async/await.

Requirements:

• Create a class MemoryPressureMonitor with a method pressureEvents() -> AsyncStream<MemoryLevel>.
• Define an enum MemoryLevel with cases .normal, .warning, and .critical.
• Inside the AsyncStream, use a Timer to yield a random MemoryLevel every 1 second.
• Crucial: Implement continuation.onTermination to invalidate the timer so it doesn't leak when the task is cancelled.
• In your main code, create a Task that iterates over this stream, but use a filter to only print a message when the level is .warning or .critical.