AsyncSequence and AsyncStream for Swift & iOS developers
Most of us have battled callback hell when handling continuous data flows—whether sensors firing events, network messages arriving, or UI state changing over time. AsyncSequence and AsyncStream for Swift & iOS developers solve this by letting you consume streams of values using familiar async/await syntax, turning nested callbacks into readable sequential code. On day 58 of building production iOS apps, understanding these tools becomes essential when you need to bridge legacy callback-based APIs or handle real-time data sources.
Why async streams matter in modern iOS development
Before Swift's concurrency model, handling continuous data meant either storing completion handlers in properties, managing complex delegate chains, or wrestling with Combine's publishers. AsyncSequence provides a simpler, more intuitive mental model: iterate over values as they arrive, just like a regular array, except the values come asynchronously over time.
This approach shines in three concrete scenarios we encounter regularly:
- WebSocket connections receiving messages
- Core Location or Core Motion sensor streams
- File system notifications or database change events
The syntax feels natural because you use for await instead of for, and the compiler ensures proper cancellation when the surrounding task ends.
For await loops over values
The simplest entry point is understanding for await syntax. Unlike a regular loop that consumes a finite collection, a for await loop consumes an AsyncSequence, pausing at each value until the next one arrives.
Here's a practical example monitoring location updates from Core Location:
import CoreLocation
func monitorUserLocation() async {
let manager = CLLocationManager()
let locations = manager.locationsStream // hypothetical AsyncSequence
for await location in locations {
print("New location: \(location.coordinate)")
// Update map, send to server, etc.
}
// Loop exits when locationsStream completes or task cancels
}
In real production code, Core Location doesn't ship an AsyncSequence by default, so you'd either wrap it yourself or use a third-party library. The principle remains: each iteration waits for the next value without blocking the thread.
Cancellation is automatic. If the surrounding task cancels—because the user dismissed the screen, for example—the loop exits cleanly. You don't manually unsubscribe or clean up listeners.
You can also add conditional logic and break early:
for await location in locations {
if location.horizontalAccuracy < 50 {
// Accurate enough, stop waiting
break
}
}
AsyncStream bridges callbacks to async/await
Most of the code we maintain doesn't use AsyncSequence natively. Instead, we receive callbacks, delegates, or notifications. AsyncStream is the bridge that converts those patterns into streamable sequences.
Imagine a third-party SDK that sends WebSocket messages via a callback:
class WebSocketClient {
var onMessageReceived: ((String) -> Void)?
func connect(url: URL) {
// ... connection logic
}
}
Wrap this in AsyncStream to make it async/await friendly:
func messageStream(from client: WebSocketClient) -> AsyncStream<String> {
AsyncStream { continuation in
client.onMessageReceived = { message in
continuation.yield(message)
}
}
}
// Usage
let client = WebSocketClient()
await client.connect(url: URL(string: "wss://echo.websocket.org")!)
for await message in messageStream(from: client) {
print("Received: \(message)")
}
The continuation object is your valve. Call yield(value) to emit a value downstream, and finish() to end the stream. AsyncStream handles the transition from callback-land to async-land seamlessly.
For more complex scenarios with error handling, use AsyncThrowingStream:
func messageStream(from client: WebSocketClient) -> AsyncThrowingStream<String, Error> {
AsyncThrowingStream { continuation in
client.onMessageReceived = { message in
continuation.yield(message)
}
client.onError = { error in
continuation.finish(throwing: error)
}
}
}
// Usage with error handling
do {
for try await message in messageStream(from: client) {
print("Received: \(message)")
}
} catch {
print("Stream error: \(error)")
}
Cancellation with the surrounding task
Cancellation is where AsyncSequence and AsyncStream shine compared to older patterns. When the task containing your for await loop gets cancelled—whether explicitly or because its scope ended—the loop exits immediately and resources clean up.
Consider monitoring sensor data that should stop when the view disappears:
@MainActor
class SensorViewModel: NSObject, ObservableObject {
@Published var acceleration: (x: Double, y: Double, z: Double) = (0, 0, 0)
private var motionTask: Task<Void, Never>?
func startMonitoring() {
motionTask = Task {
let manager = CMMotionManager()
manager.startAccelerometerUpdates()
for await data in manager.accelerometerStream {
self.acceleration = (data.acceleration.x, data.acceleration.y, data.acceleration.z)
}
// Cleanup happens here automatically when task cancels
}
}
func stopMonitoring() {
motionTask?.cancel() // Loop exits gracefully
}
deinit {
motionTask?.cancel()
}
}
When stopMonitoring() is called, the task cancels. The for await loop detects this and exits without executing further iterations. No dangling observers, no manual unsubscribe calls.
Inside AsyncStream, you can detect cancellation using a deferred cleanup block:
func sensorStream() -> AsyncStream<SensorData> {
AsyncStream { continuation in
let task = DispatchSourceTimer()
task.setEventHandler {
continuation.yield(/* current sensor data */)
}
task.resume()
continuation.onTermination = { @Sendable _ in
task.cancel()
}
}
}
The onTermination block runs when the stream ends, whether by completion, cancellation, or error. Use it to clean up timers, file handles, or network connections.
Great for WebSockets and sensors
WebSocket connections and sensor streams are where AsyncSequence and AsyncStream truly earn their place in production code. Both involve continuous, unpredictable data arrival and tight cancellation requirements.
WebSocket example: A trading app displaying live price ticks.
func priceTickStream(for symbol: String) -> AsyncThrowingStream<PriceTick, Error> {
AsyncThrowingStream { continuation in
let ws = URLSessionWebSocketTask(url: URL(string: "wss://quotes.example.com")!)
ws.resume()
Task {
do {
while true {
let message = try await ws.receive()
if case .string(let jsonString) = message,
let tick = try? JSONDecoder().decode(PriceTick.self, from: jsonString.data(using: .utf8)!) {
continuation.yield(tick)
}
}
} catch {
continuation.finish(throwing: error)
ws.cancel(with: .goingAway, reason: nil)
}
}
}
}
// Usage
@MainActor
class TickerViewModel: ObservableObject {
@Published var latestTick: PriceTick?
private var tickerTask: Task<Void, Never>?
func watch(symbol: String) {
tickerTask = Task {
do {
for try await tick in priceTickStream(for: symbol) {
self.latestTick = tick
}
} catch {
print("Ticker error: \(error)")
}
}
}
func stopWatching() {
tickerTask?.cancel()
}
}
Sensor example: Accelerometer data for a fitness app.
extension CMMotionManager {
var accelerometerStream: AsyncStream<CMAccelerometerData> {
AsyncStream { continuation in
let queue = OperationQueue()
startAccelerometerUpdates(to: queue) { data, _ in
if let data = data {
continuation.yield(data)
}
}
continuation.onTermination = { @Sendable _ in
self.stopAccelerometerUpdates()
}
}
}
}
// Usage
func detectActivity() async {
let manager = CMMotionManager()
var stepCount = 0
for await accel in manager.accelerometerStream {
if accel.acceleration.z > 1.2 {
stepCount += 1
print("Steps: \(stepCount)")
}
}
}
Common mistakes
Forgetting that for await loops must run in async context. You can't call a for await loop from synchronous code. Wrap it in a Task if needed, but verify you're not losing the async structure.
Storing a continuation outside AsyncStream. Continuations are only valid within their closure scope. Don't capture and reuse them across function calls—create a new AsyncStream each time.
Not handling cancellation in custom AsyncStream closures. If you spawn background tasks inside AsyncStream, they won't automatically cancel. Use onTermination to clean up, or structure your code so background work is tied to the continuation's lifetime.
Mixing AsyncSequence with sync iteration. A for loop won't work on AsyncSequence; you must use for await. The compiler will catch this, but it's an easy mental slip.
Ignoring backpressure. AsyncStream has a default buffer size. If you yield faster than the consumer processes, values queue up in memory. For high-frequency sources, consider limiting yield rate or implementing custom buffering strategies.
Key takeaways
- Use
for awaitloops to consume AsyncSequence values, replacing callback-based iteration with readable async/await code. - AsyncStream wraps callback-based APIs (delegates, completion handlers, notifications) into sequences you can iterate over asynchronously.
- Cancellation is automatic and clean: when the surrounding task cancels, the loop exits and
onTerminationblocks run, freeing resources without manual cleanup. - WebSocket connections and sensor streams are prime use cases where AsyncSequence eliminates boilerplate and improves code clarity.
- AsyncThrowingStream handles errors alongside values, letting you propagate failures through the stream naturally.
- Be cautious with continuation lifetime, cancellation handling in nested tasks, and buffer sizes when yielding high-frequency data.
Comments
Loading comments…