ANDROID: RxJava in Android

Modern Android applications are fundamentally event-driven. User touches, network responses, sensor readings, database changes — all of these are events that happen asynchronously, often in parallel, often needing transformation and combination before they can update the UI.

RxJava gives you a consistent, composable model for dealing with all of these asynchronous data streams in a single, unified way.

What is RxJava?

RxJava is a library for the JVM that implements the ReactiveX (Reactive Extensions) specification. ReactiveX is a cross-language standard for reactive programming — the same patterns exist in RxJS (JavaScript), RxSwift (Swift), Rx.NET (C#), and many others. Learning RxJava gives you conceptual knowledge that transfers across platforms.

At its core, RxJava is built around one idea: treat everything as a data stream.

• A button click is a stream of click events
• A network response is a stream of one result (or an error)
• A database query is a stream of result sets
• A timer is a stream of tick events

Once everything is a stream, you can apply consistent operators to transform, filter, combine, and react to data regardless of its source.

Data Source                  Operators                 Observer
-----------                  ---------                 --------
Network API  ----+
Database     ----|---> [map] -> [filter] -> [merge] --> [onNext]
Sensor data  ----|                                  --> [onError]
User input   ----+                                  --> [onComplete]

Why RxJava in Android?

Android apps constantly juggle background tasks, UI updates, and real-time data. RxJava makes it easier to:

1. Manage Asynchronous Data Streams — handle complex workflows (network call, transform response, update database, update UI) as a single composed pipeline rather than nested callbacks
2. Simplify Background Tasks — move work off the main thread and back onto it with a single operator, without manual thread management
3. Centralize Error Handling — errors propagate through the stream and are handled in one place (onError), rather than scattered across multiple callback error handlers
4. Reduce Boilerplate — replace callback hell and complex threading code with readable, declarative operator chains

Core Concepts

Observable

An Observable is the source — the entity that emits data. It represents a sequence of zero or more items, potentially followed by completion or an error.

Observable lifecycle:

  [create] --> [subscribe] --> [emit item] --> [emit item] --> [complete]
                                                         \--> [error]

An Observable emits three types of events to its subscribers:

Once an Observable emits an error or completes, it will not emit any more items.

Observer

An Observer subscribes to an Observable and defines how to handle each type of event. The Observer is the consumer — the code that reacts to the data.

An Observer implements three methods:

• onNext(item) — called for each emitted item
• onError(error) — called once if an error occurs
• onComplete() — called once when the stream ends normally

val numberObservable = Observable.just(1, 2, 3, 4, 5)

numberObservable
    .subscribe(
        { number -> println("Received: $number") },   // onNext
        { error -> println("Error: $error") },         // onError
        { println("Completed") }                       // onComplete
    )

Received: 1
Received: 2
Received: 3
Received: 4
Received: 5
Completed

When you only care about the emitted items and not errors or completion, you can use a simplified single-lambda form:

observable.subscribe { item ->
    // React to each emitted value
}

This is common for simple event streams where errors are handled at a higher level.

Operators

Operators are the heart of RxJava's power. They are methods on Observable that transform, filter, combine, or otherwise manipulate the data stream. Operators return new Observables, so they can be chained together to build complex pipelines.

Source Observable
      |
      v
  [map]         -- transform each item
      |
      v
  [filter]      -- discard items that don't match
      |
      v
  [flatMap]     -- transform each item into a new Observable, merge results
      |
      v
  [take(5)]     -- only emit the first 5 items
      |
      v
  Observer

Transformation operators:

• map — transform each item to a different type or value
• flatMap — transform each item into an Observable and flatten the results
• concatMap — like flatMap but preserves order
• switchMap — like flatMap but cancels previous inner Observable when a new item arrives

Filtering operators:

• filter — only emit items that match a predicate
• take(n) — only emit the first n items
• skip(n) — skip the first n items
• debounce — only emit after a specified time period has passed with no new items (useful for search input)
• distinctUntilChanged — suppress consecutive duplicate items

Combining operators:

• zip — combine the latest items from multiple Observables into one item
• merge — combine multiple Observables into one, interleaving their items
• combineLatest — emit an item whenever any source Observable emits, combining the latest from all

Example — transforming a stream:

Observable.just("alice", "bob", "charlie", "dave")
    .filter { it.length > 3 }           // only names longer than 3 characters
    .map { it.uppercase() }             // convert to uppercase
    .subscribe { name ->
        println(name)
    }

// Output: ALICE, CHARLIE, DAVE

Schedulers

One of RxJava's most powerful features is explicit, composable thread management through Schedulers. A Scheduler determines which thread an Observable operates on.

Two key operators control threading:

• subscribeOn(scheduler) — determines which thread the Observable's source does its work on. Typically applied once, affects the entire upstream.
• observeOn(scheduler) — determines which thread subsequent operators (and the Observer) run on. Can be applied multiple times to switch threads mid-pipeline.

Observable.create { source data }         <-- subscribeOn(Schedulers.io())
      |
  [map / filter / transform]              <-- still on IO thread
      |
  observeOn(AndroidSchedulers.mainThread())
      |
  Observer.onNext { update UI }           <-- now on Main thread

Built-in Schedulers:

Typical Android pattern:

apiService.fetchUser(userId)             // Returns Observable<User>
    .subscribeOn(Schedulers.io())        // Do network call on IO thread
    .map { user -> user.toDisplayModel() } // Transform on IO thread
    .observeOn(AndroidSchedulers.mainThread()) // Switch to UI thread
    .subscribe(
        { user -> updateUserUI(user) },
        { error -> showError(error) }
    )

Disposable

When you subscribe to an Observable, you receive a Disposable — a handle to the active subscription. A Disposable lets you cancel the subscription and stop receiving items.

This is critical in Android because of the Activity/Fragment lifecycle. If a network call completes after an Activity has been destroyed, and the subscription is still active, the callback will try to update a destroyed view — causing a crash or memory leak.

class MyActivity : AppCompatActivity() {

    private val disposables = CompositeDisposable()

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)

        val disposable = someObservable
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe { data -> updateUI(data) }

        disposables.add(disposable)
    }

    override fun onDestroy() {
        super.onDestroy()
        disposables.clear() // Cancel all active subscriptions
    }
}

