rxjava-migration

RxJava の移行

RxJava から Kotlin のコルーチン／Flow へ段階的に移行します。**ユーザーが RxJava コードの移行を明示的に依頼した場合にのみ、このスキルを実行してください。**単純なケースは直接対応できますが、複雑なケースではコードを書く前に、方針を立ててユーザーの入力を得る必要があります。

Step 1: 複雑さの判定

移行後のコードを書く前に分類してください。以下のいずれかに該当するチェーンは複雑です。内側に Single／Observable を持つネストされた flatMap／switchMap、カスタム Scheduler、複雑なエラーリカバリー（retryWhen、バックオフ、リトライ回数）、複数ソースの zip／combineLatest（3 つ以上のソース）、明示的なバックプレッシャー戦略を指定した Flowable、クラス間で共有される Subject、または API の戻り値の型が不明確な場合です。複雑さの基準に 1 つでも該当すれば、チェーン全体が複雑です。retryWhen を使う 2 オペレーターのチェーンも複雑です。

**移行前に、呼び出す API がそれぞれ何を返すか必ず確認してください。推測してはいけません。**移行済みだと思っている末端の API が、まだ Single／Observable を返す場合があります。

型のマッピング（明示してください。黙って変換してはいけません）

RxJava	Coroutines/Flow
Observable<T> / Flowable<T>	Flow<T>（Flowable の場合は buffer() / conflate() で元のバックプレッシャーに合わせます）
Single<T> / Maybe<T> / Completable	suspend fun：T / T? / Unit
PublishSubject / ReplaySubject(n)	MutableSharedFlow(replay = 0 / n)
BehaviorSubject<T>	確認してください — MutableStateFlow と MutableSharedFlow(replay = 1) のどちらにするか

BehaviorSubject の場合は、まず確認してください。StateFlow は常に現在値を持ちます（初期値が必要で、.value を公開し、新しいコレクターにその値をリプレイします）。SharedFlow(replay = 1) も最後の値をリプレイしますが、.value はなく、初期値も必要ありません。「常に初期値があり、.value へのアクセスも必要ですか？ それとも、ストリームの開始時に空の場合がありますか？」

スケジューラーと observeOn の切り替え

Schedulers.io() → Dispatchers.IO、computation() → Default、mainThread() → Main、single() → newSingleThreadContext(...)、newThread() → IO（タスクごとにスレッドを生成する方法は、コルーチンでは一般的ではありません）。

• observeOn(AndroidSchedulers.mainThread()) は、リポジトリー／ユースケース内でディスパッチャーを切り替える処理にはなりません。これは呼び出し元が Main 上でコレクションすることを意味し、コルーチンでは ViewModel の責務です（viewModelScope は Main 上で実行されます）。この切り替えについて開発者に説明してください。
• 呼び出す API がすでに suspend fun の場合は、API 自体が withContext でディスパッチャーを管理します。subscribeOn に相当するものはなく、そのまま削除します。すでに Main セーフな suspend 関数を、さらに withContext で囲んではいけません。
• subscribeOn + observeOn → flowOn は上流にのみ適用されます。それに応じて構造を変更してください。

オペレーターの完全なマッピングについては、migration-map.md を参照してください。

リトライ — 正確性に関する注意点

RxJava の retryWhen は状態を保持します。Flow の retry の述語に渡されるのはインデックスではなく、Throwable です。

// WRONG — 'attempt' is a Throwable, not a Long; this does not compile
flow.retry(3) { attempt -> delay(attempt * 1000L); true }

// CORRECT — retry(n)'s predicate gets the cause
flow.retry(3) { cause -> cause is IOException }

// CORRECT — stateful backoff: attempt IS the 0-based index here
flow.retryWhen { cause, attempt ->
    if (cause is IOException && attempt < 3) { delay((attempt + 1) * 1000L); true } else false
}

試行回数に依存するポリシーには、retryWhen { cause, attempt -> } を使用してください。

段階的な移行中の相互運用

kotlinx-coroutines-rx3（または rx2）を追加し、相互運用はレイヤーの境界でのみ行ってください。1 つの関数本体内で RxJava とコルーチンを混在させてはいけません。

observable.asFlow(); single.await(); maybe.awaitSingleOrNull(); completable.await()   // Rx → coroutines
flow.asObservable(); flow.asSingle()   // coroutines → Rx (asSingle throws if the flow emits 0 or 2+ elements)

末端から上流へ（データソース → リポジトリー → ユースケース → ViewModel）の順に移行し、レイヤー全体の移行が完了したら各ブリッジを削除して、レイヤーごとにコミットしてください。

複雑なケース — 移行前に中断して確認してください

retryWhen（回数は？ 線形バックオフと指数バックオフのどちら？ 対象となるエラーの種類は？ リトライ上限に達した後はどうする？）；カスタムのScheduler（どのCoroutineDispatcherを使う？）；Flowableのバックプレッシャー（buffer／conflate／DROP_OLDESTのどれ？）；書き込みに対するflatMap／switchMap — 読み取り（検索／ライブデータ）ではswitchMap → flatMapLatestで問題ありませんが、実行中の処理がキャンセルされるため、書き込みでは危険です；複数のクラスで共有されるSubject（状態 → StateFlow、イベント → SharedFlow？）。onErrorResumeNext → catchでは、キャンセルを再スローします：catch { e -> if (e is CancellationException) throw e else emit(fallback) }。

（onCleared()内のcompositeDisposable.clear() → 完全に削除します：ViewModelがクリアされると、viewModelScopeは自動的にキャンセルされます。）