[Swift] Understanding Swift Concurrency’s AsyncStream

이전 글에서 기존의 비동기 코드를 async/await을 활용하는 Swift의 새로운 동시성 시스템으로 연결할 수 있는 방법에 대해 설명했습니다. 여기서 설명한 메커니즘은 코드가 단일 값으로 모델링할 수 있는 단일 결과를 생성하는 코드에 적합합니다.

 

다운로드 진행률, 사용자의 현재 위치와 같이 시간에 따라 여러 값을 제공하는 경우, 비동기 for 루프를 사용하여 반복할 수 있는 AsyncSequence 객체로 모델링할 수 있습니다. 이에 대한 기본적인 예제로 URL의 lines 속성을 들 수 있습니다.

let url = URL(string: "<https://donnywals.com>")!

for try await line in url.lines {
    // use line
}

그렇다면 커스텀 비동기 시퀀스를 생성하는 가장 좋은 방법은 무엇일까요?

AsyncSequence 프로토콜을 구현하고 AsyncIterator에서 빌드하는 것은 지루하고 오류가 발생하기 쉽습니다. 다행히도 이런 작업을 할 필요가 없습니다.

이 글에서는 Swift의 AsyncStream을 활용하여 필요할 때마다 값을 생성하는 사용자 정의 async 시퀀스를 빌드하는 방법을 보여드리겠습니다.

 

간단한 비동기 스트림 (async stream) 생성하기

비동기 스트림은 다양한 방법으로 생성할 수 있습니다. 비동기 스트림을 생성하는 가장 쉬운 방법은AsyncStream(unfolding:) 생성자를 이용하는 것입니다.

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

물론 이 예제가 유용하진 않지만, AsyncStream(unfolding:)의 개념에 얼마나 간단한지 알 수 있습니다.

우리는 AsyncStream을 사용하여 비동기 스트림을 생성하고 반환할 값을 생성할 수 있습니다. unfolding에 전달되는 클로저는 비동기이므로 클로저 내에서 비동기 연산을 기다릴 수 있습니다. 스트림에 대한 값을 생성하기 시작할 것으로 예상될 떄마다 unfolding 클로저가 호출됩니다. 실제로는 클로저가 호출되어 작업을 수행하고, 값을 반환한 다음 클로저가 호출된다는 뜻입니다. 이 과정은 for 루프가 취소되거나 비동기 for loop를 포함하는 작업이 취소되거나, unfolding 클로저에서 nil을 반환할 때까지 계속됩니다.

 

AsyncStream(unfolding:)을 사용하여 값 스트림을 생성하는 방법은 매우 편리하지만 특히 다음과 같은 상황에서 유용합니다.

• 값을 생성하기 위해 기다려야 하는 비동기 작업을 수행하려는 경우
• API를 브릿징하는 동안 백 프레셔를 처리해야 하는 경우

델리게이트를 기반으로 하는 기존 API를 브리징하거나 콜백을 활용하여 결과를 전달하는 API의 경우, AsyncStream(unfolding:)을 사용할 수 없을 것입니다.

AsyncStream(unfolding:)은 가장 간단하고 오류가 적게 발생하는 방법이지만, 기존 코드를 Swift Concurrency로 연결하는 데 가장 제한적이며 적합하지 않은 경우가 많습니다.

AsyncStream에 대한 continuation API에서 더 많은 유연성을 찾을 수 있습니다.

continuation을 사용하여 비동기 스트림 (async stream) 생성하기

비동기 클로저가 자체 비동기 스트림을 생성하는 데 적합하지 않은 경우, continuation 기반 접근 방식이 훨씬 더 나은 해결책이 될 수 있습니다. continuation을 사용하면 비동기 스트림 객체를 구성하고 값을 사용할 수 있게 될 때마다 비동기 스트림을 통해 값을 전송할 수 있습니다.

 

다음과 같이 AsyncStream(build:) 생성자를 사용해 비동기 스트림을 생성할 수 있습니다.