CompositeDisposable is the standard pattern — collect all Disposables into one container and clear them all at once in onDestroy().

RxJava Observable Types

RxJava offers several Observable variants optimized for different scenarios:

For most Android use cases like network calls, Single<T> is the most appropriate:

// Single — emits exactly one response or one error
apiService.getUser(id): Single<User>
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(
        { user -> showUser(user) },
        { error -> showError(error) }
    )

A Practical Android Example

Here is a real-world-style example that fetches data from a network, transforms it, and updates the UI:

class UserRepository(private val api: ApiService, private val db: UserDao) {

    fun getUser(userId: Int): Observable<User> {
        return api.fetchUser(userId)
            .subscribeOn(Schedulers.io())
            .doOnNext { user -> db.insert(user) }  // Side effect: save to DB
            .onErrorResumeNext { _ ->
                // If network fails, return cached DB result
                db.getUser(userId).toObservable()
            }
    }
}

class UserViewModel(private val repo: UserRepository) {

    private val disposables = CompositeDisposable()
    val userLiveData = MutableLiveData<User>()

    fun loadUser(id: Int) {
        val disposable = repo.getUser(id)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(
                { user -> userLiveData.value = user },
                { error -> /* handle error */ }
            )
        disposables.add(disposable)
    }

    override fun onCleared() {
        disposables.clear()
    }
}

Summary

RxJava is a mature, powerful tool for reactive programming in Android. Its greatest strengths are in complex async workflows, explicit threading, and rich data transformation pipelines. While Kotlin Coroutines and Flow are now the recommended choice for new Android projects, RxJava remains widely used and understanding it is essential for working with existing Android codebases.

Mastering RxJava means mastering the art of thinking in streams.