AsyncStream

iOS/Swift Concurrency

AsyncStream

Hamp 2025. 5. 25. 17:11

👋 들어가기 전

이번 포스팅은 컨커런시에서 제공해주는 비동기 반복문에 사용되는 객체를

내가 직접 만들기 위한 여정을 알아보자.

비동기 반복문은 for awiat in ... 형태인데

여기서 ...에 들어가는 내용이 바로 AsyncStream이다.

또한 실제 동기 for문에서도 쓰이는 여러가지 collection들은 반복문들과 사용하기위해

Sequence + IteratorProtocol을 채택하고 있다.

비동기 반복문 역시도 AsyncSequence + AsyncIteratorProtocol을 채택해야한다.

차례대로 알아보자.

🏁 학습할 내용

AsyncSequence
AsyncIteratorProtocol
AsyncStream

🚂 AsyncSequence

먼저 알아볼 내용은 AsyncSequence 프로토콜이다.

AsyncSequence = Async(비동기) + Seqeunce

Sequence

시퀀스 프로토콜은 갇나히 말하면 Iterator(반복자)를 생성해내는 프로토콜이다.

struct CustomSequence: Sequence {
  func makeIterator() -> CustomIterator {

  }
}

정리하면 AsyncSequence는 비동기 반복자를 만들어내는 프로토콜이다.

♻️ AsyncIteratorProtocol

다음은 AsyncIteratorProtocol이다.

이 내용도 마찬가지로 2개로 분리할 수 있다.

AsyncIteratorProtocol = Async(비동기) + IteratorProtocol

IteratorProtocol

Sequence 프로토콜은 반복자를 생성해내는 프토토콜이면

IteratorProtocol은 그 반복자를 정의하는 프로토콜이다.

여기서 next함수의 리턴타입이 바로, 반복문이 뱉어내는 Element 타입의 옵셔널 형태이다.

struct CustomIterator: IteratorProtocol {
  func next() -> Int? {

  }
}

두 프로토콜틀을 조금 길게쓰면 다음과 같다.

struct CustomIterator: IteratorProtocol {
  typealias Element = Int
  func next() -> Int? {

  }
}

struct CustomSequence: Sequence {
  typealias Element = Int
  typealias Iterator = CustomIterator
  
  func makeIterator() -> CustomIterator {

  }
}

두개 다 Async 버전으로 하면 함수가 async로 바뀌고 명칭만 조금 바뀔뿐

struct AsyncCustomIterator: AsyncIteratorProtocol {
  typealias Element = Int
  func next() async -> Int? {

  }
}

struct AsyncCustomSequence: AsyncSequence {
  typealias Element = Int
  typealias AsyncIterator = AsyncCustomIterator

  func makeAsyncIterator() async -> AsyncCustomIterator {

  }
}

🫷AsyncStream

⭐️ 정의

AsyncStream은 위에서 말한 비동기 반복문에 사용될 AsyncSequence를 생산하는 구조체다.

🏭 생성 방법

생성 방법은 크게 2가지 방법이 있다.

그 중 continuation을 이용하는 방법이 있는데, 이 방법은 continuation을 생성했을 때

배운 목적과 아주 일치한다.

바로 이전 API와 호환성이 좋다.

🪜 Continuation 이용

대표적인 공통점은 똑같이 error 여부로 Throwing이 붙냐 안붙냐 차이가있다.

차이점은 continuation 생성할 때는 , resume 키워드를 사용했지만.

stream 생성 시에는 yiled와 finish를 사용한다.

let stream1 = AsyncStream<Int> { continuation in
  continuation.finish() // 종료
  continuation.yield(element 타입)
  continuation.yield(with: Result 타입)
}

struct CustomError: Error {
}

let stream2 = AsyncThrowingStream<Int, CustomError> { continuation in
  continuation.finish()
  continuation.finish(throwing: CustomError())
  continuation.yield(element타입)
}

🗂️ unfolding

간단한 비동기 반복 작업을 생성할 때 사용, 자주 사용되지는 않는다.

약간 테스트 할 때 사용할만한 ??

let stream1 = AsyncStream(unfolding: {
    return Int.random(in: 0..<Int.max)
})

struct CustomError: Error {}

let stream2 = AsyncThrowingStream {
  throw CustomError()
}

📦 makeStream

마지막 방법은 바로 makeStream 메서드를 이용하는 방법이다.

이 방법의 특이한 점은 continuatino과 stream이 튜플 형태로 반환되기 때문에

클로저 방식처럼 안에 있는 것보다 더 많은 유연성을 제공한다.

버퍼 정책

unbounded: 제한 없음
bufferingOldest(number): 버퍼가 가득차면, 가장 오래된 number개까지 유지
bufferingNewest(number): 버퍼가 가득차면, 가장 최근 number개까지 유지

import Foundation

let (stream, continuation) = AsyncStream<Int>.makeStream(bufferingPolicy: .unbounded)


Task {
  for await st in stream {
    print(st)
  }

}


continuation.yield(1)
continuation.yield(2)
continuation.yield(3)
continuation.yield(4)

약간 Rx나 Combine 같은 느낌이다.

관찰은 stream으로 하고 값은 continuation yield를 통해보내고