rxjava-migration

RxJava 迁移

逐步将 RxJava 迁移到 Kotlin 协程/Flow。仅当用户明确要求迁移 RxJava 代码时，才调用此技能。 简单情况可直接映射；复杂情况需要先制定策略并征求用户意见，然后才能编写任何代码。

第 1 步：复杂度门槛

编写任何迁移后的代码之前，先进行分类。满足以下任意条件的链都属于复杂：嵌套的 flatMap/switchMap 中包含内部 Single/Observable；自定义 Scheduler；复杂的错误恢复（retryWhen、退避、重试次数）；多源 zip/combineLatest（3 个及以上源）；带显式背压策略的 Flowable；跨类共享的 Subject；或不明确的 API 返回类型。任何一项复杂条件都会使整个链变为复杂——即使是只有两个操作符、但包含 retryWhen 的链也属于复杂情况。

迁移前务必确认每个被调用的 API 返回什么——不要想当然。 你认为已经迁移的叶节点，仍可能返回 Single/Observable。

类型映射（明确说明——不要默默转换）

RxJava	协程/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 时，它会通过 withContext 管理自己的调度器——subscribeOn 没有对应项，直接删除即可。不要再用另一个 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）；互操作仅限于分层边界，切勿在同一个函数体内混用 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 会自动取消。）