7. AsyncSequence, AsyncStream & continuations

What & why

Sequence produces values you can pull synchronously. An AsyncSequence produces values over time, each one awaited: lines arriving from a socket, events from a sensor, results from a task group, notifications. You consume them with for await, and the usual operators (map, filter, reduce…) work on them too.

Consuming: for await

for await number in Countdown(from: 3) {
    print(number)
}

// With throwing sequences:
for try await line in URL(string: "file:///tmp/log.txt")!.lines {
    print(line)
}

The loop body runs once per element; between elements the task is suspended, not blocked. Cancelling the task ends the loop.

Some built-in async sequences:

• URL.lines, FileHandle.bytes / .AsyncBytes — stream a file or network resource.
• NotificationCenter.notifications(named:) — notifications as a stream.
• TaskGroup / ThrowingTaskGroup — results as tasks complete.
• AsyncStream / AsyncThrowingStream — build your own (below).

Reading a large file line by line without loading it into memory:

import Foundation

func countLines(in path: String) async throws -> Int {
    var count = 0
    for try await _ in URL(fileURLWithPath: path).lines {
        count += 1
    }
    return count
}

Operators

Async sequences get the same vocabulary as regular ones:

let squares = Countdown(from: 10)
    .filter { $0.isMultiple(of: 2) }
    .map { $0 * $0 }

for await s in squares { print(s) }                 // 100 64 36 16 4

let total = await Countdown(from: 4).reduce(0, +)   // 10
let firstBig = await Countdown(from: 100).first { $0 < 50 }   // 49
let all = await Countdown(from: 3).reduce(into: [Int]()) { $0.append($1) }   // [3, 2, 1]

For richer operators — merge, zip, debounce, throttle, chunks, AsyncChannel — use Apple's swift-async-algorithms package.

Implementing AsyncSequence

struct Countdown: AsyncSequence {
    typealias Element = Int
    let start: Int

    init(from start: Int) { self.start = start }

    struct AsyncIterator: AsyncIteratorProtocol {
        var current: Int
        mutating func next() async -> Int? {
            guard current > 0 else { return nil }
            defer { current -= 1 }
            try? await Task.sleep(for: .milliseconds(100))   // simulate waiting
            return current
        }
    }

    func makeAsyncIterator() -> AsyncIterator { AsyncIterator(current: start) }
}

next() returns nil when finished. Add throws to the iterator and declare typealias Failure for sequences that can fail. (Swift 6 also supports typed failure: AsyncSequence<Int, Never>.)

AsyncStream: the easy way

You rarely implement iterators by hand. AsyncStream gives you a continuation to push values into:

let ticks = AsyncStream<Int> { continuation in
    let task = Task {
        for i in 1...5 {
            try? await Task.sleep(for: .milliseconds(200))
            continuation.yield(i)           // deliver a value
        }
        continuation.finish()               // end the stream
    }
    continuation.onTermination = { _ in task.cancel() }    // cleanup when the consumer stops
}

for await tick in ticks { print("tick", tick) }

Key points:

• yield is synchronous and non-blocking; it buffers when the consumer is slow.
• Control the buffer: AsyncStream(Int.self, bufferingPolicy: .bufferingNewest(10)) { ... } — drop old values rather than growing without bound.
• Always call finish() (or throw for AsyncThrowingStream), or consumers wait forever.
• onTermination runs when the consumer cancels or the stream finishes — release resources there.
• A stream is single-consumer by default: don't iterate the same stream from two tasks.

Bridging a callback/delegate API:

final class Sensor {
    var onReading: ((Double) -> Void)?
    func start() { /* delivers readings via onReading */ }
    func stop() {}
}

extension Sensor {
    var readings: AsyncStream<Double> {
        AsyncStream { continuation in
            self.onReading = { continuation.yield($0) }
            continuation.onTermination = { [weak self] _ in self?.stop() }
            self.start()
        }
    }
}

