[Swift]RxSwiftとSwift Concurrencyの相互変換

概要

同じプロジェクト内にRxSwiftとSwift Concurrencyのasync/awaitで書かれた処理が混在している場合に、相互変換して合成可能にする方法を紹介します。

バージョン

• XCode 13.2.1
• Swift 5.5.2
• RxSwift 6.5.0

相互変換

await(Swift Concurrency)->Single(RxSwift)

以下のコードでSwift Concurrencyのasync/await関数からRxSwiftのSingleに変換できます。
引数にasync/await関数を実行するクロージャを指定しています。

await->Single

extension PrimitiveSequence where Trait == SingleTrait {
    static func fromAwait(task: @escaping () async throws -> Element) -> Single<Element> {
        Single<Element>.create { observer in
            Task.detached {
                do {
                    let value = try await task()
                    observer(.success(value))
                } catch {
                    observer(.failure(error))
                }
            }
            return Disposables.create()
        }
    }
}

以下の例では2秒待ってからコンソールへの表示が実行されます。
引数に指定した秒数を待つTask.sleepをSingleに変換しています。

使用例

_ = Single.fromAwait {
    try await Task.sleep(nanoseconds: UInt64(seconds) * 1000 * 1000 * 1000)
    return "await"
}
.subscribe(onSuccess: { print($0) }) // 2秒後に文字列"await"を表示

Single(RxSwift)->await(Swift Concurrency)

以下のコードでRxSwiftのSingleからSwift Concurrencyのasync/await関数に変換できます。

await->Single

extension PrimitiveSequence where Trait == SingleTrait {
    func toAwait() async throws -> Element {
        try await withCheckedThrowingContinuation { continuation in
            _ = self.subscribe(onSuccess: { continuation.resume(returning: $0) },
                               onFailure: { continuation.resume(throwing: $0) })
        }
    }
}

以下ではSingle.timerをawaitして利用しています。

使用例

Task.detached {
    let result = try! await Single<Int>.timer(.seconds(2), scheduler: MainScheduler.instance)
        .map { _ in "rx" }
        .toAwait()
    print(result)  // 2秒後に文字列"rx"を表示
}

合成

Swift Concurrencyに統一する例

以下のコードでSingleで表現されたタスクを最終的にasync/awaitの流れに組み込んでいます。
両者ともタイマーで2秒待つタスクですが、並行して実行しているので2秒で終わります。

Task.detached {
    print(Date())  // 開始時刻の確認
    async let rxTask = Single<Int>.timer(.seconds(2), scheduler: MainScheduler.instance)
        .map  { _ in "rx" }
        .toAwait()
    async let awaitTask: String = {
        try! await Task.sleep(nanoseconds: 2 * 1000 * 1000 * 1000)
        return "await"
    }()
    let rxResult = try! await rxTask
    let awaitResult = try! await awaitTask
    print(Date()) // 2秒後に実行される
    print("\(rxResult), \(awaitResult)") // "rx, await"
}

RxSwiftに統一する例

以下のコードでasync/awaitの関数をSingle.zipで合成しています。
先程の例と同様に並行して実行しているので2秒で終わります。

let rxTask = Single<Int>.timer(.seconds(2), scheduler: MainScheduler.instance)
    .map { _ in "rx" }

let awaitTask: Single<String> = Single.fromAwait {
    try await timer(seconds: 2)
    return "await"
}

_ = Single.zip(rxTask,
               awaitTask)
    .do(onSubscribe: {
        print(Date()) // 開始時刻の確認
    })
    .subscribe(onSuccess: { rxResult, awaitResult in
        print("\(rxResult), \(awaitResult)") // "rx, await"
        print(Date()) // 2秒後に実行される
    })