Observable → Async / Await 으로 변환하여 API 처리하기: AsyncThrowingStream, withCheckedThrowingContinuation

이번 글에서는 Rx Observable 기반의 기존 API 호출 로직을 Swift의 async/await 구조로 효율적으로 변환하는 방법을 다룹니다.

특히, 단일 호출에는 withCheckedThrowingContinuation을, 지속적인 응답이나 웹소켓에는 AsyncThrowingStream 또는 AsyncStream을 활용하는 방법을 자세히 알아보겠습니다.

--------------------------------------------------------------------------------

기존 Observable 기반 삭제 기능의 문제점

기존 앱의 삭제 기능은 Observable을 사용하여 구현되어 있었습니다.

private func deleteImage(accessKey: String) -> Observable<Response> { return networking.request(.deleteImage(accessKey: accessKey)).asObservable() }

문제는 해당 API를 요청하는 곳에서는 async/await 구조를 사용하고 있다는 점입니다.

그렇다고 해서 Base 네트워크 구현체까지 수정할 수는 없는 상황이었습니다.

따라서 응답을 처리하는 부분에서 Rx (Observable)를 async/await으로 변환해주는 로직이 필요했습니다.

 

 

단일 API 호출 처리: withCheckedThrowingContinuation

웹소켓처럼 지속적인 응답을 받는 경우가 아니라 API 호출처럼 1회성 호출이라면, withCheckedThrowingContinuation을 사용하는 것이 더 적합합니다.

func deleteImage(accessKey: String) async throws -> Response {
    return try await withCheckedThrowingContinuation { continuation in
        let disposable = deleteImage(accessKey: accessKey)
            .subscribe(
                onSuccess: { response in
                    continuation.resume(returning: response)
                },
                onFailure: { error in
                    continuation.resume(throwing: error)
                }
            )
        // Note: In real-world scenarios, consider disposing of 'disposable'
        // when the continuation is resumed or when the task is cancelled.
        // For brevity, it's omitted here but crucial for memory management.
    }
}

사용 예시:

_ = try await Service().deleteImage(accessKey: key)

 

 

지속적인 응답 처리: AsyncStream & AsyncThrowingStream

웹소켓이나 지속적인 응답을 필요로 할 때 적합한 방법은 AsyncStream 또는 AsyncThrowingStream을 사용하는 것입니다.

 

AsyncThrowingStream: 에러를 던져야 할 때

await에서 에러를 받아서 사용해야 하는 경우에는 AsyncStream 대신 AsyncThrowingStream을 사용해야 합니다.

AsyncStream은 에러를 던져줄 수 없기 때문입니다.

 

func deleteImage(accessKey: String) -> AsyncThrowingStream<Response, Error> {
    AsyncThrowingStream { continuation in
        let disposable = deleteImage(accessKey: accessKey)
            .subscribe(
                onNext: { progressResponse in
                    continuation.yield(progressResponse) // 진행 상황 전달
                },
                onError: { error in
                    continuation.finish(throwing: error) // 실패
                },
                onCompleted: {
                    continuation.finish() // 성공
                }
            )
        continuation.onTermination = { _ in
            disposable.dispose()
        }
    }
}

 

AsyncStream: 에러 처리 없이 성공만 필요할 때

만약 에러 처리가 필요 없고 성공만 처리하면 된다면, AsyncStream을 활용할 수 있습니다. 필요한 경우에만 선택하여 사용하세요.

func deleteImage(accessKey: String) -> AsyncStream<Response> {
    AsyncStream { continuation in
        let disposable = deleteImage(accessKey: accessKey)
            .subscribe(
                onNext: { progressResponse in
                    continuation.yield(progressResponse)
                },
                onCompleted: {
                    continuation.finish()
                }
            )
        continuation.onTermination = { _ in
            disposable.dispose()
        }
    }
}

사용 예시:

AsyncStream을 사용하는 곳에서는 for await 구문을 사용해야 합니다.

for await _ in Service().deleteImage(accessKey: key) { }

 

AsyncThrowingStream을 활용한 Progress 및 응답 처리

 

만약 진행 상황(progress)과 성공, 실패 모두를 처리해야 한다면 AsyncThrowingStream을 아래와 같이 구현하고 사용할 수 있습니다.

let stream = KageService().uploadDocument(path: path, fileName: fileName) // AsyncThrowingStream 리턴 받음
for try await progressResponse in stream { // 스트림에서 response 꺼내옴
    guard let response = progressResponse.response else { continue } // progressResponse.response 있으면 끝난거
    let document = try response.map(KageDocumentResponse.self)
    return document // 응답값을 mapping 후 리턴
}
throw CommunityPostError.uploadFail // 리턴 안했으면 에러 던짐

여기서 response는 진행 중에는 nil이 되며, 이 경우 continue를 통해 다음 반복문을 실행시킬 수 있습니다.

만약 response가 nil이 아니라면, 통신이 끝났고 응답 값을 받은 것이므로 응답 값을 리턴해 줄 수 있습니다.

만약 리턴을 하지 못했는데 반복문이 끝났다면 에러가 발생한 것이므로 throw CommunityPostError.uploadFail와 같이 명시적으로 에러를 던져주면 됩니다. 아니면 catch 문을 써서 error를 받아 처리할 수도 있습니다.