let stream2 = AsyncStream { continuation in
	continuation.yield(Int.random(in: 0..<Int.max))
}

위의 예제는 하나의 정수 값을 생성하는 비동기 스트림을 생성합니다. 이 값은 continuation에서 yield를 호출하여 생성합니다. 전송할 값이 있을 때마다 continuation에서 yield를 호출해야 합니다.

 

델리게이트 기반 API를 래핑하는 AsyncStream을 생성할 경우, 델리게이트 객체에 continuation을 보유 (유지)하고 관련 델리게이트 메서드가 호출될 때마다 yield를 호출할 수 있습니다.

 

예를 들어, 새로운 사용자의 위치를 사용할 수 있을 때마다 CLLocationManagerDelegate 내에서 contiunation.yield를 호출할 수 있습니다.

class AsyncLocationStream: NSObject, CLLocationManagerDelegate {
    lazy var stream: AsyncStream<CLLocation> = {
        AsyncStream { (continuation: AsyncStream<CLLocation>.Continuation) -> Void in
            self.continuation = continuation
        }
    }()
    var continuation: AsyncStream<CLLocation>.Continuation?

    func locationManager(_ manager: CLLocationManager, didUpdateLocations locations: [CLLocation]) {

        for location in locations {
            continuation?.yield(location)
        }
    }
}

위의 예제는 사용자 위치의 비동기 스트림을 만들기 위한 매우 기본적인 시작점입니다. 위치 관찰을 취소 및 시작하거나 위치 권한을 요청하는 등 완전히 고려하지 않은 몇 가지 사항이 있습니다.

하지만 기본적으로 이 예제는 비동기 스트림을 실험하기 위한 좋은 출발점이 될 수 있습니다.

이 접근 방식은 비동기 스트림의 소비자가 다음 값을 스트림으로 보내기 전에 이전 값을 완전히 소비할 때까지 기다리지 않습니다. 대신 전송하는 모든 값은 기본적으로 비동기 스트림에서 버퍼링되며, 이는 사용자가 원하는 값일 수도 있고 아닐 수도 있습니다.

 

즉, 실제로 소비자가 값을 처리하는 속도보다 더 빠른 속도로 값을 보낼 경우, 버퍼가 가득 차게 되며 소비하는 for 루프에 값이 지연되어 전달됩니다. 이는 정확히 필요한 것일 수도 있지만 전송하는 값이 다소 시간에 민감하고 일시적인 경우, 소비하는 for 루프가 값을 받을 준비가 되지 않았을 때 값을 버리는 것이 합리적일 수 있습니다.

 

예를 들어 오래된 데이터를 처리하지 않기 위해 1개 이상의 위치를 보유하지 않고 마지막으로 알려진 위치만 버퍼링하기로 결정할 수 있습니다. 비동기 스트림에 다음과 같은 버퍼링 정책을 설정하면 이 작업을 수행할 수 있습니다:

lazy var stream: AsyncStream<CLLocation> = {
    AsyncStream(bufferingPolicy: .bufferingNewest(1)) { (continuation: AsyncStream<CLLocation>.Continuation) -> Void in
        self.continuation = continuation
    }
}()

이 코드는 버퍼링 정책으로 .bufferingNewest(1)를 AsyncStream에 전달합니다. 즉, for 루프가 항목을 충분히 빠르게 처리하지 못하는 경우 단일 값만 버퍼링하고, 최신 위치만 유지하기 위해 이전 값은 버립니다.

스트림이 자연스럽게 종료되면 continuation에 finish()를 호출하여 값 스트림을 종료할 수 있습니다.

 

스트림이 에러와 함께 실패할 수 있는 경우 AsyncStream 대신에 AsyncThrowingStream을 생성할 수도 있습니다. 주요 차이점은 에러를 throw하는 스트림의 소비자는 try await 대신에 단순히 await를 사용하여 새 값을 기다려야 한다는 것입니다. 스트림이 에러를 throw하게 만들려면 continuation에서 finish(throwing:)를 호출하거나 Result 객체를 사용하여 yield(with:)를 호출할 수 있습니다.

