WWDC21 Meet AsyncSequence (완전 정리본)

원본 영상: https://developer.apple.com/videos/play/wwdc2021/10058/
대상: Swift Concurrency(Async/Await) / AsyncSequence를 처음부터 끝까지 이해하고 싶은 iOS/macOS 개발자

---

목차

1. 개요(Executive Summary)
2. AsyncSequence란 무엇인가
3. 반복(Iteration) — for await, 오류 처리, break/continue, 동시성/취소
4. 표준/파운데이션 제공 비동기 시퀀스(API 모음)
5. 데모 해설 — Quakes CSV(라인 스트리밍) 예제
6. 내 코드 어댑트하기 — AsyncStream / AsyncThrowingStream
7. 조합/연산: map/filter/reduce/first(where:) 등
8. 베스트 프랙티스 — 취소·정리, 버퍼링, UI/MainActor, 테스트 포인트
9. 실전 스니펫(응용 예제)
10. FAQ
11. 레퍼런스/관련 세션

---

1) 개요 (Executive Summary)

• AsyncSequence는 시간의 흐름에 따라 비동기적으로 생성되는 값들의 시퀀스입니다. 익숙한 Sequence 사용감으로, for await ... in으로 순회합니다.
• 반복은 각 요소마다 일시 중단(suspend) 후 재개되는 형태로 동작하며, 정상 종료(더 이상 값 없음) 또는 오류 발생으로 끝날 수 있습니다.
• Foundation은 파일/네트워크/알림 등 즉시 활용 가능한 비동기 시퀀스를 제공합니다: FileHandle.bytes, URL.lines, URLSession.AsyncBytes(bytes(from:)), NotificationCenter.notifications 등.
• 기존 콜백/델리게이트 패턴은 AsyncStream / AsyncThrowingStream으로 간단히 브리지할 수 있으며, onTermination을 통해 취소/정리를 확실히 보장합니다.
• 무한 또는 장기 스트림은 Task 수명과 함께 관리하고 필요 시 cancel()로 안전하게 중단합니다.

---

2) AsyncSequence란 무엇인가

• 정의: 시간이 지나며 0개 이상 값을 방출하고, 끝나면 종료(또는 오류 시 중단)하는 비동기 시퀀스. 종료 후에는 이터레이터의 next()가 nil을 반환합니다.
• 프로토콜 구성:
  • associatedtype Element
  • associatedtype AsyncIterator: AsyncIteratorProtocol
  • func makeAsyncIterator() -> AsyncIterator
  • AsyncIteratorProtocol.next() async throws -> Element?
• 핵심 차이: 일반 Sequence와 유사하지만, 각 요소를 비동기적으로 전달합니다.

---

3) 반복(Iteration) — 핵심 문법·오류·동시성/취소

3.1 기본 문법

for await element in asyncSequence {
  // 요소를 하나씩 비동기적으로 소비
}

• 오류 가능 스트림:

do {
  for try await element in throwingSequence {
    // ...
  }
} catch {
  // 스트림 과정에서 throw된 에러 처리
}

3.2 break / continue 지원

• 일반 for-in과 동일하게, 조건에 따라 조기 종료(break)나 건너뛰기(continue)가 가능합니다.

3.3 컴파일러 관점(개념 이해)

• 컴파일러는 for-in을 내부적으로 이터레이터/while let ... next()로 풀어냅니다.
• 비동기화 시에는 next()를 await 가능한 버전으로 바꾼 형태가 됩니다.

3.4 동시성·취소

• 장기/무한 스트림은 Task로 감싸서 소유자 수명에 묶고, 필요 시 cancel()로 중단합니다.
• 예: 두 스트림을 동시에 소비하고 나중에 취소.

let t1 = Task { for await e in streamA { handle(e) } }
let t2 = Task {
  do { for try await e in streamB { handle(e) } }
  catch { log(error) }
}
// ...나중에
t1.cancel(); t2.cancel()

---

4) 표준/파운데이션 제공 비동기 시퀀스(API 모음)

다음 API들은 iOS 15 / macOS 12 등에서 사용할 수 있습니다.

4.1 파일/표준입력

for try await line in FileHandle.standardInput.bytes.lines {
  // stdin 라인 스트리밍
}

4.2 파일/URL 라인 스트림

for try await line in URL(fileURLWithPath: "/tmp/data.csv").lines {
  // 파일을 라인 단위로 소비
}

4.3 네트워크 바이트 스트림

let (bytes, response) = try await URLSession.shared.bytes(from: url)
guard let http = response as? HTTPURLResponse, http.statusCode == 200 else {
  throw URLError(.badServerResponse)
}
for try await b in bytes {
  consume(b)
}

4.4 알림 스트림

let center = NotificationCenter.default
let firstMatch = await center.notifications(named: .NSPersistentStoreRemoteChange)
  .first { note in
    (note.userInfo?["Key"] as? String) == "ExpectedValue"
  }

---

5) 데모 해설 — Quakes CSV(라인 스트리밍)

• 세션에서는 USGS 지진 CSV를 URL.lines로 읽고, dropFirst()로 헤더를 건너뛰며 각 라인을 파싱해 실시간 처리합니다.
• 포인트: 전체 다운로드를 기다리지 않고, 도착하는 대로 라인이 소비되어 응답성이 높아집니다.

