AsyncSequence
Swift Concurrency에서 async/await를 사용하면, await 지점에서 현재 스레드의 제어권을 시스템에 넘기고 다른 작업이 실행될 수 있도록 한다. 비동기 작업이 완료되면 다시 해당 함수의 실행을 재개하며, await 이후의 코드를 이어서 수행한다.
만약 여러 개의 비동기 결과가 연속적으로 필요한 상황에서도 async/await로 코드를 작성하는 것이 가능할까?
이 질문의 대답으로 AsyncSequence를 사용하면 비동기적인 결과를 연속적으로 전달 받는 것이 가능해진다.
Combine과 AsyncSequence
AsyncSequence는 Combine으로 구성된 데이터 스트림 형태의 코드에서 Swift concurrency 기능을 혼용해서 사용할 수 있도록 브릿지 역할을 해줄 수 있다.
아래와 같이 간단한 정수를 순차적으로 방출하는 Publisher가 있다고 가정해보자.
import Combine
let numbers = [1, 2, 3, 4, 5]
let publisher = numbers.publisher이때 publisher의 values 프로퍼티를 사용하면 AsyncSequence 프로토콜를 따르는 AsyncPublisher를 얻을 수 있다. 이를 통해 for-await-in 구문으로 비동기 스트림을 마치 일반 for 문처럼 간단하게 순회할 수 있다.
let numberSequence = publisher.values
for await number in numberSequence {
print(number)
}연산자
AsyncSequence는 기존의 Sequence 프로토콜과 유사하게 map, filter, compactMap 등 다양한 연산자를 제공한다. 이를 활용하면 비동기 스트림에 대해서도 익숙한 방식으로 변환이나 필터링 같은 고차함수를 적용할 수 있다.
let numberSequence = publisher.values.filter { $0 % 2 == 0 }
for try await number in numberSequence {
print(number)
}AsyncStream
Combine 프레임워크 외에 레거시 코드에서 completion handler 콜백 함수를 활용했을 때, 연속적인 비동기 값도 async/await를 사용하여 Swift Concurrency로 브릿징할 수 있을까?
이러한 상황에서, AsyncStream은 임의의 비동기 콜백이나 이벤트 소스를 AsyncSequence로 감싸는 도구로, 기존 콜백 기반 API를 자연스럽게 Swift Concurrency로 통합할 수 있도록 해준다.
import Foundation
let numberSequence = AsyncStream { continuation in
DispatchQueue.global().async {
for i in 1...5 {
continuation.yield(i)
}
continuation.finish()
}
}
for await number in numberSequence {
print(number)
}위의 코드에서 볼 수 있듯, yield는 스트림으로 Element를 전달하고, finish는 해당 스트림을 종료하는 메서드이다.
AsyncThrowingStream
AsyncStream 자체로는 에러를 발생 시키지 않는다. 만약 비동기 시퀀스가 값을 지속적으로 방출하면서도 상황에 따라 에러를 발생시켜야 한다면, 오류 처리까지 지원하는 AsyncThrowingStream을 사용해야 한다.
사용 방법은 AsyncStream과 크게 다르지 않다. 다만, 오류를 스트림으로 전달 할 수 있다는 점이 다르며, for-try-await-in을 통해 스트림에서 전달하는 값을 하나씩 전달 받을 수 있다는 점이 다르다.
import Foundation
let numberSequence = AsyncThrowingStream { continuation in
DispatchQueue.global().async {
for i in 1...5 {
continuation.yield(i)
}
if Bool.random() {
continuation.finish(throwing: NSError(domain: "error", code: -1))
} else {
continuation.finish()
}
}
}
for try await number in numberSequence {
print(number)
}onTermination
AsyncStream.Continuation과 AsyncThrowingStream.Continuation에 onTermination을 활용하면, 스트림이 종료 되었을 때와 취소 되었을 때를 감지하여, 수행할 작업을 정의할 수 있다.
import Foundation
let numberSequence = AsyncStream<Int> { continuation in
continuation.onTermination = { termination in
switch termination {
case .finished:
print("finished")
case .cancelled:
print("cancelled")
}
}
for i in 1...5 {
print(i)
continuation.yield(i)
}
continuation.finish()
}예를 들어, 위의 코드에서는 스트림이 종료될 때 실행할 로직을 onTermination에 클로저로 등록하고 있다. 스트림이 정상적으로 finish()를 통해 끝난 경우에는 .finished가, 사용되는 측에서 Task.cancel() 등의 이유로 취소된 경우에는 .cancelled가 전달된다.
위 예제에서는 스트림이 종료될 때마다 종료 상태에 따라 메시지를 출력하도록 구현하여, 스트림의 수명 주기를 명확하게 추적할 수 있다. 이를 응용하면 리소스 정리, 네트워크 스트림 해제, 파일 핸들 닫기 등 다양한 종료 처리 로직을 구현할 수 있다.
정리
지금까지 AsyncSequence에 대해 알아보았다. AsyncSequence는 여러 개의 비동기 값을 순차적으로 처리할 수 있도록 도와주며, async/await와 결합해 더 간결한 방식으로 비동기 스트림에 접근할 수 있는 API를 제공한다.
위에서 간단히 몇 가지 연산자를 살펴보았지만, AsyncSequence의 가장 큰 장점 중 하나는 for-await-in 구문을 사용하기 때문에 기존 for 문에서 사용하던 제어 흐름 키워드를 그대로 활용할 수 있다는 점이다.
예를 들어, 특정 조건을 만족하지 않는 값은 건너뛰고 싶다면 filter 연산자를 사용하지 않고도 continue로 간단히 처리할 수 있다.
for try await number in numberSequence {
guard number.isMultiple(of: 2) else { continue }
print(number)
}위 예제처럼 continue를 활용하면 원하는 조건만 골라 처리할 수 있다. 이는 복잡한 조건을 가진 스트림에서 연산자 체이닝 없이도 명확한 흐름 제어가 가능하다는 장점이 있다.
또한 루프를 완전히 중단하고 싶다면 break를 사용할 수도 있다.
for try await number in numberSequence {
guard number < 3 else { break }
print(number)
}여기서 break를 사용하면 해당 AsyncSequence의 반복을 즉시 종료시킨다.
즉, 비동기 스트림을 다루는 동안에도 조건 분기, 루프 건너뛰기, 조기 종료와 같은 제어 흐름을 기존 방식 그대로 활용할 수 있다는 것이다.
