Reactive Programming with RxJava

Update note: I wrote this series for RxJava 2.x in 2019. RxJava 3.x changed some APIs. See the RxJava 3.x migration guide when using that version.

I enjoyed Thomas Nield’s Learning RxJava. These are my notes on RxJava 2.

Why RxJava?

• Combine events from several sources into one stream.
• Move work to a scheduler and deliver results to a chosen thread.
• Transform values, stop subscriptions, and recover from errors with operators.

Quickstart

In RxJava 2, an Observable<T> emits values of type T to a subscriber. Operators can transform those values before the subscriber receives them.

This Observable<String> contains three strings:

fun main() {
    val observable = Observable.just("Hello", "world", "!")
}

This method creates an observable but does not subscribe. Add a subscriber to print its values:

fun main() {
    val observable = Observable.just("Hello", "world", "!")
    observable.subscribe {
        print("$it ")
    }
}

This time, the output is the following:

Hello world !␠

The subscriber prints each string followed by a space. In the output above, ␠ marks the final space.

Use map() to transform each string before printing it:

fun main() {
    val observable = Observable.just("Hello", "world", "!")
    observable.map { it.uppercase() }.subscribe { print("$it ") }
}

The output uses the same spacing:

HELLO WORLD !␠

RxJava vs Java 8 streams

An Observable emits items to subscribers. Java streams and Kotlin sequences yield items when consumers request them. RxJava also supplies operators for asynchronous events and scheduler control.

RxJava Series Guide

This is a comprehensive guide to RxJava organized by topic. Follow the links below for in-depth coverage:

Fundamentals

• Observable & Observer - Core concepts and Observable factories
• Hot vs Cold Observable - Understanding observable behavior patterns
• Observable Factories - Additional factory methods (range, interval, timer, etc.)
• Disposing - Resource management and stopping emissions

Operators

• Filtering & Control: Suppressing - filter, take, skip, distinct
• Transformation: Transforming - map, flatMap, concatMap, switchMap
• Aggregation: Reducing - count, reduce, all, any
• Collection: Collection - toList, toMap, collect
• Error Handling: Recovery - onErrorReturn, onErrorResumeNext
• Side Effects: Action - doOnNext, doOnComplete, doOnError

Advanced Topics

• Combining Observables - merge, concat, zip, combineLatest
• Multicasting - ConnectableObservable and sharing streams
• Replaying and Caching - replay() and cache() operators
• Subjects - PublishSubject, BehaviorSubject, and more
• Concurrency - subscribeOn and observeOn with Schedulers
• Parallelisation - Parallel execution strategies

Flow Control

• Buffering - Batch emissions into collections
• Windowing - Batch emissions into separate Observables
• Throttling - Control emission rate
• Switching - Cancel previous Observables

Backpressure

• Backpressure - Understanding and handling backpressure
• Flowable - Observable with backpressure support
• Subscriber - Consuming Flowables

Customization

• Transformers - Reusable operator chains
• Custom Operators - Building your own operators

Sources

• Learning RxJava
• ReactiveX Documentation
• RxJava Github
• RxMarbles

These Kotlin examples use RxJava directly. RxKotlin offers Kotlin-specific extensions.

Mouaad Aallam

Software Engineer