ttstd_family_care/.agents/skills/kotlin-tooling-java-to-kotlin/references/frameworks/RXJAVA.md

RxJava to Coroutines/Flow Conversion Guide

When This Applies

Detected when imports match io.reactivex.* or rx.*. This is a significant paradigm shift — RxJava reactive types map to Kotlin coroutines and Flow.

Key Rules

1. Dependency setup

Add kotlinx-coroutines-core and kotlinx-coroutines-rx3 (or kotlinx-coroutines-rx2) as dependencies if performing a gradual migration. The bridge library provides extension functions like asFlow() and asObservable() for interop at module boundaries.

2. Type mapping

RxJava	Kotlin
Observable<T>	Flow<T>
Flowable<T>	Flow<T> (backpressure is built-in)
Single<T>	suspend fun: T
Maybe<T>	suspend fun: T?
Completable	suspend fun returning Unit
Disposable	Job (from coroutines)
CompositeDisposable	CoroutineScope (structured concurrency)

3. Operator mapping

RxJava	Kotlin Flow
subscribeOn(Schedulers.io())	flowOn(Dispatchers.IO)
observeOn(AndroidSchedulers.mainThread())	flowOn(Dispatchers.Main) or collect on Main
flatMap	flatMapMerge or flatMapConcat
map	map (same)
filter	filter (same)
zip	combine or zip
merge	merge
concat	flatMapConcat
onErrorReturn	catch { emit(default) }
doOnNext	onEach
subscribe()	collect {} in a coroutine scope

4. Error handling

RxJava's onError callback maps to Flow's catch operator or a try-catch block wrapping the collect call. In suspend functions (replacing Single/Completable), use standard try-catch.

5. Backpressure

Flow has built-in backpressure via suspension. There is no need for a separate Flowable type — all Flow instances support backpressure by default.

6. Threading

flowOn changes the upstream dispatcher (analogous to subscribeOn). Collection always happens on the caller's dispatcher. To collect on a specific dispatcher, launch the collecting coroutine in the desired scope.

7. Lifecycle and cancellation

RxJava's Disposable / CompositeDisposable pattern is replaced by structured concurrency. Cancelling a CoroutineScope cancels all child coroutines and flow collections automatically.

---

Example: Converting an Observable Chain to Flow

Java Input

package com.acme.data;

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.schedulers.Schedulers;
import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.disposables.CompositeDisposable;

/**
 * Repository that streams user data from a remote source.
 */
public class UserRepository {

    private final UserApi api;
    private final CompositeDisposable disposables = new CompositeDisposable();

    public UserRepository(UserApi api) {
        this.api = api;
    }

    public Observable<List<User>> getActiveUsers() {
        return api.getAllUsers()
                .subscribeOn(Schedulers.io())
                .map(users -> filterActive(users))
                .doOnNext(users -> logCount(users))
                .onErrorReturn(throwable -> Collections.emptyList());
    }

    public void observeUsers(UserCallback callback) {
        disposables.add(
            getActiveUsers()
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(
                    users -> callback.onUsers(users),
                    error -> callback.onError(error)
                )
        );
    }

    public void clear() {
        disposables.clear();
    }

    private List<User> filterActive(List<User> users) {
        return users.stream().filter(User::isActive).collect(Collectors.toList());
    }

    private void logCount(List<User> users) {
        System.out.println("Active users: " + users.size());
    }
}

Kotlin Output

package com.acme.data

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.launch

/**
 * Repository that streams user data from a remote source.
 */
class UserRepository(
    private val api: UserApi
) {
    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Main)

    fun getActiveUsers(): Flow<List<User>> =
        api.getAllUsers()
            .map { users -> users.filter { it.isActive } }
            .onEach { users -> println("Active users: ${users.size}") }
            .catch { emit(emptyList()) }
            .flowOn(Dispatchers.IO)

    fun observeUsers(callback: UserCallback) {
        scope.launch {
            getActiveUsers().collect { users ->
                callback.onUsers(users)
            }
        }
    }

    fun clear() {
        scope.cancel()
    }
}

Key points:

• Observable<List<User>> becomes Flow<List<User>>.
• subscribeOn(Schedulers.io()) becomes flowOn(Dispatchers.IO) at the end of the chain (it affects all upstream operators).
• CompositeDisposable is replaced by a CoroutineScope with SupervisorJob. Calling scope.cancel() cancels all active collections.
• doOnNext becomes onEach.
• onErrorReturn becomes catch { emit(emptyList()) }.
• observeOn(AndroidSchedulers.mainThread()) is unnecessary because scope already uses Dispatchers.Main, and collect runs on the collector's dispatcher.
• Java streams (filter + collect) become Kotlin's filter directly on the list.