AsyncStream에 대해 알아보자

스터디2025.11.10

Swift

Stream이란?

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

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

즉, 데이터를 한 번에 몽땅 처리하는 것이 아닌, 하나씩 또는 여러 개씩 일정한 간격으로 생성되어 이를 연속적으로 처리하는 방식을 스트림이라고 칭한다.

스트림은 많은 데이터를 처리하는 경우 용이하다.

또한, 여러 개의 데이터를 처리할 때 코드의 가독성과 유지보수성이 좋아진다.

→ 실시간으로 데이터를 처리하고, 데이터가 발생하는 즉시 처리할 수 있다!

→ 이러한 특성 때문에 대규모 데이터 처리에 매우 유용하다.

AsyncStream이란?

: 순서가 있고 비동기적으로 생성된 요소들의 Sequence 이다.

기존 Sequence 를 생성하기 위해선 반복자(IteratorProtocol)를 직접 정의해야했다.

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

⭐ 제일 큰 이점이 AsyncStream 은 클로저 형태로 사용된다!

사용법

AsyncStream의 생성자를 확인해보자.

init(
    _ elementType: Element.Type = Element.self,
    bufferingPolicy limit: AsyncStream<Element>.Continuation.BufferingPolicy = .unbounded,
    _ build: (AsyncStream<Element>.Continuation) -> Void
)

AsyncStream.init의 파라미터를 확인해보자.

elementType

: AsyncStream에 사용될 타입을 정의한다.

bufferingPolicy

: AsyncStream의 버퍼링 동작을 정의한다. (버퍼는 임시 데이터 저장 장소이다.)

: 기본값은 .unbounded 로 무제한 버퍼링을 진행한다.

bufferingPolicy의 유형은 아래와 같이 3가지가 존재한다.

  ```swift
  case unbounded             // 무제한으로 요소를 추가시킬 수 있다.
  case bufferingOldest(Int)  // 버퍼가 가득차면 오래된 N개의 요소를 제외하고 나머지를 버린다.
  case bufferingNewest(Int)  // 버퍼가 가득차면 최신 요소 N개를 제외하고 나머지를 버린다.
  ```

build

: 클로저에서 continuation을 통해서 생산(yield) 이나 종료(finish) , 종료 처리(onTermination) 등을 정의한다.

이때, bufferingPolicy의 설정을 다시 확인해보자.

기본값으론 .unbounded 로 설정하여 버퍼에 이벤트를 무한정으로 적재할 수 있다.

한 번 bufferingOldest를 사용하여 버퍼의 크기를 3으로 정의해보자.

let stream = AsyncStream<Int>(Int.self,
                              bufferingPolicy: .bufferingOldest(3)) { continuation in
    for num in 1 ... 6 {
            continuation.yield(num)
        }
        continuation.finish()
}

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

// 1
// 2
// 3

// 만약 bufferingPolicy를 bufferingNewest로 처리를 하면 아래와 같이 출력될 것이다.
// 4
// 5
// 6

위와 같이 버퍼가 가득찼을 때 버퍼의 정책을 생성자에서 따로 설정해둔다면 데이터들을 의도한 대로 처리할 수 있다.

다만, 일반적인 AsyncStream 은 스트림의 성공과 실패를 가질 수 없다.

그 말은 따로 에러를 던지(throw)거나 잡을(catch) 수 없다.

이러한 상황에 사용할 수 있는 부분이 AsyncThrowingStream 이다.

`makeStream(of:bufferingPolicy:)`

