/kotlin-tooling-java-to-kotlin

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

3. Operator mapping

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.