前言

AsyncThrowingStream 和 AsyncStream 是 Swift 5.5 中由 SE-314[1] 引入的并發(fā)框架的一部分。異步流允許你替換基于閉包或 Combine 發(fā)布器的現(xiàn)有代碼。

在深入研究圍繞拋出流的細節(jié)之前，如果你還沒有閱讀我的文章，我建議你先閱讀我的文章，內容包括async-await。本文解釋的大部分代碼將使用那里解釋的API。

什么是 AsyncThrowingStream？

你可以把 AsyncThrowingStream 看作是一個有可能導致拋出錯誤的元素流。他的值隨著時間的推移而傳遞，流可以通過一個結束事件來關閉。一旦發(fā)生錯誤，結束事件既可以是成功，也可以是失敗。

什么是 AsyncStream?

AsyncStream 類似于拋出的變體，但絕不會導致拋出錯誤。一個非拋出型的異步流會根據(jù)明確的完成調用或流的取消而完成。

注意： 在這篇文章中，我們將解釋如何使用AsyncThrowingStream。除了發(fā)生錯誤處理的部分，代碼示例與AsyncStream類似。

AsyncThrowingStream

如何使用 AsyncThrowingStream

AsyncThrowingStream 可以很好地替代現(xiàn)有的基于閉包的代碼，如進度和完成處理程序。為了更好地理解我的意思，我將向你介紹我們在 WeTransfer 應用程序中遇到的一個場景。

在我們的應用程序中，我們有一個基于閉包的現(xiàn)有類，叫做 FileDownloader：

struct FileDownloader {
    enum Status {
        case downloading(Float)
        case finished(Data)
    }

    func download(_ url: URL, progressHandler: (Float) -> Void, completion: (Result<Data, Error>) -> Void) throws {
        // .. Download implementation
    }
}

文件下載器接受一個URL，報告進度情況，并完成一個包含下載數(shù)據(jù)的結果或在失敗時顯示一個錯誤。

文件下載器在文件下載過程中報告一個數(shù)值流。在這種情況下，它報告的是一個狀態(tài)值流，以報告正在運行的下載的當前狀態(tài)。FileDownloader 是一個完美的例子，你可以重寫一段代碼來使用 AsyncThrowingStream。然而，重寫需要你在實現(xiàn)層面上也重寫你的代碼，所以讓我們定義一個重載方法來代替：

extension FileDownloader {
    func download(_ url: URL) -> AsyncThrowingStream<Status, Error> {
        return AsyncThrowingStream { continuation in
            do {
                try self.download(url, progressHandler: { progress in
                    continuation.yield(.downloading(progress))
                }, completion: { result in
                    switch result {
                    case .success(let data):
                        continuation.yield(.finished(data))
                        continuation.finish()
                    case .failure(let error):
                        continuation.finish(throwing: error)
                    }
                })
            } catch {
                continuation.finish(throwing: error)
            }
        }
    }
}

正如你所看到的，我們把下載方法包裹在一個 AsyncThrowingStream 里面。我們將流的值 Status 的類型描述為一個通用的類型，允許我們用狀態(tài)更新來延續(xù)流。

只要有錯誤發(fā)生，我們就會通過拋出一個錯誤來完成流。在完成處理程序的情況下，我們要么通過拋出一個錯誤來完成，要么用一個不拋出的完成回調來跟進數(shù)據(jù)的產生。

switch result {
case .success(let data):
    continuation.yield(.finished(data))
    continuation.finish()
case .failure(let error):
    continuation.finish(throwing: error)
}

在收到最后的狀態(tài)更新后，不要忘記 finish() 回調，這一點至關重要。否則，我們將保持流的存活，而實現(xiàn)層面的代碼將永遠不會繼續(xù)。

我們可以通過使用另一個 yield 方法來重寫上述代碼，接受一個 Result 枚舉作為參數(shù)：

continuation.yield(with: result.map { .finished($0) })
continuation.finish()

重寫后的代碼簡化了我們的代碼，并去掉了 switch-case 代碼。我們必須映射我們的 Reslut 枚舉以匹配預期的 Status 值。如果我們產生一個失敗的結果，我們的流將在拋出包含的錯誤后結束。

AsyncThrowingStream 迭代

一旦你配置好你的異步拋出流，你就可以開始在數(shù)值流上進行迭代。在我們的 FileDownloader 例子中，它將看起來如下所示：

do {
    for try await status in download(url) {
        switch status {
        case .downloading(let progress):
            print("Downloading progress: \(progress)")
        case .finished(let data):
            print("Downloading completed with data: \(data)")
        }
    }
    print("Download finished and stream closed")
} catch {
    print("Download failed with \(error)")
}

我們處理任何狀態(tài)的更新，并且我們可以使用 catch 閉包來處理任何發(fā)生的錯誤。你可以使用基于 AsyncSequence 接口的 for ... in 循環(huán)進行迭代，這對 AsyncStream 來說是一樣的。

如果你遇到了類似的編譯錯誤:

‘async’ in a function that does not support concurrency

上述代碼示例中的打印語句有助于你理解 AsyncThrowingStream 的生命周期。你可以替換打印語句來處理進度更新和處理數(shù)據(jù)，為你的用戶實現(xiàn)可視化。

調試 AsyncStream

如果一個流不能報告數(shù)值，我們可以通過放置斷點來調試流產生的回調。雖然也可能是上面的 “Download finished and stream closed” 的打印語句不會調用，這意味著你在實現(xiàn)層的代碼永遠不會繼續(xù)。后者可能是一個未完成的流的結果。

為了驗證，我們可以利用 onTermination 回調：

func download(_ url: URL) -> AsyncThrowingStream<Status, Error> {
    return AsyncThrowingStream { continuation in

        ///  配置一個終止回調，以了解你的流的生命周期。
        continuation.onTermination = { @Sendable status in
            print("Stream terminated with status \(status)")
        }

        // ..
    }
}

回調在流終止時被調用，它將告訴你你的流是否還活著。

如果出現(xiàn)了錯誤，輸出結果可能如下:

Stream terminated with status finished(Optional(FileDownloader.FileDownloadingError.example))

上述輸出只有在使用 AsyncThrowingStream 時才能實現(xiàn)。如果是一個普通的 AsyncStream，完成的輸出看起來如下：

Stream terminated with status finished

而取消的結果對這兩種類型的流來說都是這樣的:

Stream terminated with status cancelled

你也可以在流結束后使用這個終止回調進行任何清理。例如，刪除任何觀察者或在文件下載后清理磁盤空間。

取消一個 AsyncStream

一個 AsyncStream 或 AsyncThrowingStream 可以由于一個封閉的任務被取消而取消。一個例子可以如下：

let task = Task.detached {
    do {
        for try await status in download(url) {
            switch status {
            case .downloading(let progress):
                print("Downloading progress: \(progress)")
            case .finished(let data):
                print("Downloading completed with data: \(data)")
            }
        }
    } catch {
        print("Download failed with \(error)")
    }
}
task.cancel()

一個流在超出范圍或包圍的任務取消時就會取消。如前所述，取消將相應地觸發(fā) onTermination 回調。

結論

AsyncThrowingStream 或 AsyncStream 是重寫基于閉包的現(xiàn)有代碼到支持 async-awai t的替代品的好方法。你可以提供一個連續(xù)的值流，并在成功或失敗時完成一個流。你可以使用基于 AsyncSequence APIs 的 for 循環(huán)在實現(xiàn)層面上迭代值。

參考資料

[1]SE-314: ??https://github.com/apple/swift-evolution/blob/main/proposals/0314-async-stream.md。??