async/await and structured concurrency · advanced

AsyncSequence, AsyncStream and continuations

Values that arrive over time, and the two bridges that connect Swift's old callback world to async/await.

12 min read13 min practice0/2 exercises4 recall cards

By the end you will be able to

Consume asynchronous streams of values with for await
Bridge a one-shot callback API with a checked continuation, honouring resume-exactly-once
Bridge a multi-event source with AsyncStream and choose a buffering policy

Guess firstAnswering before you read makes the explanation stick — even when you get it wrong.

await fetchUser() produces one value. What is the async equivalent of "a value that keeps arriving" — location updates, websocket messages, download progress?

`for await` — iteration that waits

An AsyncSequence looks like a collection but produces elements asynchronously:

for await location in locationUpdates {
    updateMap(with: location)
}

Each pass suspends until the next element arrives; the loop ends when the sequence finishes. Throwing sequences use for try await. The familiar operators come along — map, filter, prefix, contains — lazily, element by element:

for try await line in url.lines.prefix(10) { … }        // first 10 lines of a download
for await note in NotificationCenter.default
    .notifications(named: .userLoggedOut) { … }          // notifications as a sequence

You have already used one: for await result in group in the task-groups lesson — a task group is an AsyncSequence of its children's results. And two properties matter in practice: consumption is pull-based (slow consumers naturally back-pressure the loop), and cancellation ends iteration — a .task { for await … } in SwiftUI stops when the view disappears, closing the whole pipeline.

Bridge 1: one callback, one continuation

Vast amounts of real API predate async: fetchThumbnail(completion:), delegate callbacks, CLGeocoder. The bridge for a single-shot callback is a continuation:

func thumbnail(for id: String) async throws -> UIImage {
    try await withCheckedThrowingContinuation { continuation in
        legacyFetchThumbnail(id: id) { image, error in
            if let image {
                continuation.resume(returning: image)
            } else {
                continuation.resume(throwing: error ?? FetchError.unknown)
            }
        }
    }
}

withCheckedThrowingContinuation suspends the async function and hands you a token; calling resume wakes it with a result. The entire contract is one sentence with teeth:

Resume exactly once — on every path.

Resume twice → the checked continuation traps. Loudly, at the double-resume.
Never resume → the async caller is suspended forever. The checked variant logs a leak warning; the task silently never completes — the spinner that never stops.

The word checked is the runtime verification of this contract; withUnsafeContinuation skips the checking for hot paths. Start checked, always.

Predict, then runA wrong prediction you have committed to is worth more than a right answer you read.

This model enforces the continuation contract. Three legacy APIs are bridged — one correct, one that forgets an error path, one that double-fires. What prints?

struct Continuation {
    let name: String
    var resumed = false

    mutating func resume(with value: String) {
        if resumed {
            print("\(name): FATAL — resumed twice")
            return
        }
        resumed = true
        print("\(name): resumed with \(value)")
    }

    func checkOnExit() {
        if !resumed {
            print("\(name): LEAK — never resumed, caller suspended forever")
        }
    }
}

// Correct: every path resumes once.
var good = Continuation(name: "good")
let succeeded = true
if succeeded {
    good.resume(with: "image")
} else {
    good.resume(with: "error")
}
good.checkOnExit()

// Bug 1: the error path forgets to resume.
var leaky = Continuation(name: "leaky")
let found = false
if found {
    leaky.resume(with: "image")
}
leaky.checkOnExit()

// Bug 2: a retry path fires the callback twice.
var doubled = Continuation(name: "doubled")
doubled.resume(with: "first response")
doubled.resume(with: "retry response")
doubled.checkOnExit()

Bridge 2: many events, one AsyncStream

A continuation resumes once; ongoing sources — delegates that fire repeatedly, timers, socket messages — need AsyncStream:

func locationStream() -> AsyncStream<Location> {
    AsyncStream { continuation in
        let delegate = LocationDelegate()
        delegate.onUpdate = { location in
            continuation.yield(location)          // called many times
        }
        delegate.onFinish = {
            continuation.finish()                 // ends the for-await loop
        }
        continuation.onTermination = { _ in
            delegate.stop()                       // consumer cancelled → stop the source
        }
        manager.delegate = delegate
        manager.start()
    }
}

