AsyncSequenceについて

値のシーケンスを反復することは、“for”ループを書くのと同じくらい簡単になりました。新しいAsyncSequenceプロトコルにより、通知をはじめ、サーバからのバイトストリームまで、あらゆるものを反復処理するための自然でシンプルな構文を実現する方法について確認します。また、既存のコードを応用して独自の非同期シーケンスを作成する方法も紹介します。このセッションを最大限活かしていただくためには、「Swiftのasync/awaitについて」を先にご確認いただくことをお勧めします。

リソース

関連ビデオ

WWDC22

WWDC21

ダウンロード

♪ (AsyncSequenceについて) こんにちは Philippeです今回はSwiftの新機能 AsyncSequenceをご紹介できてとても嬉しいです今日は非同期シーケンスとは何かやその根本にあるものについて説明しますその後非同期シーケンスのコードでの使い方や新しいAsynceSequence APIにも触れます最後には独自の非同期シーケンスの構築方法を説明しますでは始めましょう非常にシンプルなツールです　AsyncSequenceでできる素晴らしい新機能をいくつかまとめましたこのツールではエンドポイントへの URLから始めます最近のQuake 地震を一覧にしています何かをダウンロードするのは非同期のタスクで時間がかかることがありますでもこのケースでは全部のダウンロードが済むのを待つのではなく受け取れたものを示したいですそこで少し変えて新しいasync/await機能を使うことでエンドポイントから返る行を得ることにしましたフェッチしていないデータがコンマ区切りでフォーマットされているので各行がは完全なデータの行です行のAsyncSequenceは受け取った通りに各行を出しますつまりこの先大量のダウンロードがある可能性があるということですしかし受け取った時に表示することでとてもレスポンシブになりまた一番すごいのはこの新しい非同期コンテキストは通常のシーケンスでいつも使っているのと同じものを使えることです for-await-in 構文を使った iterate や map filter reduce などの機能またはこの例のように dropFirst 機能を使ってこうした値を操作することができますこれはどんな仕組みになっているのでしょうか？今日お話しすることの多くは async/await を基盤としたものですが主なポイントをいくつかまとめてみましょう Async機能ではコールバックなしで await キーワードを使って並列コードを書くことができます非同期関数の呼び出しは停止され値やエラーが生成されると再開されます一方 AsyncSequence では各要素は停止し再開は背景にあるイテレータが値を生成したりエラースロー時に行われます基本的には名前の通り通常の Sequence と似ていますがいくつか大きな違いがありますすなわち各要素が非同期で引き渡されますしかし非同期デリバリーなので失敗が起こる可能性も必ずあります非同期シーケンスの中にはエラーをスローするものもしないものもありますエラーをスローする関数と同様コンパイラは反復や構成を行う時にエラーが確実に処理されるようにします一般的に言うと非同期シーケンスとは時間の経過とともに値を生み出す方法を表したものですそのため AppSequence は０以上値でありイテレータからnilを返すことで完了を示しますシーケンスと同様ですエラー発生時が発生すると AsyncSequence は終了状態となりエラー発生後の次のイテレータのシーケンス呼び出しで nil を返しますでは定義の仕組みについて詳しく見ていきましょうまずは通常のイテレーションからです非常によくあるパターンですねこれは for-in ループですこのケースではシーケンスから quake を反復し大きさが一定レベルを超えたときに関数を呼び出しますコンパイラにはイテレーションの仕組みについて知識がありますが魔法をかけるわけではありませんコンパイルの段階では単純明快な変換が行われるだけですこれらの変換について調べてこれの非同期形式が何であるかを理解できるようにしましょうこれは以前のコードのビルドでコンパイラが行う内容をざっと示したものですまずイテレータ変数の作成から始め whileループを使ってnextが呼び出される際にイテレータが生成するquakeを全て取得します新しい async/await 機能を使うにはちょっとした微調整で済みます nextの関数を非同期関数に変えるのと同じように簡単ですこれで次のquakeを待つことによって反復を Swiftコンカレンシーにすることができますさて少し戻りループが AsyncSequence にある場合はどうなるか見ていきましょう先程述べた通り非同期シーケンスから各アイテムを待つ必要がありますこれは新しい for-await-in 構文に反映されています Sequence の使い方を知っていれば AsyncSequence の使い方もよく分かっているということです非同期シーケンスを活用するには幾つかの方法があります今触れたように新しい for-await-in 構文を使うか AsyncSequence がエラーをスローする場合は新しい for-try-await-in 構文を使うかですこれで非同期に生成された値を簡単に反復できクロージャをいじる必要もなく既に慣れている構文で反復を行うことができます break文やcontinue文でもですこれで非同期シーケンスの理論をよく理解できたと思います反復についてもう少し探ってみましょう非同期シーケンスのソースを与えられたら for-await-in 構文を使って各値を待つことができますこれはイテレータに生成された各アイテムを保留し終端に達するとループを完了します非同期シーケンスのループ中で反復を早めに終了するには break を使いますこれは通常のシーケンスと同じように機能しますここでは quake にロケーションデータがない時にブレークしていますまたはスキップしたい値がある場合は continue を使いますこのケースでは depth がある値より大きい場合これらをスキップし続行して次の quake を待ちますダウンロードからのこの次の反復は以前と同じように機能しますがこのケースではソースからエラーがスローされることがありますエラーをスローする関数と同じように反復中の AsyncSequence のスローは各要素の処理に try が必要になりますまたエラーをスローする関数と同じように try が無いとコンパイラが検知し Fix-it を出してミスの訂正を促しますこれはつまり非同期シーケンスがエラーを生成する可能性がある場合言語がそのエラーを throw するか catch することを強制するため常に安全であることを意味します次の反復は最初のループ反復の後連続的に実行されますコードをシーケンシャルに実行するのがいつも望ましいとは限りません進行中の他のことと同時に反復を実行することが有用な場合は反復をカプセル化する新しい非同期タスクを作成できますこれは使用中の非同期シーケンスが無制限に実行されると分かって入る時には便利でしょうさてシーケンスが無制限だったとしても発生可能性は非常に低いですしかし非同期動作の世界ではよく発生しており使う時にはよく検討する必要があります有難いことに Swiftのコンカレンシー機能によりこれはとても簡単かつ安全になっています反復を外部でキャンセルする可能性がある時には便利でもあるでしょうこうして2つの反復を並列して実行し後で反復を終了することができます無制限かもしれない反復作業をコンテナの寿命にスコープするタスクでは非常に簡単ですでは AsyncSequence API をご紹介しますこれは macOS Monterey・iOS 15・tvOS 15 watchOS 8で利用可能です沢山ありますがハイライトだけご紹介します非同期動作ではファイルからの読み込みが主なユースケースです FileHandleから非同期シーケンスで読み込むを行う新しいプロパティ AsyncBytes を FileHandle に追加しましたこれは非同期シーケンスのバイトを行に変換する AsyncSequenceの新しいExtensionと組み合わせて使用できますファイルの処理は非常に一般的であるため URL には bytes と lines の両方のアクセッサを用意しましたこれは最初の例の APIと同じものですファイルまたはネットワークのどちらからでもコンテンツから line の AsyncSequence を返す URL のコンビニエンスプロパティですこれで以前は非常に複雑だったタスクの多くが簡単で安全になるでしょうネットワークから取得する時応答や認証にもっとコントロールが必要になることがありますそのため URLSession に URL や URLRequest で指定されたバイトの AsyncSequence をフェッチする bytes 関数ができました詳しくは「URLSessionにおけるasync/awaitの使用」をご覧ください URLSession の新しい非同期機能についてより詳しく学習できますしかし AsyncSequence で意味を成すのはファイルやネットワークだけではありません新しい Notification API では通知も待機できるようになりましたまた反復以外から AsyncSequence の利用が可能になりましたこの例では CoreData の UUID 指定のレコードのリモートでの変更の最初の通知を待機しています firstWhere などのメソッドを AsyncSequence の通知と合わせて使うと以前は複雑なロジックを実装していたコードをコンパクトで読みやすくすることができ非常に優れた新しいデザインパターンが可能になりますまたこれだけでなく非同期シーケンスからの値を非同期で操作する新しいAPIも多く追加されましたこれはとても親しみやすいと思いますその一部は Sequence の機能と同じものだからです dropFirst や firstWhere など既にいくつか触れましたがこれ以外にもたくさんあります Sequence で使おうと思うもののほとんど全てに AsyncSequence で動かすための非同期のペアがありますあまりにも多いので「新しいAPIは確かに素晴らしくて構文もすっきりしているが自分の非同期シーケンスはどう作れば良いのか？」と思われるかもしれませんではやってみましょう AsyncSequence の実装方法はいくつかありますが既存のコードへの応用方法にフォーカスしたいと思います特に AsyncSequence とうまく合うデザインパターンがいくつかあり既存のものをこの新しいコンセプトに合わせる素晴らしい機能も用意していますそのうちいくつかはクロージャのように複数回呼び出しされますが良い感じに動作する delegate もあります応答を返す必要がなく発生する新しい値を伝えるだけならほとんどの場合 AsyncSequence 作成のトップ候補になれますこれらはよくあるデザインパターンで皆様のAppの中にももうあるかもしれませんこれはよくあるハンドラパターンの一例ですハンドラプロパティや Start と Stop メソッドを持つクラスです AsyncSequence の候補としては完璧に見えますね既存の使い方はこんな感じでしょうかモニターが作成され値を取得するハンドラが割り当てられてモニターが開始されるので Quakeをハンドラに送信できます後に生成中のイベントをキャンセルするためモニターが停止されるかもしれません同じインターフェイスを使って新しい AsyncSequence タイプに応用することができますほんの少しのコードだけで AsyncSequence を構築可能です AsyncStream を構築する時は要素タイプと構築クロージャを指定しますクロージャは値を複数回生成 / 終了したり中断処理を行う continuation を受け取りますこのケースではモニターを構築クロージャ内で作成していますまたQuakeを生成するハンドラを continuation に割り当てることも可能ですそしてonTerminationでキャンセルやクリーンアップを処理できますその後モニタリングを開始します先程使ったのと同じモニターコードは AsyncStream のコンストラクタ内で簡単にカプセル化可能ですこれで同じロジックを使うたびに複製する必要がなくなりますこれが AsyncStream の使い方です filter などパワフルな変換機能や新しい for-await-in 構文が使えますこれによりコーディングに集中でき簿記の複製に悩むこともありません全て1つの場所にまとまっているからです AsyncStream の並外れた柔軟性は独自の非同期シーケンス作成を可能にしますこれは1つの例ですが自分のコードに応用できるケースは沢山あります AsyncStream では既存のコードを応用して AsyncSequence にすることができます safety や反復放棄など AsyncSequence からと思われるものなら何でも処理しますがバッファリングの処理もできます AsyncStream は独自の非同期シーケンスや独自のAPIからの適切な戻り型を構築するには確実な方法です生成中される要素のソースが唯一のコンストラクタだからですではスローエラーを表示するには？そのための型があります！ AsyncThrowingStream は AsyngStream と似ていますがエラーの処理ができます AsyngStream の柔軟性と安全性を備えつつ反復からスローされるエラーを処理できるのです AsyncSequence は本当にパワフルなツールで複数の非同期値を安全かつ簡単に処理することが可能です Sequence の使い方を知っているなら AsyncSequence の使い方は知っているも同然ですさて非同期シーケンスの内容や使い方 AsyncStream の導入方法を説明してきましたまたその理論や定義を深く掘り下げ新しく導入された非同期シーケンスにも触れましたそして自分で構築する方法も説明しました皆様が今後どう活用されるか楽しみにしています ♪

