Building Custom AsyncStreams

Sometimes you need to create values on demand rather than iterate over existing ones. That's where AsyncStream comes in.

Follow along with the code: iOS-Practice on GitHub

The Basic Pattern

let countdown = AsyncStream<Int> { continuation in
    for i in (0...10).reversed() {
        continuation.yield(i)
        try? await Task.sleep(nanoseconds: 1_000_000_000)
    }
    continuation.finish()
}

for await number in countdown {
    print(number)  // 10, 9, 8, ... 0
}

Three key operations:

• yield(_:) — emit a value
• finish() — end the stream
• finish(throwing:) — end with an error (AsyncThrowingStream only)

Timer Example

Convert a Timer to an async stream:

func makeTimerStream(interval: TimeInterval) -> AsyncStream<Date> {
    AsyncStream { continuation in
        let timer = Timer.scheduledTimer(withTimeInterval: interval, repeats: true) { _ in
            continuation.yield(Date())
        }

        // Cleanup when stream is cancelled
        continuation.onTermination = { @Sendable _ in
            timer.invalidate()
        }
    }
}

// Usage
for await tick in makeTimerStream(interval: 1.0) {
    print("Tick: \(tick)")
}

The onTermination Handler

Critical for cleanup:

AsyncStream { continuation in
    let resource = ExpensiveResource()
    resource.start()

    continuation.onTermination = { @Sendable termination in
        resource.stop()

        switch termination {
        case .cancelled:
            print("Stream was cancelled")
        case .finished:
            print("Stream finished normally")
        @unknown default:
            break
        }
    }
}

This runs when:

• Consumer stops iterating
• Task is cancelled
• You call finish()

Event Emitter Pattern

Perfect for converting callback-based event sources:

class WebSocketWrapper {
    func messages() -> AsyncStream<String> {
        AsyncStream { continuation in
            self.onMessage = { message in
                continuation.yield(message)
            }

            self.onClose = {
                continuation.finish()
            }

            continuation.onTermination = { @Sendable _ in
                self.disconnect()
            }
        }
    }
}

Yielding from Anywhere

The continuation is thread-safe. Yield from any context:

AsyncStream { continuation in
    DispatchQueue.global().async {
        continuation.yield("From background")
    }

    DispatchQueue.main.async {
        continuation.yield("From main")
    }
}

Common Mistake: Forgetting to Finish

// BAD: Stream never ends
AsyncStream<Int> { continuation in
    continuation.yield(1)
    continuation.yield(2)
    // Consumer waits forever for next value!
}

// GOOD: Signal completion
AsyncStream<Int> { continuation in
    continuation.yield(1)
    continuation.yield(2)
    continuation.finish()  // Consumer's loop exits
}

Interview Tip

AsyncStream is how you bridge imperative code (callbacks, delegates, timers) to structured concurrency. When asked about modernizing legacy code, this is often the answer.