× Home Tutorials Training Consulting Products Books Company Donate Contact us









NOW Hiring

Quick links

Share

This tutorial contains notes about RxJava 2.0.

1. RxJava 2.0

1.1. What is RxJava and reactive programming

In reactive programming the consumer reacts to the data as it comes in. This is the reason why asynchronous programming is also called reactive programming. Reactive programming allows to propagates event changes to registered observers.

RxJava is a port from Netflix of the Reactive Extensions (Rx) to Java. RxJava was open sourced 2014 and is hosted at http://reactivex.io/. The Java version of this concept is called RxJava and is hosted under https://github.com/ReactiveX/RxJava. RxJava is published under the Apache 2.0 license.

RxJava describes itself as an API for asynchronous programming with observable streams

1.2. Define a dependency to RxJava 2.0

As of this writing the version 2.0.4 is currently the released one. Replace g.a.v with 2.0.4 or a later version.

For Maven you can add a dependency with the following snippet

<dependency>
    <groupId>io.reactivex.rxjava2</groupId>
    <artifactId>rxjava</artifactId>
    <version>g.a.v</version>
</dependency>

For a Gradle build you can add RxJava via the following dependency statement.

compile group: 'io.reactivex.rxjava2', name: 'rxjava', version: 'g.a.v'

For OSGi environments, e.g., Eclipse RCP development, https://dl.bintray.com/simon-scholz/RxJava-OSGi/ can be used as p2 update site.

rxjava update site

1.3. Async programming

Nowadays programming in an imperative single threaded way usually leads to strange behaviors, blocking non responsive UIs and therefore a bad user experience.

This can be avoided by handling unpredicted things asynchronously. For example actively waiting for a database query or a webservice call can cause an application freeze, if the network is not responsive.

An example would be:

public List<Todo> getTodos() {

        List<Todo> todosFromWeb = // query a webservice (with bad network latency)

        return todosFromDb;
}

Calling the getTodos() method from the main thread or an UI thread will cause a non responsive application until the todosFromWeb arrive.

To improve this the query, which takes an unpredictable amount of time, should run in a different thread and inform the main thread when a result comes in.

public void getTodos(Consumer<List<Todo>> todosConsumer) {

        Thread thread = new Thread(()-> {
                List<Todo> todosFromWeb = // query a webservice

                todosConsumer.accept(todosFromWeb);
        });
        thread.start();
}

Now after calling the getTodos(Consumer<List<Todo>> todosConsumer) the main thread can continue working, is not blocked and can react once the accept method of the given Consumer is called.

1.4. Observables, Observers and Subscriptions

RxJava provides Observables and Observers. Observables can send out values. Observers, watch Observables by subscribing to them. Observers are notified when an Observable emits a value, when the Observable says an error has occurred. They are also notified when the Observable sends the information that it no longer has any values to emit. The corresponding functions are onNext, onError, and onCompleted() from the Observer interface. A instance of Subscription represents the connection between an observer and an observable. In you call unsubscribe() on this instance to remove the connection. This is for example useful to avoid updates to widgets after they have been disposed.

1.5. Defining the execution thread and the thread for observing

You can specify the thread in with the observable is executed via the subscribeOn() method. The thread in which the observers are executed can be defined via the observeOn() method.

2. Operators

You can register operations on your observers which allows you to manipulate the emission of an observable before passing it to the observer. For example the map method allows to register a Func1 which translates the input.

3. Using delays

Via the debounce(delay, TimeUnit.MILLISECONDS) method on the observer, you can instruct it to only emit the changes if the value has not changed for a predefined delay.

4. Creating Observables and Observers

