AsyncStream

Overview 概要

AsyncStream conforms to AsyncSequence, providing a convenient way to create an asynchronous sequence without manually implementing an asynchronous iterator. In particular, an asynchronous stream is well-suited to adapt callback- or delegation-based APIs to participate with async-await. AsyncStreamはAsyncSequenceに準拠します、そして手動で非同期イテレータを実装することなく非同期シーケンスを作成するある簡便な方法を提供します。とりわけ、非同期ストリームは、コールバックや委任に基づくAPIをasync-awaitと関与するよう適応させるのに最適です。

You initialize an AsyncStream with a closure that receives an AsyncStream.Continuation. Produce elements in this closure, then provide them to the stream by calling the continuation’s yield(_:) method. When there are no further elements to produce, call the continuation’s finish() method. This causes the sequence iterator to produce a nil, which terminates the sequence. The continuation conforms to Sendable, which permits calling it from concurrent contexts external to the iteration of the AsyncStream. あなたはAsyncStreamを、AsyncStream.Continuationを受け取るあるクロージャで初期化します。要素それらをこのクロージャにおいて生み出して、それからそれらをストリームへと、継続のもつyield(_:)メソッドを呼び出すことによって提供してください。それ以上は要素を生み出さない場合は、継続のもつfinish()メソッドを呼び出してください。これはシーケンスイテレータにnilを生み出させ、それはそのシーケンスを終端します。継続はSendableに準拠します、それはAsyncStreamの反復処理に関係がない並行文脈からそれを呼び出すことを許可します。

An arbitrary source of elements can produce elements faster than they are consumed by a caller iterating over them. Because of this, AsyncStream defines a buffering behavior, allowing the stream to buffer a specific number of oldest or newest elements. By default, the buffer limit is Int.max, which means the value is unbounded. 要素いくつかからなるある随意のソースは、要素それらを、それらのすべてにわたって反復している呼び出し側によってそれらが消費されるよりも速く生み出すことがありえます。このことから、AsyncStreamはあるバッファ挙動を定義して、ストリームに特定の数の古いまたは新しい要素をバッファすることを許可します。省略時では、バッファ限度はInt.maxです、それはその値が無制限であることを意味します。

Adapting Existing Code to Use Streams 既存のコードを適応させてストリームを使うようにする

To adapt existing callback code to use async-await, use the callbacks to provide values to the stream, by using the continuation’s yield(_:) method. 既存のコールバックコードをasync-awaitを使うように適応させるには、コールバックそれらを使って値それらをストリームへと、継続のもつyield(_:)メソッドを使うことによって提供してください。

Consider a hypothetical QuakeMonitor type that provides callers with Quake instances every time it detects an earthquake. To receive callbacks, callers set a custom closure as the value of the monitor’s quakeHandler property, which the monitor calls back as necessary. ある仮定的なQuakeMonitor型を考えてください、それは呼び出し側それらにQuakeインスタンスそれらを、それがある地震を検出するたびごとに提供します。コールバックそれらを受け取るには、呼び出し側それらはあるあつらえのクロージャをモニタのもつquakeHandlerプロパティの値として設定します、それはモニタが必要に応じて折り返し呼び出しするものです。

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


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

To adapt this to use async-await, extend the QuakeMonitor to add a quakes property, of type AsyncStream<Quake>. In the getter for this property, return an AsyncStream, whose build closure – called at runtime to create the stream – uses the continuation to perform the following steps: これをasync-awaitを使うように適応させるには、QuakeMonitorを拡張してquakesプロパティ、型AsyncStream<Quake>の、を加えてください。このプロパティに対するゲッターにおいて、あるAsyncStreamを返してください、それのbuildクロージャ – 実行時に呼び出されてストリームを作成するもの – は継続を使って以下の段階を追います：

1. Creates a QuakeMonitor instance. あるQuakeMonitorインスタンスを作成します。

2. Sets the monitor’s quakeHandler property to a closure that receives each Quake instance and forwards it to the stream by calling the continuation’s yield(_:) method. モニタのもつquakeHandlerプロパティをあるクロージャへと設定します、それは各Quakeインスタンスを受け取って、そしてそれをストリームへと、継続のもつyield(_:)メソッドを呼び出すことによって転送します。

3. Sets the continuation’s onTermination property to a closure that calls stopMonitoring() on the monitor. 継続のもつonTerminationプロパティをあるクロージャへと設定します、それはstopMonitoring()をモニタ上で呼び出します。

4. Calls startMonitoring on the QuakeMonitor. startMonitoringをQuakeMonitor上で呼び出します。

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

   }

Because the stream is an AsyncSequence, the call point can use the for-await-in syntax to process each Quake instance as the stream produces it: ストリームがAsyncSequenceであることから、呼び出しポイントはfor-await-in構文を使用して各Quakeインスタンスをストリームがそれを生み出すにつれて処理できます：

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