0:37 - QuakesTool

@main
struct QuakesTool {
    static func main() async throws {
        let endpointURL = URL(string: "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_month.csv")!

        // skip the header line and iterate each one 
        // to extract the magnitude, time, latitude and longitude
        for try await event in endpointURL.lines.dropFirst() {
            let values = event.split(separator: ",")
            let time = values[0]
            let latitude = values[1]
            let longitude = values[2]
            let magnitude = values[4]
            print("Magnitude \(magnitude) on \(time) at \(latitude) \(longitude)")
        }
    }
}

3:24 - Iterating a Sequence

for quake in quakes {
    if quake.magnitude > 3 {
        displaySignificantEarthquake(quake)
    }
}

3:52 - How the compiler handles iteration

var iterator = quakes.makeIterator()
while let quake = iterator.next() {
    if quake.magnitude > 3 {
        displaySignificantEarthquake(quake)
    }
}

4:11 - How the compiler handles asynchronous iteration

var iterator = quakes.makeAsyncIterator()
while let quake = await iterator.next() {
    if quake.magnitude > 3 {
        displaySignificantEarthquake(quake)
    }
}

4:28 - Iterating an AsyncSequence

for await quake in quakes {
    if quake.magnitude > 3 {
        displaySignificantEarthquake(quake)
    }
}