// Consumption reads like any sequence:
for await location in locationStream() { updateMap(location) }

yield feeds elements, finish ends the stream, and onTermination is the piece juniors miss: when the consuming task is cancelled (the view disappeared), the stream tells you so you can turn off the GPS — the resource cleanup half of the bridge.

Buffering is the design decision: the producer may yield faster than the consumer awaits. AsyncStream(bufferingPolicy:) chooses who loses:

Policy	Keeps	Right for
`.unbounded` (default)	everything	must-not-miss events (purchases)
`.bufferingNewest(n)`	latest n, drops oldest	UI state — only the freshest matters
`.bufferingOldest(n)`	first n, drops new	first-wins semantics

bufferingNewest in action — the slow consumer sees only the freshest values.

struct BufferedStream {
    let capacity: Int
    private(set) var buffer: [String] = []
    private(set) var dropped = 0

    mutating func yield(_ value: String) {
        buffer.append(value)
        if buffer.count > capacity {
            buffer.removeFirst()          // bufferingNewest: oldest goes
            dropped += 1
        }
    }

    mutating func next() -> String? {
        if buffer.isEmpty { return nil }
        return buffer.removeFirst()
    }
}

// A sensor produces six readings while the UI is busy rendering:
var stream = BufferedStream(capacity: 3)
for reading in ["10°", "11°", "12°", "13°", "14°", "15°"] {
    stream.yield(reading)
}

print("dropped \(stream.dropped) stale readings")

// The consumer finally catches up:
while let value = stream.next() {
    print("consumed \(value)")
}

The UI missed 10°–12° and never cared — it wanted current temperature, not history. Choosing .unbounded here would waste memory replaying stale values; choosing bufferingNewest for a purchase queue would silently drop revenue. The policy is a product decision wearing an API.

Check yourself

You are wrapping a delegate that fires didReceiveMessage repeatedly for a chat socket. Continuation or AsyncStream — and what must onTermination do?

›

Implement the buffering policies

Write the code

Both policies, side by side — implement yield for each so the outputs match. The difference is one line; the product consequences are the table above.

struct NewestBuffer {
    let capacity: Int
    var buffer: [Int] = []

    mutating func yield(_ value: Int) {
        // keep at most `capacity`, dropping the OLDEST when full
    }
}

struct OldestBuffer {
    let capacity: Int
    var buffer: [Int] = []

    mutating func yield(_ value: Int) {
        // keep at most `capacity`, dropping the NEW value when full
    }
}

var newest = NewestBuffer(capacity: 3)
var oldest = OldestBuffer(capacity: 3)
for value in [1, 2, 3, 4, 5] {
    newest.yield(value)
    oldest.yield(value)
}
print(newest.buffer)
print(oldest.buffer)

Solution unlocks after 3 attempts

›

Guard the continuation contract

Write the code

Build the checked-continuation wrapper: SafeContinuation must allow exactly one resume, report a second as fatal, and report a leak if dropped unresumed.

struct SafeContinuation {
    let name: String
    var resumed = false

    // resume(_:) -> String : first call "resumed: <value>",
    //                        later calls "FATAL: double resume ignored"
    // finish() -> String   : "ok" if resumed, else "LEAK: \(name) never resumed"
}

var c1 = SafeContinuation(name: "fetchAvatar")
print(c1.resume("image.png"))
print(c1.resume("retry.png"))
print(c1.finish())

var c2 = SafeContinuation(name: "fetchProfile")
print(c2.finish())

Solution unlocks after 3 attempts

Design itSaved on this device. Never graded.

Pick a delegate-based API you know — CLLocationManager, a WebSocket library, AVAudioRecorder metering. Design its bridge: continuation or stream (why), what yields, what finishes, what onTermination must clean up, and which buffering policy — justified by what the user loses if events drop.

Checkpoint

You can now:

Consume ongoing values with for await and reason about cancellation
Bridge one-shot callbacks with checked continuations, honouring resume-exactly-once
Bridge event sources with AsyncStream, cleaning up in onTermination
Choose a buffering policy as a product decision

Next up: Combine — the framework this machinery replaced, which your next job's codebase still contains.

How well do you know this now? Rating yourself honestly, then being tested on it, is how you find out where your intuition is wrong.