rxjava-to-coroutines-migration

RxJava から Kotlin Coroutines への移行スキル

Android または Kotlin のコードベースを、RxJava から Kotlin Coroutines と Flow へ安全かつ慣用的にリファクタリングするための、専門的なスキルです。

移行マッピングガイド

RxJava のコンポーネントを Kotlin Coroutines に移行する際は、次の標準的なマッピングを使用します。

1. 基本型

• Single<T> -> suspend fun ...(): T
  • 単一の非同期値です。
• Maybe<T> -> suspend fun ...(): T?
  • 存在しない場合もある、単一の非同期値です。
• Completable -> suspend fun ...()
  • 値を返さずに完了する非同期処理です。
• Observable<T> -> Flow<T>
  • コールドな値のストリームです。
• Flowable<T> -> Flow<T>
  • Coroutines の Flow はバックプレッシャーをネイティブに処理します。

2. Subject からホットフローへ

• PublishSubject<T> -> MutableSharedFlow<T>
  • 複数のサブスクライバーにイベントをブロードキャストします。バッファリングが必要な場合は、MutableSharedFlow(extraBufferCapacity = ...) を使用します。
• BehaviorSubject<T> -> MutableStateFlow<T>
  • 状態を保持し、新しいサブスクライバーに現在または最新の値を出力します。初期値が必要です。
• ReplaySubject<T> -> MutableSharedFlow<T>(replay = N)
  • 新しいサブスクライバーに、出力済みの直近 N 個の値を再出力します。

3. Scheduler から Dispatcher へ

• Schedulers.io() -> Dispatchers.IO
• Schedulers.computation() -> Dispatchers.Default
• AndroidSchedulers.mainThread() -> Dispatchers.Main
• コンテキストの切り替え: subscribeOn と observeOn は通常、withContext(Dispatcher) または Flow の flowOn(Dispatcher) に置き換えます。

4. オペレーター

• map -> map
• filter -> filter
• flatMap -> flatMapMerge（並行）または flatMapConcat（逐次）
• switchMap -> flatMapLatest
• doOnNext / doOnSuccess -> onEach
• onErrorReturn / onErrorResumeNext -> catch { emit(...) }
• startWith -> onStart { emit(...) }
• combineLatest -> combine
• zip -> zip
• delay -> delay（サスペンド関数）または onEach { delay(...) }

5. 実行とライフサイクル

• subscribe() -> Flow の場合は collect {}、サスペンド関数の場合は CoroutineScope 内で直接呼び出します。
• Disposable.dispose() -> Job.cancel()
• CompositeDisposable.clear() -> 親の CoroutineScope または Job をキャンセルします。

実行手順

1. RxJava のチェーンを分析する: ソース型（Single、Observable など）、使用されているオペレーター、サブスクリプションが行われる場所を特定します。
2. ソースを変換する: まずリポジトリ層またはデータソース層の戻り値の型を変更します。単発の処理は suspend 関数に、ストリームは Flow に変換します。
3. オペレーターを書き換える: RxJava のオペレーターを、Flow または Coroutine の同等の処理に置き換えます。多くの RxJava オペレーターは、map または onEach ブロック内の標準的な Kotlin コレクション／シーケンス操作にそのまま置き換えられることに注意してください。
4. サブスクリプションを更新する: ViewModel または Presenter で、.subscribe(...) を launch { ... } と .collect { ... } に置き換えます。起動処理が適切なライフサイクルスコープ（例: viewModelScope）に紐付いていることを確認します。
5. エラーを処理する: onError ブロックを、サスペンド関数を囲む try/catch、または Flow の .catch { } オペレーターに置き換えます。
6. スレッド処理を行う: .subscribeOn() と .observeOn() を削除します。必要に応じて withContext を使用するか、上流のフローのコンテキストを変更するために .flowOn() を使用します。

変換例

RxJava:

fun getUser(id: String): Single<User> { ... }

disposable.add(
    getUser("123")
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe({ user ->
            view.showUser(user)
        }, { error ->
            view.showError(error)
        })
)

Coroutines/Flow:

suspend fun getUser(id: String): User { ... } // Internally uses withContext(Dispatchers.IO) if needed

viewModelScope.launch {
    try {
        val user = getUser("123")
        view.showUser(user)
    } catch (e: Exception) {
        view.showError(e)
    }
}

ベストプラクティス

• サスペンド関数を優先する: 複数の値が時間の経過とともに実際に流れる場合を除き、Flow ではなく suspend 関数を基本とします。Single と Completable は、ほとんどの場合 suspend 関数になります。
• 状態の処理: BehaviorSubject や LiveData の代わりに、ViewModel で StateFlow を使用して UI に状態を公開します。
• ライフサイクルへの対応: UI レイヤーで Flow を収集する際は、ビューが表示されていない間のバックグラウンド処理を避けるために、repeatOnLifecycle または flowWithLifecycle を使用します。