AsyncStream.makeStream(of:) (Swift 5.9+) returns the stream and its continuation separately — handy when the producer lives elsewhere:

let (stream, continuation) = AsyncStream.makeStream(of: String.self)
Task {
    for await message in stream { print("got", message) }
}
continuation.yield("hello")
continuation.yield("world")
continuation.finish()

Continuations: bridging one-shot callbacks

For a callback that fires once, use a checked continuation:

func askName() async -> String {
    await withCheckedContinuation { continuation in
        legacyAskName { name in
            continuation.resume(returning: name)    // must be called exactly once
        }
    }
}
func legacyAskName(_ completion: @escaping (String) -> Void) { completion("Ada") }

func load() async throws -> Data {
    try await withCheckedThrowingContinuation { continuation in
        legacyLoad { result in continuation.resume(with: result) }
    }
}
func legacyLoad(_ completion: @escaping (Result<Data, Error>) -> Void) { completion(.success(Data())) }

Rules:

• Resume exactly once. Never resuming hangs the awaiting task forever (checked continuations log a warning if the continuation is leaked). Resuming twice is a bug (checked continuations trap in debug).
• withUnsafeContinuation skips the checks — use only after profiling proves it matters.
• If the callback API can be cancelled, combine with withTaskCancellationHandler.

Merging concurrent producers

A task group can feed one AsyncStream:

func combined(_ streams: [AsyncStream<Int>]) -> AsyncStream<Int> {
    AsyncStream { continuation in
        let task = Task {
            await withTaskGroup(of: Void.self) { group in
                for stream in streams {
                    group.addTask { for await v in stream { continuation.yield(v) } }
                }
            }
            continuation.finish()
        }
        continuation.onTermination = { _ in task.cancel() }
    }
}

(swift-async-algorithms offers merge(_:_:) that handles this robustly.)

Back-pressure and buffering

AsyncStream is a push source: the producer doesn't wait for the consumer. If the producer is faster, values pile up in the buffer. Choose a policy deliberately:

Policy	Behaviour
.unbounded (default)	keep everything — can grow without limit
.bufferingNewest(n)	keep the latest n; drop older ones (good for UI state)
.bufferingOldest(n)	keep the first n; drop newer ones

If you need genuine back-pressure (producer waits when the consumer is slow), use AsyncChannel from swift-async-algorithms, whose send suspends until a consumer takes the value.

Common mistakes

• Forgetting finish() so for await never ends.
• Resuming a continuation twice, or not at all (e.g. on an early-return path or error branch).
• Two consumers on one AsyncStream — values are split unpredictably.
• Unbounded buffers with a fast producer — memory growth.
• Not handling cancellation in custom iterators (Task.checkCancellation() / Task.isCancelled).
• Retaining self strongly in onTermination and creating cycles.
• Doing blocking work inside next().

Exercises

1. Implement Fibonacci: AsyncSequence that yields the first n values with a short delay, and consume it with for await and .prefix(5).
2. Wrap Timer (or a DispatchSourceTimer) as an AsyncStream<Date> that cleans up on termination.
3. Convert a callback-based function to async with a checked continuation, then deliberately forget to resume on one branch and observe the warning.
4. Write func lines(of path: String) -> AsyncThrowingStream<String, Error> using a background FileHandle read loop (not URL.lines).
5. Build a debounce operator of your own for AsyncStream<String>, then compare it with swift-async-algorithms.
6. Merge three timers ticking at different rates into one stream and print values for five seconds, then cancel.

Recap

• AsyncSequence + for await consume values over time; operators like map/filter/reduce work.
• AsyncStream / AsyncThrowingStream build sequences from callbacks and tasks — mind buffering and finish().
• Continuations bridge one-shot callbacks to async functions: resume exactly once.
• Cancellation, cleanup (onTermination), and back-pressure are your responsibility.

Next: Ownership: noncopyable types & borrowing →