[Swift Concurrency] AsyncStream, AsyncThrowingStream, Continuation이란?

1. Stream이란?

스트림(Stream)이란 데이터의 흐름을 의미한다.

이러한 데이터 흐름은 단방향이며, 한번에 일부 데이터를 처리한다.

즉, 데이터를 한번에 모두 처리하지 않고, 데이터가 하나씩 또는 여러 개씩 일정한 간격으로 생성되어 이를 연속적으로 처리하는 방식이다.

스트림은 많은 데이터를 처리하는 경우 유용하며, 여러 개의 데이터를 처리할 때 코드의 가독성과 유지보수성이 좋아진다.

스트림은 실시간으로 데이터를 처리하고, 데이터가 발생하는 즉시 처리할 수 있어 대규모 데이터 처리에 유용하다.

 

 

2. AsyncStream이란?

An asynchronous sequence generated from a closure that calls a continuation to produce new elements.

 

struct AsyncStream<Element>

 

순서가 있고,비동기적으로 생성된 요소들의 Sequence이다. 클로저 내부에서 하고 싶은 일을 정의할 수 있다.

기존의 Sequence를 생성하려면 IteratorProtocol(반복자)를 직접 구현해서 만들어 줘야 했지만,

[IteratorProtocol 직접 구현 예시]

struct TestIterator: IteratorProtocol {
    typealias Element = Int
    var count: Int
    
    mutating func next() -> Int? {
        if count == 0 { return nil }
        defer { count -= 1 }
        return count
    }
}

struct Test: Sequence {
    func makeIterator() -> TestIterator {
        return TestIterator(count: 5)
    }
}

let test = Test()
for value in test {
    print(value)
}

// Sequence 채택을 위해서는 필수 메소드인 next()와 makeIterator()를 구현해야 한다.

 

AsyncStream은 AsyncSequence를 준수하여 Iterator를 직접 구현할 필요 없이 간편하게 사용할 수 있다.

 

 

 

AsyncSequence이란?

기존의 Sequence는 한 번에 하나씩, 단계별(step)로 진행할 수 있는 값 목록(list of values)입니다.

여기에 비동기성을 추가한 것이 바로 AsyncSequence이다.

async iterator를 구현하는 것 없이 async sequence를 만들어낸다.

 

 

 

 

돌아와서 AsyncStream의 클로저는 AsyncStream.Continuation을 받는 클로저이다.

continuation’s yield(_:) method를 통해 stream에 element를 만들어낼 수 있다.

더 이상 만들어낼 element가 없다면 continuation’s finish()를 호출한다.

 

[예시]

class QuakeMonitor {
    var quakeHandler: ((Quake) -> Void)?


    func startMonitoring() {…}
    func stopMonitoring() {…}
}

extension QuakeMonitor {


    static var quakes: AsyncStream<Quake> {
        AsyncStream { continuation in
            let monitor = QuakeMonitor()
            monitor.quakeHandler = { quake in
                continuation.yield(quake)
            }
            continuation.onTermination = { @Sendable _ in
                 monitor.stopMonitoring()
            }
            monitor.startMonitoring()
        }
    }
}

for await quake in QuakeMonitor.quakes {
    print("Quake: \(quake.date)")
}
print("Stream finished.")

 

[예시2]

let stream = AsyncStream<Int> { continuation in
    Task.detached {
        for num in 1 ... 5 {
            try await Task.sleep(nanoseconds: 1 * 1_000_000_000)
            continuation.yield(num)
        }
        continuation.finish()
    }
}

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

 

 

3. AsyncThrowingStream이란?

An asynchronous sequence generated from an error-throwing closure that calls a continuation to produce new elements.

 

struct AsyncThrowingStream<Element, Failure> where Failure : Error

 

AsyncStream과 구조는 거의 유사하다.

다만, AsyncThrowingStream은 에러를 던질(throw) 수 있어,  Stream의 완료 값으로 성공 또는 실패를 가지게 된다.

에러가 발생하면 Sequence는 종료된다.

에러 타입도 함께 지정해주어야 한다.

finish() 대신 finish(throwing:)을 사용하여 에러를 던지며 stream을 종료할 수 있다.

 

 

 

[예시]

enum NumError: Error {
    case overThree
}

let stream = AsyncThrowingStream<Int, Error> { continuation in
    Task.detached {
        for num in 1 ... 5 {
            try await Task.sleep(nanoseconds: 1 * 1_000_000_000)
            if num > 3 {
                continuation.finish(throwing: NumError.overThree)
            } else {
                continuation.yield(num)
            }
        }
        continuation.finish()
    }
}

Task {
    do {
        for try await num in stream {
            print(num)
        }
    } catch {
      print(error)
    }
}

 

 

4. Continuation이란?

A mechanism to interface between synchronous code and an asynchronous stream.

 

struct Continuation

 

yield 메소드를 통해 stream에 element를 제공하거나 finish 메소드를 사용하여 stream을 끝내기 위해 continuation을 사용할 수 있다.

 

 

 

 

+

iOS15에서 AsyncThrowingStream 안에서 continuation을 사용하지 않고 사용하게 되면 무한 루프를 도는 이슈(CPU와 메모리 사용량 폭증)이 있다. iOS 16, 17에서는 이가 해소된 듯 한데, iOS15에서는 이를 조심히 사용해야 할 듯 싶다. async block을 그냥 사용하지 말고 continuation 사용을 보장할 것.

 

 

참고

- https://developer.apple.com/documentation/swift/asyncstream

- https://ios-daniel-yang.tistory.com/45

- https://developer.apple.com/documentation/swift/asyncstream/continuation