비동기 스트림을 구축하는 기본 사항은 특별히 복잡하지는 않지만, 우리가 생성하는 객체의 수명 주기를 어떻게 관리할지 신중하게 생각해야 합니다. 특히 기존 델리게이트 기반 코드를 브리징할 때 continuation이 스트림보다 오래 유지되지 않도록 주의해야 합니다.

스트림 수명주기 관리하기

비동기 스트림이 종료되는 방법에는 기본적으로 2가지 방법이 있습니다.

 

첫째, 더 이상 값이 생성되지 않아 스트림이 자연스럽게 값 생성을 종료할 수 있습니다. continuation에 finish를 호출하고 동시에 수행해야 할 모든 정리 작업을 할 수 있습니다. 예를 들어 의도치 않게 더 이상 continuation을 사용하지 않도록 continuation을 nil로 설정할 수 있습니다.

 

또는, 비동기 스트림을 실행하는 데 사용되는 작업이 취소되어 스트림이 종료될 수 있습니다. 다음을 고려하세요.

let locations = AsyncLocationStream()

let task = Task {
    for await location in locations.stream {
        print(location)
    }
}

task.cancel()

위와 같은 상황이 발생하면 이미 완료된 스트림에 값을 보내지 않도록 continuation에 yield를 호출하지 않아야 합니다. 이러한 종료를 감지하고 대응하기 위해 continuation에 onTermination 핸들러를 설정할 수 있습니다:

self.continuation?.onTermination = { result in
    print(result)
    self.continuation = nil
}

가능하면 이 핸들러를 비동기 스트림을 처음 생성할 때 즉시 설정하는 것이 좋습니다.

또한 스트림이 취소되거나 다른 범위를 벗어난 경우에도 루프를 탈출할 수 있으므로 작업이 종료됩니다. 일반적으로 이 경우 스트림이 종료되는 것이 아니므로, 루프를 탈출하여 스트림을 종료하려면 이 점을 직접 고려해야 합니다.

 

글의 작성자는 개인적으로 스트림을 가장 쉽게 정리할 수 있는 방법은 스트림을 생성하는 객체에 비동기 루프에서 벗어나는 대신 스트림을 취소하는 메서드를 사용하는 것이라고 생각합니다. 이렇게 하면 정리를 수행하고 아무도 수신하지 않는데도 값을 전송하는 스트림을 만들지 않을 수 있습니다.

또한 앞서 설명한 패턴은 한 명의 소비자가 위치 스트림 객체를 사용하는 경우에만 작동한다는 점을 명심해야 합니다. 기본적으로 비동기 시퀀스에는 반복을 여러 루프와 공유하는 기능이 없기 때문에 Swift Concurrency에서는 단일 스트림을 반복하는 루프에 대해 여러 개의 루프를 가질 수 없습니다.

In Summary

이 포스팅에서는 비동기 스트림과 자신만의 비동기 시퀀스를 생성하는 방법에 대해 많은 것을 배웠습니다. 먼저, 비동기 스트림을 빌드하는 언폴딩 접근 방식을 살펴보고 이 접근 방식은 비교적 간단하지만 기존 델리게이트 또는 콜백 기반 API를 연결해야 하는 사람들에게는 그다지 유용하지 않을 수 있다는 것을 배웠습니다.

언폴딩에 대해 잠시 살펴본 후 비동기 스트림에 대한 빌드 클로저에 대해 살펴보았습니다. 이 접근 방식은 필요할 때 호출하여 값을 생성할 수 있는 연속 객체를 활용한다는 것을 배웠습니다.

CLLocationManager를 비동기 대기 상태로 연결하는 객체의 아주 기초적인 예제를 보셨고, 이미 완료된 스트림으로 값을 보내지 않도록 연속을 올바르게 관리하는 방법에 대해 배웠습니다.

 

 

원문

https://www.donnywals.com/understanding-swift-concurrencys-asyncstream/