5:36 - Terminating iteration early by breaking

for await quake in quakes {
    if quake.location == nil {
        break
    }
    if quake.magnitude > 3 {
        displaySignificantEarthquake(quake)
    }
}

5:51 - Skipping values by continuing

for await quake in quakes {
    if quake.depth > 5 {
        continue
    }
    if quake.magnitude > 3 {
        displaySignificantEarthquake(quake)
    }
}

6:05 - AsyncSequence might throw

do {
    for try await quake in quakeDownload {
        ...
    }
} catch {
    ...
}

7:15 - Concurrently iterating inside an async task

let iteration1 = Task {
    for await quake in quakes {
        ...
    }
}

let iteration2 = Task {
    do {
        for try await quake in quakeDownload {
            ...
        }
    } catch {
        ...
    }
}

//... later on  
iteration1.cancel()
iteration2.cancel()

7:56 - Reading bytes from a FileHandle

for try await line in FileHandle.standardInput.bytes.lines {
    ...
}

8:16 - Reading lines from a URL

let url = URL(fileURLWithPath: "/tmp/somefile.txt")
for try await line in url.lines {
    ...
}

8:49 - Reading bytes from a URLSession

let (bytes, response) = try await URLSession.shared.bytes(from: url)

guard let httpResponse = response as? HTTPURLResponse,
      httpResponse.statusCode == 200 /* OK */
else {
    throw MyNetworkingError.invalidServerResponse
}

for try await byte in bytes {
    ...
}

9:12 - Notifications

let center = NotificationCenter.default
let notification = await center.notifications(named: .NSPersistentStoreRemoteChange).first {
    $0.userInfo[NSStoreUUIDKey] == storeUUID
}

11:10 - Using an AsyncStream

class QuakeMonitor {
    var quakeHandler: (Quake) -> Void
    func startMonitoring()
    func stopMonitoring()
}

let quakes = AsyncStream(Quake.self) { continuation in
    let monitor = QuakeMonitor()
    monitor.quakeHandler = { quake in
        continuation.yield(quake)
    }
    continuation.onTermination = { @Sendable _ in
        monitor.stopMonitoring()
    }
    monitor.startMonitoring()
}

let significantQuakes = quakes.filter { quake in
    quake.magnitude > 3
}

for await quake in significantQuakes {
    ...
}

特定のトピックをお探しの場合は、上にトピックを入力すると、関連するトピックにすばやく移動できます。