RxJava provides several methods to create an observable.

  • Observable.just() - Allows to create an observable as wrapper around other data types

  • Observable.from() - takes a collection or an array and emits their values in their order in the data structure

  • Observable.fromCallable() - Allows to create an observable for a Callable`

To create observers:

  • Implement Action1 - Allow you to create a simple observer which has a call methods. This method is called if a new object is emitted.

4.1. Example for using Observable.just()

Observable.just() creates an Observable such that when an Observer subscribes, the onNext() of the Observer is immediately called with the argument provided to Observable.just().

import java.util.Arrays;
import java.util.List;

import rx.Observable;

public class RxJavaExample {
        public static void main(String[] args) {
                List<String> list = Arrays.asList("Android", "Ubuntu", "Mac OS"); (1)
                Observable<List<String>> listObservable = Observable.just(list);  (2)
                listObservable.subscribe(new Observer<List<String>>() {                          (3)

            @Override
            public void onCompleted() {}

            @Override
            public void onError(Throwable e) {}

            @Override
            public void onNext(List<String> list) {
                                System.out.println(list);
            }
        });
        }
}
1 Creates a list
2 Creates the Observable
3 Registers the Observer

5. Performing conversions

The following is a simple example of RxJava in which some conversion is performed.

import java.util.Arrays;
import java.util.List;

import rx.Observable;

public class RxJavaExample {
        public static void main(String[] args) {
                List<String> list = Arrays.asList("Hello", "Streams", "Not");
                Observable.from(list).
                                filter(s -> s.contains("e")).
                                map(s -> s.toUpperCase()).
                                reduce(new StringBuilder(), StringBuilder::append).
                                subscribe(System.out::print, e -> {},
                                () -> System.out.println("!"));
        }
}

5.1. Subjects

Subjects are objects that are both an Observable and an Observer. Subjects can be used as a pipe to translate data. An example for a subject is a PublishSubject. As soon as something is send to a PublishSubject it is immediately send out again.

6. Single Object aka Promise

In reactive programming Observables and Observers are used to interact reactively with each other.

Using promises for asynchronous calls has become really popular, especially when working with JavaScript. So a promise basically is a placeholder for a value, which can inform it’s observer once the promised value arrived.

RxJava comes with a type called Single, which is pretty similar to a promise.

Instances of Single work similar to Observable but they have only two callbacks onSuccess() and onError().

import io.reactivex.Single;

public Single<List<Todo>> getTodos() {

        return Single.create(emitter -> {
                Thread thread = new Thread(() -> {
                        try {
                                List<Todo> todosFromWeb = // query a webservice

                                emitter.onSuccess(todosFromWeb);
                        } catch (Exception e) {
                                emitter.onError(e);
                        }
                });
                thread.start();
        });
}

The Single instance can be used like this:

import io.reactivex.Single;
import io.reactivex.disposables.Disposable;
import io.reactivex.observers.DisposableSingleObserver;

Single<List<Todo>> todosSingle = getTodos();

todosSingle.subscribeWith(new DisposableSingleObserver<List<Todo>>() {

        @Override
        public void onSuccess(List<Todo> todos) {
                // work with the resulting todos
        }

        @Override
        public void onError(Throwable e) {
                // handle the error case
        }
});

7. Disposing subscriptions and using CompositeDisposable

When listers or subscribers are attached they usually are not supposed to listen eternally.

So it could happen that due to some state change the event being emitted by an observable might be not interesting any more.

import io.reactivex.Single;
import io.reactivex.disposables.Disposable;
import io.reactivex.observers.DisposableSingleObserver;

Single<List<Todo>> todosSingle = getTodos();

Disposable disposable = todosSingle.subscribeWith(new DisposableSingleObserver<List<Todo>>() {

        @Override
        public void onSuccess(List<Todo> todos) {
                // work with the resulting todos
        }

        @Override
        public void onError(Throwable e) {
                // handle the error case
        }
});

// continue working and dispose when value of the Single is not interesting any more
disposable.dispose();

The Single class and other observable classes offer different subscribe methods, which return a Disposable object.

When working with multiple subscriptions, which may become obsolete due to the same state change using a CompositeDisposable is pretty handy to dispose a collection of subscriptions.

import io.reactivex.Single;
import io.reactivex.disposables.Disposable;
import io.reactivex.observers.DisposableSingleObserver;
import io.reactivex.disposables.CompositeDisposable;

CompositeDisposable compositeDisposable = new CompositeDisposable();

Single<List<Todo>> todosSingle = getTodos();

Single<Happiness> happiness = getHappiness();

compositeDisposable.add(todosSingle.subscribeWith(new DisposableSingleObserver<List<Todo>>() {

        @Override
        public void onSuccess(List<Todo> todos) {
                // work with the resulting todos
        }

        @Override
        public void onError(Throwable e) {
                // handle the error case
        }
}));

compositeDisposable.add(happiness.subscribeWith(new DisposableSingleObserver<List<Todo>>() {

        @Override
        public void onSuccess(Happiness happiness) {
                // celebrate the happiness :-D
        }

        @Override
        public void onError(Throwable e) {
                System.err.println("Don't worry, be happy! :-P");
        }
}));

// continue working and dispose all subscriptions when the values from the Single objects are not interesting any more
compositeDisposable.dispose();

8. Caching values of completed observables

When working with observables doing async calls on every subscription on an observable is often not necessary.

It likely happens that observables are passed around in the application, without the need to do an such an expensive call all the time a subscription is added.

The following code does the expensive web query 4 times, even though doing this once would be fine, since the same Todo objects should be shown, but only in different ways.

Single<List<Todo>> todosSingle = Single.create(emitter -> {
        Thread thread = new Thread(() -> {
                try {
                        List<Todo> todosFromWeb = // query a webservice

                        System.out.println("Called 4 times!");

                        emitter.onSuccess(todosFromWeb);
                } catch (Exception e) {
                        emitter.onError(e);
                }
        });
        thread.start();
});

todosSingle.subscribe(... " Show todos times in a bar chart " ...);

showTodosInATable(todosSingle);

todosSingle.subscribe(... " Show todos in gant diagram " ...);

anotherMethodThatsSupposedToSubscribeTheSameSingle(todosSingle);

The next code snippet makes use of the cache method, so that the Single instance keeps its result, once it was successful for the first time.

Single<List<Todo>> todosSingle = Single.create(emitter -> {
        Thread thread = new Thread(() -> {
                try {
                        List<Todo> todosFromWeb = // query a webservice

                        System.out.println("I am only called once!");

                        emitter.onSuccess(todosFromWeb);
                } catch (Exception e) {
                        emitter.onError(e);
                }
        });
        thread.start();
});

// cache the result of the single, so that the web query is only done once
Single<List<Todo>> cachedSingle = todosSingle.cache();

cachedSingle.subscribe(... " Show todos times in a bar chart " ...);

showTodosInATable(cachedSingle);

cachedSingle.subscribe(... " Show todos in gant diagram " ...);

anotherMethodThatsSupposedToSubscribeTheSameSingle(cachedSingle);

9. Flowable<T> and Backpressure

RxJava 2.0 introduced a new type Flowable<T>, which is pretty much the same as an Observable<T> in regards of the API, but Flowable<T> supports backpressure and Observable<T> does not.

Back in RxJava 1.0 the concept of backpressure came too late and was added to Observable<T> types, but some did throw a MissingBackpressureException, so distinguishing between Flowable<T> and Observable<T> is a good thing.

Besides Observable<T> also Maybe<T>, Single<T> and Completable<T> have no backpressure.

10. Conversion between types

It is easy to convert between different RxJava types.

Table 1. Conversion between types
From / To Flowable Observable Maybe Single Completable

Flowable

toObservable()

reduce()
elementAt()
firstElement()
lastElement()
singleElement()

scan()
elementAt()
first()/firstOrError()
last()/lastOrError()
single()/singleOrError()
all()/any()/count()
(and more…​)

ignoreElements()

Observable

toFlowable()

reduce()
elementAt()
firstElement()
lastElement()
singleElement()

scan()
elementAt()
first()/firstOrError()
last()/lastOrError()
single()/singleOrError()
all()/any()/count()
(and more…​)

ignoreElements()

Maybe

toFlowable()

toObservable()

toSingle()
sequenceEqual()

toCompletable()

Single

toFlowable()

toObservable()

toMaybe()

toCompletable()

Completable

toFlowable()

toObservable()

toMaybe()

toSingle()
toSingleDefault()

11. Testing RxJava Observables and Subscriptions

11.1. Testing the observables

You can test an observable via the TestSubscriber class provided by the RxJava library.

Observable<String> obs = ...// assume creation code here
TestSubscriber<String> testSubscriber = new TestSubscriber<>();
obs.subscribe(testSubscriber);

testSubscriber.assertNoErrors();
List<String> chickens = testSubscriber.getOnNextEvents();

// TODO assert your string integrity...

11.2. Testing the observables

RxJava provides a way to override the schedulers exposed, so that observables are called synchronously. See http://fedepaol.github.io/blog/2015/09/13/testing-rxjava-observables-subscriptions/ for an example.

12. About this website

13. RxJava resources

13.1. vogella GmbH training and consulting support

TRAINING SERVICE & SUPPORT

The vogella company provides comprehensive training and education services from experts in the areas of Eclipse RCP, Android, Git, Java, Gradle and Spring. We offer both public and inhouse training. Whichever course you decide to take, you are guaranteed to experience what many before you refer to as “The best IT class I have ever attended”.

The vogella company offers expert consulting services, development support and coaching. Our customers range from Fortune 100 corporations to individual developers.

Copyright © 2012-2016 vogella GmbH. Free use of the software examples is granted under the terms of the EPL License. This tutorial is published under the Creative Commons Attribution-NonCommercial-ShareAlike 3.0 Germany license.

See Licence.