rxjava-migration

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

RxJava Migration

RxJava 迁移

Migrate RxJava to Kotlin coroutines/flows incrementally. Only invoke this skill when the user explicitly asks to migrate RxJava code. Simple cases map directly; complex cases need a strategy and user input before any code is written.

逐步将RxJava迁移到Kotlin协程/Flow。仅当用户明确要求迁移RxJava代码时才调用此技能。简单场景可直接映射；复杂场景则需要先制定策略并获取用户输入，再进行代码编写。

Step 1: Complexity gate

步骤1：复杂度判断

Classify before writing any migrated code. A chain is Complex if ANY of: nested

flatMap

switchMap

with an inner

Single

Observable

; a custom

Scheduler

; complex error recovery (

retryWhen

, backoff, retry counts); multi-source

zip

combineLatest

(3+ sources);

Flowable

with an explicit backpressure strategy; a

Subject

shared across classes; or an unclear API return type. Any single Complex criterion makes the whole chain Complex — a two-operator chain with

retryWhen

is Complex.

Always confirm what each called API returns before migrating — do not assume. A leaf you think is migrated may still return

Single

Observable

在编写迁移代码前先进行分类。如果满足以下任一条件，链式调用即为复杂场景：嵌套的

flatMap

switchMap

且内部包含

Single

Observable

；自定义

Scheduler

；复杂错误恢复（

retryWhen

、退避策略、重试次数）；多源

zip

combineLatest

（3个及以上源）；带有显式背压策略的

Flowable

；跨类共享的

Subject

；或不明确的API返回类型。只要满足任一复杂场景判定条件，整个链式调用就属于复杂场景——比如包含

retryWhen

的两操作符链式调用也属于复杂场景。

迁移前务必确认每个被调用API的返回类型——不要假设。你认为已完成迁移的叶子节点可能仍返回

Single

Observable

。

Type mapping (state it explicitly — don't silently transform)

类型映射（需明确说明，不要静默转换）

RxJava	Coroutines/Flow
`Observable<T>` / `Flowable<T>`	`Flow<T>` (Flowable: match the original backpressure with `buffer()` / `conflate()` )
`Single<T>` / `Maybe<T>` / `Completable`	`suspend fun` : `T` / `T?` / `Unit`
`PublishSubject` / `ReplaySubject(n)`	`MutableSharedFlow(replay = 0 / n)`
`BehaviorSubject<T>`	ask — `MutableStateFlow` vs `MutableSharedFlow(replay = 1)`

BehaviorSubject
— ask first:

StateFlow

always has a current value (needs an initial, exposes

.value

, replays it to new collectors);

SharedFlow(replay = 1)

also replays the last emission but has no

.value

and no initial-value requirement. "Do you always have an initial value and need
.value
access, or is the stream sometimes empty at start?"

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
，还是流在启动时有时为空？”

Schedulers and the

observeOn

shift

Scheduler 与

observeOn

的转变

Schedulers.io()

→

Dispatchers.IO

computation()

→

Default

mainThread()

→

Main

single()

→

newSingleThreadContext(...)

newThread()

→

IO

(per-task thread spawning isn't idiomatic in coroutines).

observeOn(AndroidSchedulers.mainThread())
does NOT become a dispatcher switch in the repository/use case — it means the caller collects on
```
Main
```
, which in the coroutines model is the ViewModel's job (
```
viewModelScope
```
runs on
```
Main
```
). Explain this shift to the developer.
When the called API is already a
suspend fun
, it manages its own dispatcher via
```
withContext
```
—
```
subscribeOn
```
has no equivalent and simply disappears. Do not wrap an already-main-safe suspend function in another
withContext
.
```
subscribeOn
```
+
```
observeOn
```
→
```
flowOn
```
applies upstream only; restructure accordingly.

For the full operator mapping, see

migration-map.md

Schedulers.io()

→

Dispatchers.IO

，

computation()

→

Default

，

mainThread()

→

Main

，

single()

→

newSingleThreadContext(...)

，

newThread()

→

IO

（在协程中，为每个任务生成线程并非惯用做法）。

observeOn(AndroidSchedulers.mainThread())
在仓库/用例层不会转换为调度器切换，这意味着调用者需在
```
Main
```
调度器上收集流，而在协程模型中这是ViewModel的职责（
```
viewModelScope
```
运行在
```
Main
```
调度器上）。需向开发者解释这一转变。
当被调用API已是
suspend fun
时，它会通过
```
withContext
```
管理自身的调度器——
```
subscribeOn
```
没有等效操作，直接移除即可。不要将已主线程安全的suspend函数再包裹在另一个
withContext
中。
```
subscribeOn
```
+
```
observeOn
```
→
```
flowOn
```
仅作用于上游；需相应调整结构。

完整的操作符映射请查看

migration-map.md

。

Retry — the accuracy trap

重试——准确性陷阱

RxJava's

retryWhen

is stateful; Flow's

retry

predicate receives the Throwable
, not an index.

kotlin

// 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
}

Use

retryWhen { cause, attempt -> }

for any policy that depends on the attempt count.

RxJava的

retryWhen

是有状态的；Flow的

retry

谓词接收的是**

Throwable

**，而非索引。

kotlin

// 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 -> }

。

Interop during incremental migration

增量迁移期间的互操作

Add

kotlinx-coroutines-rx3

(or

rx2

); keep interop at layer boundaries only, never mixing RxJava and coroutines inside one function body.

kotlin

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)

Migrate leaf-up (data source → repository → use case → ViewModel), remove each bridge once its layer is fully migrated, and commit per layer.

添加

kotlinx-coroutines-rx3

（或

rx2

）；仅在层边界使用互操作，切勿在单个函数体内混合RxJava和协程。

kotlin

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），每层完全迁移后移除对应的桥接代码，并按层提交。

Complex cases — stop and ask before migrating

复杂场景——迁移前先暂停并询问

retryWhen

(count? linear vs exponential? which error types? what after exhaustion?); a custom

Scheduler

(which

CoroutineDispatcher

?);

Flowable

backpressure (

buffer

conflate

DROP_OLDEST

?);

flatMap

switchMap

over writes —

switchMap

→

flatMapLatest

is safe for reads (search/live data) but cancels in-flight work, so it's dangerous for writes; a

Subject

shared across classes (state →

StateFlow

, event →

SharedFlow

?). For

onErrorResumeNext

→

catch

, rethrow cancellation:

catch { e -> if (e is CancellationException) throw e else emit(fallback) }

(

compositeDisposable.clear()

onCleared()

→ delete it entirely:

viewModelScope

is cancelled automatically when the ViewModel is cleared.)

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

会自动取消。）

rxjava-migration

Original

Translation

RxJava Migration

RxJava 迁移

Step 1: Complexity gate

步骤1：复杂度判断

Type mapping (state it explicitly — don't silently transform)

类型映射（需明确说明，不要静默转换）

Schedulers and the
`observeOn`
shift

Scheduler 与
`observeOn`
的转变

Retry — the accuracy trap

重试——准确性陷阱

Interop during incremental migration

增量迁移期间的互操作

Complex cases — stop and ask before migrating

复杂场景——迁移前先暂停并询问