developer.apple.com/documentation/swift/asyncstream/makestream(of:bufferingpolicy:

이때, AsyncStream과 Continuation을 튜플로 한 번에 생성할 수 있는 편의 메서드가 존재한다.

static func makeStream(
    of elementType: Element.Type = Element.self,
    bufferingPolicy: AsyncStream.Continuation.BufferingPolicy = .unbounded
) -> (stream: AsyncStream, continuation: AsyncStream.Continuation)

기존 방식은 아래와 같이 클로저 안에서 continuation을 캡쳐해야하지만,

makeStream을 사용하면 stream 과 continuation 을 분리해서 관리할 수 있다.


// 기존 방식 - continuation을 클로저 안에서만 사용 가능
let stream = AsyncStream { continuation in
    // 여기서만 continuation 사용 가능
    continuation.yield(1)
}

// makeStream 방식 - continuation을 외부에서 자유롭게 사용 가능
let (stream, continuation) = AsyncStream.makeStream()

// 어디서든 continuation 사용 가능
continuation.yield(1)
continuation.yield(2)
continuation.finish()

특히 Delegate 패턴이나 콜백 기반 API를 브릿징할 때 유용하게 사용할 수 있다.

class LocationManager {
    let (stream, continuation) = AsyncStream.makeStream()
    
    func locationManager(_ manager: CLLocationManager, didUpdateLocations locations: [CLLocation]) {
        // delegate에서 바로 continuation 사용
        for location in locations {
            continuation.yield(location)
        }
    }
}

즉, 생성할 때도 편하고 continuation을 프로퍼티로 저장해두고 필요할 때마다 yield를 호출시킬 수 있어 깔끔하다.

AsyncThrowingStream

AsyncThrowingStream은 Error Throw가 가능하다.

따라서 스트림의 완료 값으로 성공 또는 실패를 가질 수 있다.

→ 이때, 에러가 발생하면 Sequence는 종료된다.

AsyncThrowingStream의 생성부는 아래와 같다.

struct AsyncThrowingStream<Element, Failure> where Failure : Error

아래 코드는 https://ios-daniel-yang.tistory.com/entry/iOSSwift-AsyncStream-AsyncThrowingStream#article-3--asyncthrowingstream-%EB%9E%80 에서 예시로 작성하신 부분을 그대로 끌고와봤다.

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)
    }
}

//1초 마다

// 1
// 2
// 3
// overThree

1초마다 숫자를 생성시키고, num이 3보다 커지면 에러를 던지도록 코드가 작성되어있다

커스텀 에러를 처리하기 위해선 continiation의 finish()가 아닌 finish(throwing:) 을 사용하면 쉽게 처리할 수 있다.

Continuation이란?

AsyncStream이든 AsyncThrowingStream이든 공통적으로 yield(예일~) 이나 finish 같은 메서드들이 존재한다.

이는 AsyncStream.Continuation에 포함돼 있는 메서드이다.

AsyncStream.Continuation | Apple Developer Documentation

한 번 하나하나 중요한 요소를 확인해보자.

1. `yield`

func yield(_ value: Element) -> AsyncStream<Element>.Continuation.YieldResult

yield는 AsyncStream에 element를 제공한다.

만약, 코드가 실행되는 중 yield를 만나게 되면 다음 반복 작업을 기다린다.

2. `finish`

func finish()

Sequence의 이터레이션은 nil을 만날 시 종료된다.

→ Sequence와 AsyncSequence는 아래 Zed님의 블로그에 정말 잘 설명되어있다.

→ https://zeddios.tistory.com/1339

따라서 finish()는 nil을 반환시키며 이는 반복 작업을 종료한 뒤 다음 작업을 진행하도록 하는 의미이다.

3. `onTermination`

var onTermination: ((AsyncStream<Element>.Continuation.Termination) -> Void)? { get nonmutating set }

enum Termination {
    case finished  // Stream이 종료 될 때
    case cancelled // Stream이 취소 될 때
}

onTermination은 AsyncStream의 종료나 취소 시 호출되는 콜백이며 이를 통해 스트림의 활성 상태 여부를 확인할 수 있다.

즉, 아래와 같이 활용할 수 있을 것이다.

// 코드 참고 : https://ios-daniel-yang.tistory.com/entry/iOSSwift-AsyncStream-AsyncThrowingStream#article-3--asyncthrowingstream-%EB%9E%80
let stream = AsyncStream<Int> { continuation in
    continuation.onTermination = { status in
        switch status {
        case .finished: 
            print("Stream terminated with status : \(status)")  
        case .cancelled:                 
            print("Stream terminated with status : \(status)")  
        }
    }
    
    Task.detached {
        for num in 1...5 {
            try await Task.sleep(nanoseconds: 1 * 1_000_000_000)
            continuation.yield(num)
        }
        print("Progress finish before")
        continuation.finish()
        print("Progress finish after")
    }
}

let task = Task {
    for await num in stream {
        print(num)
    }
}

// 1
// 2
// 3
// 4
// 5
// Progress finish before
// Stream terminated with status : finished
// Progress finish after

만약, 도중에 AsyncStream의 작업을 취소하고자 하면 해당 Task에 cancel()을 걸어 onTermination의 .cancelled를 호출시킬 수 있다.

let task = Task {
   for await num in stream {
      print(num)
   }
}
sleep(3)
task.cancel()