let endpoint = URL(string: "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_month.csv")!
for try await line in endpoint.lines.dropFirst() {
  let parts = line.split(separator: ",")
  // time, lat, lon, magnitude 등 파싱
}

---

6) 내 코드 어댑트 — AsyncStream / AsyncThrowingStream

6.1 콜백/델리게이트 → AsyncStream

final class Monitor {
  var handler: ((Event) -> Void)?
  func start() { /* 이벤트 발생 시 handler?(event) */ }
  func stop()  { /* 정리 */ }
}

func makeStream() -> AsyncStream<Event> {
  AsyncStream(Event.self, bufferingPolicy: .bufferingOldest(100)) { c in
    let m = Monitor()
    m.handler = { c.yield($0) }
    c.onTermination = { _ in m.stop() }  // 취소/종료 시 정리
    m.start()
  }
}

• 요소 방출: yield(_:)
• 정상 종료: finish()
• 취소/종료 정리: onTermination에서 타이머/소켓 해제 등

6.2 오류가 필요할 때 — AsyncThrowingStream

enum StreamError: Error { case failure }
func throwingStream() -> AsyncThrowingStream<Int, Error> {
  AsyncThrowingStream(Int.self) { c in
    // 값: c.yield(v), 오류 종료: c.finish(throwing: error)
  }
}

6.3 버퍼링 정책(백프레셔)

• .unbounded / .bufferingNewest(n) / .bufferingOldest(n) 중 상황에 맞게 선택.
• 실시간 UI는 최신성 위주(Newest), 순서 보존은 Oldest가 유리.

---

7) 조합/연산(Declarative Pipeline)

• map, compactMap, filter, reduce, first(where:), dropFirst, prefix 등 대부분의 익숙한 연산이 비동기 버전으로 제공됩니다.

let numbers = URL(fileURLWithPath: "/tmp/numbers.txt").lines
  .compactMap(Int.init)
  .filter { $0 % 2 == 0 }
  .prefix(10)

for try await n in numbers {
  print(n)
}

---

8) 베스트 프랙티스

1) Task 수명 관리: 화면/소유자 종료 시 task.cancel() 호출.
2) 정리 보장: onTermination에서 관찰자/소켓/타이머 정리.
3) 버퍼링: 생산>소비인 상황을 가정하고 정책 명시.
4) UI/MainActor: UI 갱신은 @MainActor 또는 메인 스레드에서.
5) 에러/종료 시그널: finish()/finish(throwing:)로 명시적 종료가 테스트에 유리.
6) 테스트: 수집/타임아웃/정리 호출 여부를 검증하는 유틸을 활용.

---

---

9) 실전 스니펫

9.1 알림 한 건만 대기(조건 매칭)

let note = await NotificationCenter.default
  .notifications(named: .NSPersistentStoreRemoteChange)
  .first { $0.userInfo?["SomeKey"] as? String == "Target" }

9.2 무한 타이머 스트림

func timerStream(interval: TimeInterval) -> AsyncStream<Date> {
  AsyncStream { c in
    let t = Timer.scheduledTimer(withTimeInterval: interval, repeats: true) { _ in
      c.yield(Date())
    }
    c.onTermination = { _ in t.invalidate() }
  }
}

9.3 최신 검색어만 반영(mapLatest 스타일)

extension AsyncSequence {
  func mapLatest<T>(_ transform: @escaping (Element) async -> T) -> AsyncStream<T> {
    AsyncStream { c in
      let task = Task {
        var latest: Task<Void, Never>?
        for await v in self {
          latest?.cancel()
          latest = Task {
            let out = await transform(v)
            guard !Task.isCancelled else { return }
            c.yield(out)
          }
        }
        c.finish()
      }
      c.onTermination = { _ in task.cancel() }
    }
  }
}

---

10) FAQ

Q. AsyncSequence는 언제 종료되나요?
A. 더 이상 값이 없으면 이터레이터의 next()가 nil을 반환하며 종료합니다. 오류가 발생하면 그 시점에 종료되며 이후 next()는 nil입니다.

Q. 무한 스트림은 어떻게 다뤄야 하나요?
A. Task로 감싸 수명을 소유자와 묶고, 필요 시 cancel()로 중단합니다. onTermination에서 정리를 보장하세요.

Q. 백프레셔는 어떻게 처리하나요?
A. AsyncStream 생성 시 버퍼링 정책을 명시해 생산/소비 속도 차이를 제어합니다.

Q. 에러를 던지는 스트림은 어떻게 만들죠?
A. AsyncThrowingStream<Element, Error>를 사용하고, 오류 발생 시 finish(throwing:)를 호출합니다.

---

11) 레퍼런스/관련 세션

• Meet AsyncSequence (WWDC21/10058) — 세션 영상/트랜스크립트
• AsyncSequence (Swift 표준) — 개념/연산자
• AsyncStream / AsyncThrowingStream — 브리지/버퍼/종료/정리
• URLSession.AsyncBytes — 바이트 스트림
• NotificationCenter.Notifications — 알림을 비동기 시퀀스로