What is AsyncStream, and how do you use it?

TL;DR

AsyncStream is an AsyncSequence you build yourself. You push values in with continuation.yield(_:) and end it with continuation.finish(). Consumers read it with for await value in stream. Use it to bridge APIs that send many values over time (delegates, callbacks, notifications). For one value, use a continuation instead.

Basic example

let numbers = AsyncStream<Int> { continuation in
    for i in 1...3 {
        continuation.yield(i)
    }
    continuation.finish()
}

for await n in numbers {
    print(n)          // 1, 2, 3
}
print("done")         // after finish()

Bridging a delegate API

A classic interview task: wrap CLLocationManager.

import CoreLocation

final class LocationStream: NSObject, CLLocationManagerDelegate {
    private let manager = CLLocationManager()
    private var continuation: AsyncStream<CLLocation>.Continuation?

    func updates() -> AsyncStream<CLLocation> {
        AsyncStream { continuation in
            self.continuation = continuation
            continuation.onTermination = { [weak self] _ in
                self?.manager.stopUpdatingLocation()
            }
            manager.delegate = self
            manager.startUpdatingLocation()
        }
    }

    func locationManager(_ manager: CLLocationManager, didUpdateLocations locations: [CLLocation]) {
        for location in locations {
            continuation?.yield(location)
        }
    }
}

onTermination runs when the stream finishes or when the consuming task is cancelled. Always clean up there.

`makeStream`: get the continuation outside

Since Swift 5.9 you can create the stream and its continuation together:

let (stream, continuation) = AsyncStream.makeStream(of: String.self)

continuation.yield("hello")
continuation.finish()

for await word in stream { print(word) }

This is handy when you store the continuation in a property.

Throwing streams

AsyncThrowingStream can end with an error.

let downloads = AsyncThrowingStream<Double, Error> { continuation in
    let task = Task {
        do {
            for progress in stride(from: 0.0, through: 1.0, by: 0.25) {
                try await Task.sleep(for: .milliseconds(100))
                continuation.yield(progress)
            }
            continuation.finish()
        } catch {
            continuation.finish(throwing: error)
        }
    }
    continuation.onTermination = { _ in task.cancel() }
}

do {
    for try await p in downloads { print(p) }
} catch {
    print("Failed: \(error)")
}

Buffering

If the producer is faster than the consumer, values wait in a buffer. You choose the policy:

AsyncStream(Int.self, bufferingPolicy: .bufferingNewest(1)) { continuation in
    // only the latest value is kept, older ones are dropped
}

Policy	Keeps	Good for
`.unbounded` (default)	Everything	Every value matters
`.bufferingOldest(n)`	First `n` waiting values	Queue of jobs with a cap
`.bufferingNewest(n)`	Last `n` waiting values	Location, sensor data, UI state

yield returns a result that tells you if a value was dropped.

Consuming in a view model

@MainActor
@Observable
final class MapModel {
    var lastLocation: CLLocation?
    private var task: Task<Void, Never>?

    func start(stream: AsyncStream<CLLocation>) {
        task = Task { [weak self] in
            for await location in stream {
                self?.lastLocation = location
            }
        }
    }

    func stop() { task?.cancel() }
}

Cancelling the task ends the for await loop and triggers onTermination.

Important limits

One consumer. AsyncStream is meant for a single loop. If two tasks iterate the same stream, values are split between them, not copied. For broadcast, create one stream per consumer.
No back-pressure. yield never waits. Use a buffering policy to avoid unbounded memory growth.
Finish it. If you never call finish(), a for await loop waits forever.

Common follow-up questions

AsyncStream vs continuation? A continuation resumes once, for one result. A stream yields many values.
AsyncStream vs Combine publisher? Both deliver values over time. Streams use for await and structured cancellation. Combine has many more operators. Swift Async Algorithms adds operators like debounce and merge for async sequences.
Is AsyncStream.Continuation Sendable? Yes, when the element type is Sendable. You can call yield from any thread.
How do you turn NotificationCenter into a stream? Use the built-in NotificationCenter.default.notifications(named:), which is already an AsyncSequence.
What happens after finish()? Further yield calls are ignored. The loop ends after the buffered values are delivered.

What is AsyncStream, and how do you use it?

#TL;DR

#Basic example

#Bridging a delegate API

#makeStream: get the continuation outside

#Throwing streams

#Buffering

#Consuming in a view model

#Important limits

#Common follow-up questions