// 1
// 2
// Stream terminated with status : cancelled <- onTermination의 .cancelled가 호출됩니다.

AsyncStream 소비하기

위에서 AsyncStream과 AsyncThrowingStream을 정의하고 생성하는 방법을 확인했다.

그렇다면 실제로 이벤트를 어떻게 소비 하는지 사용처 측면에서 확인해보자.

AsyncStream은 AsyncSequence 프로토콜을 준수하기 때문에 for await in 구문으로 간편하게 사용할 수 있다.

→ 자세한 내용은 아래 Zed님 블로그 링크에.. ~~(절대 힘들어서 그런게 아니다..ㅎㅎ)~~

→ https://zeddios.tistory.com/1339

가장 기본적인 소비 패턴

AsyncStream으로 이벤트를 소비하는 가장 기본적인 방법은 위에서 말한 것과 같이 for await in 루프를 사용하는 것이다.

let stream = AsyncStream<Int> { continuation in
   for num in 1...5 {
      continuation.yield(num)
   }
   continuation.finish()
}

Task {
   for await value in stream {
      print("받은 값 : \(value)")
   }
   print("스트림 종료")
}

// 받은 값: 1
// 받은 값: 2
// 받은 값: 3
// 받은 값: 4
// 받은 값: 5
// 스트림 종료

스트림에서 finish()가 호출되어 자동으로 for 루프가 종료되고 다음 코드로 진행된다.

AsyncThrowingStream에서 소비하기

에러를 던질 수 있는 AsyncThrowingStream 은 do-catch 구문과 함께 사용될 수 있다.

let stream = AsyncThrowingStream<String, Error> { continuation in
    continuation.yield("첫번째 데이터")
    continuation.yield("두번째 데이터")
    continuation.finish(throwing: NSError(domain: "TestError", code: -1))
}

Task {
    do {
        for try await value in stream {
            print("받은 값: \(value)")
        }
        print("정상 종료")
    } catch {
        print("에러 발생: \(error)")
    }
}

// 받은 값: 첫번째 데이터
// 받은 값: 두번째 데이터
// 에러 발생: Error Domain=TestError Code=-1 "(null)"

위와 같이 finish(throwing:) 을 만나게 되면 catch 블록이 실행되어 스트림이 종료된다. (자동 종료다.)

파일 다운로드 진행률 예시

실제 파일 다운로드나 업로드가 이뤄질 때 진행률 을 같은 상황을 추적할 때 아래와 같이 활용된다.

enum DownloadStatus {
    case downloading(progress: Double)
    case completed(data: Data)
}

func downloadFile(url: URL) -> AsyncThrowingStream<DownloadStatus, Error> {
    AsyncThrowingStream { continuation in
        Task {
            // 다운로드 진행 상황을 계속 yield
            for progress in stride(from: 0.0, through: 1.0, by: 0.1) {
                try await Task.sleep(nanoseconds: 100_000_000) // 0.1초 대기
                continuation.yield(.downloading(progress: progress))
            }
            
            // 완료 시 데이터 전달
            let data = Data("다운로드 완료!".utf8)
            continuation.yield(.completed(data: data))
            continuation.finish()
        }
    }
}

// 사용하는 쪽
Task {
    do {
        for try await status in downloadFile(url: URL(string: "https://example.com")!) {
            switch status {
            case .downloading(let progress):
                print("다운로드 진행률: \(Int(progress * 100))%")
            case .completed(let data):
                print("다운로드 완료! 데이터 크기: \(data.count)바이트")
            }
        }
    } catch {
        print("다운로드 실패: \(error)")
    }
}

사용하는 쪽의 코드를 보면 for await in 구문을 통해 간단하게 처리할 수 있다!

Task 취소와 연동

만약, 스트림을 취소해야한다?

그렇다면 스트림을 취소하는 Task 자체를 취소하면 스트림도 자동으로 취소된다!

let downloadTask = Task {
    do {
        for try await status in downloadFile(url: someURL) {
            // 다운로드 처리
        }
    } catch {
        print("에러: \(error)")
    }
}

// 3초 후 취소
Task {
    try await Task.sleep(nanoseconds: 3_000_000_000)
    downloadTask.cancel()
    print("다운로드 취소됨")
}

위와 같이 downloadTask.cancel()을 통해 Task를 취소하면 위에서 설명한 스트림의 onTermination 에서 .cancelled 가 호출되기 때문에 스트림이 자동으로 취소된다.