AsyncThrowingStream and Error Handling

When your stream can fail, use AsyncThrowingStream. It's like AsyncStream but propagates errors to consumers.

Follow along with the code: iOS-Practice on GitHub

Basic Throwing Stream

enum StreamError: Error {
    case connectionLost
    case invalidData
}

let dataStream = AsyncThrowingStream<Data, Error> { continuation in
    networkClient.onData = { data in
        continuation.yield(data)
    }

    networkClient.onError = { error in
        continuation.finish(throwing: error)
    }

    networkClient.onComplete = {
        continuation.finish()
    }
}

Consuming with Error Handling

do {
    for try await data in dataStream {
        process(data)
    }
    print("Stream completed successfully")
} catch {
    print("Stream failed: \(error)")
}

The try keyword is required. When finish(throwing:) is called, the error propagates to the consumer.

Simulating Failures

Useful for testing:

func makeUnreliableStream() -> AsyncThrowingStream<Int, Error> {
    AsyncThrowingStream { continuation in
        Task {
            for i in 1...10 {
                try? await Task.sleep(nanoseconds: 500_000_000)

                // Random failure
                if i == 7 && Bool.random() {
                    continuation.finish(throwing: StreamError.connectionLost)
                    return
                }

                continuation.yield(i)
            }
            continuation.finish()
        }
    }
}

Retry Pattern

Wrap a throwing stream with retry logic:

func withRetry<T>(
    maxAttempts: Int,
    stream: @escaping () -> AsyncThrowingStream<T, Error>
) -> AsyncThrowingStream<T, Error> {
    AsyncThrowingStream { continuation in
        Task {
            var attempts = 0

            while attempts < maxAttempts {
                do {
                    for try await value in stream() {
                        continuation.yield(value)
                    }
                    continuation.finish()
                    return
                } catch {
                    attempts += 1
                    if attempts >= maxAttempts {
                        continuation.finish(throwing: error)
                        return
                    }
                    // Wait before retry
                    try? await Task.sleep(nanoseconds: 1_000_000_000)
                }
            }
        }
    }
}

Error vs Completion

Be clear about the difference:

// Normal completion — stream ended successfully
continuation.finish()

// Error completion — something went wrong
continuation.finish(throwing: NetworkError.timeout)

Consumers can distinguish these:

do {
    for try await item in stream {
        handle(item)
    }
    // Only reaches here if finish() was called (no error)
    print("All items processed")
} catch {
    // Only reaches here if finish(throwing:) was called
    print("Processing failed: \(error)")
}

When to Use Each

Interview Tip

Discuss error boundaries. With AsyncThrowingStream, you're saying "the consumer should handle failures." With AsyncStream, you're saying "I'll handle errors internally." This decision affects your API design.