Grokay * RxJava, Part One: The Basics

* from the translator: I thought for a long time how to translate the verb “to grok” into Russian. On the one hand, this word translates as “understand” or “realize”, and on the other hand, when translating Robert Heinlein’s novel “A Stranger in a Foreign Country” (in which this word was first born), translators made it Russian “ grokat ". I did not read the novel, so I considered that this word had some semantic shades that were not transmitted by Russian analogues, and therefore in its translation used the same tracing from English.

RxJava is, now, one of the hottest topics for discussion with Android programmers. The only problem is that it is quite difficult to understand the very basics of it, if you have not encountered anything like it. Functional reactive programming is quite difficult to understand, if you come from the imperative world, but as soon as you deal with it, you will understand how cool it is!
I will try to give you some general idea of ​​RxJava. The goal of this series of articles is not to explain everything down to the last comma (I could hardly do it), but rather to interest you in RxJava and how it works.

The basics

The basic building blocks of the reactive code are Observables and Subscribers ¹ . Observable is the data source, and Subscriber is the consumer.
Observable data generation always occurs in accordance with the same order of actions: Observable "radiates" a certain amount of data (including Observable may not emit anything), and completes its work, either successfully or with an error. For each Subscriber subscribed to Observable , the Subscriber.onNext() method is called for each element of the data stream, after which Subscriber.onComplete() and Subscriber.onError() can be called.
All this is very similar to the usual “Observer” pattern , but there is one important difference: Observables often do not begin to generate data until someone explicitly subscribes to them ² . In other words: if a tree falls and nobody is near, then the sound of its fall is not audible .

Hello world!

Let's deal with a small example. First, create a simple Observable :
')

 Observable<String> myObservable = Observable.create( new Observable.OnSubscribe<String>() { @Override public void call(Subscriber<? super String> sub) { sub.onNext("Hello, world!"); sub.onCompleted(); } } ); 

Our Observable spawns the string “Hello, world!”, And ends its work. Now we will create a Subscriber to accept the data and do something with it.

 Subscriber<String> mySubscriber = new Subscriber<String>() { @Override public void onNext(String s) { System.out.println(s); } @Override public void onCompleted() { } @Override public void onError(Throwable e) { } }; 

All that Subscriber does is print the lines passed to it by Observable . Now that we have myObservable and mySubscriber , we can link them together using the subscribe() method:

 myObservable.subscribe(mySubscriber); //  "Hello, world!" 

As soon as we signed mySubscriber to myObservable , myObservable calls the mySubscriber onNext() and onComplete() methods of onNext() , and as a result, mySubscriber outputs “Hello, world!” To the console and finishes its execution.

Simplify the code

Generally speaking, we wrote too much code for such a simple task as outputting “Hello, world!” To the console. I specifically wrote this code so that you could easily figure out what was happening. There are many more rational ways to solve a similar problem in RxJava.
First, let's simplify our Observable . In RxJava, there are methods for creating Observable , suitable for solving the most typical tasks. In our case, Observable.just() spawns one data element, and then completes its execution, just like our first version ³ :

 Observable<String> myObservable = Observable.just("Hello, world!"); 

Next, let's simplify our Subscriber . We are not interested in the onCompleted() and onError() methods, so we can use another base class to determine what to do in onNext() :

 Action1<String> onNextAction = new Action1<String>() { @Override public void call(String s) { System.out.println(s); } }; 

Action can be used to replace any part of Subscriber : Observable.subscribe() can accept one, two, or three Action parameters that will be executed instead of onNext() , onError() and onCompete() . That is, we can replace our Subscriber like this:

 myObservable.subscribe(onNextAction, onErrorAction, onCompleteAction); 

But, since we do not need onError() and onCompete() , we can simplify the code even more:

 myObservable.subscribe(onNextAction); //  "Hello, world!" 

Now let's get rid of the variables by resorting to the chained method call:

 Observable.just("Hello, world!") .subscribe(new Action1<String>() { @Override public void call(String s) { System.out.println(s); } }); 

And finally, we can use lambdas from Java 8 to simplify the code even more:

 Observable.just("Hello, world!") .subscribe(s -> System.out.println(s)); 

If you are writing under Android (and therefore cannot use Java 8), I highly recommend retrolambda , which will help simplify the very verbose code in some places.

Transformation

Let's try something new.
For example, I want to add my signature to the “Hello, world!” Output to the console. How to do it? First, we can change our Observable :

 Observable.just("Hello, world! -Dan") .subscribe(s -> System.out.println(s)); 

This may work if you have access to the source code in which your Observable is defined, but this will not always be the case - for example, when you use someone’s library. Another problem: what if we use our Observable in many places, but we want to add a signature only in some cases ?
You can try to rewrite Subscriber :

 Observable.just("Hello, world!") .subscribe(s -> System.out.println(s + " -Dan")); 

This option is also unsuitable, but for other reasons: I want my subscribers to be as lightweight as possible, since I can run them in the main thread. At a more conceptual level, subscribers must respond to the data arriving at them, not change them.
It would be great if you could change “Hello, world!” At some intermediate step.

Introduction to Operators

And such an intermediate step, designed to transform the data, is. Names are operators, and they can be used between the Observable and Subscriber for data manipulation. There are a lot of operators in RxJava, so for a start it will be better to focus on only a few.
For our particular situation, the map() operator would be best suited for converting one data element to another:

 Observable.just("Hello, world!") .map(new Func1<String, String>() { @Override public String call(String s) { return s + " -Dan"; } }) .subscribe(s -> System.out.println(s)); 

And again, you can resort to lambda:

 Observable.just("Hello, world!") .map(s -> s + " -Dan") .subscribe(s -> System.out.println(s)); 

Cool, yeah? Our map() operator, roughly speaking, is Observable , which transforms a data item into it. We can create a chain of as many map() as we see fit to give the data the most convenient and simple form in order to facilitate the task to our Subscriber .

More about map ()

An interesting property of map() is that it does not have to generate data of the same type as the original Observable .
Suppose that our Subscriber should display not a generated text, but its hash:

 Observable.just("Hello, world!") .map(new Func1<String, Integer>() { @Override public Integer call(String s) { return s.hashCode(); } }) .subscribe(i -> System.out.println(Integer.toString(i))); 

Interesting: we started with strings, and our Subscriber accepts Integer. By the way, we again forgot about lambdas:

 Observable.just("Hello, world!") .map(s -> s.hashCode()) .subscribe(i -> System.out.println(Integer.toString(i))); 

As I said earlier, we want our Subscriber do as little work as possible, so let's apply another map() to convert our hash back to String :

 Observable.just("Hello, world!") .map(s -> s.hashCode()) .map(i -> Integer.toString(i)) .subscribe(s -> System.out.println(s)); 

Take a look at this - our Observable and Subscriber now look the same as at the very beginning! We just added a few intermediate steps that transform our data. We could even add the code again, adding my signature to the generated lines:

 Observable.just("Hello, world!") .map(s -> s + " -Dan") .map(s -> s.hashCode()) .map(i -> Integer.toString(i)) .subscribe(s -> System.out.println(s)); 

So, what is next?

Now you probably think: “Well, as usual: they show a simple example, and they say that the technology is cool, because it allows you to solve this problem in two lines of code”. I agree, an example is really simple. But from it you can take a couple of useful ideas:

Idea # 1: Observable and Subscriber can do anything

Do not limit your imagination; everything you want is possible.
Your Observable can be a database query, and Subscriber can display the results of a query. Observable can also be a click on the screen, Subscriber can contain a reaction to this click. Observable can be a stream of bytes received from the network, while a Subscriber can write this data to a storage device.
This is a general-purpose framework that can handle almost any problem.

Idea # 2: Observable and Subscriber do not depend on intermediate steps between them

You can insert as many map() calls as you like between the Observable and the Subscriber subscribed to it. The system is easily composable , and with its help it is very easy to manage the flow of data. If the operators work with the correct input / output data, you can write a chain of transformations of infinite length ⁴ .

Take a look at these key ideas together and you will see a system with great potential. Now, however, we have only one map() operator, and you cannot write much with it. In the second part of this article, we will look at the large number of operators available to you out of the box when you use RxJava.

Go to the second part.

---

¹ Subscriber implements the Observer interface, and therefore the “basic building block” can be called the last, but in practice you will most often use Subscriber , because it has several additional useful methods, including Subscriber.unsubscribe() .
² RxJava has “hot” and “cold” Observables . The hot Observable spawns data all the time, even if no one subscribes to it. Cold Observable , respectively, generates data only if it has at least one subscriber (it is the cold Observables that are used in the article). For the initial stages of learning RxJava, this difference is not so important.
³ Strictly speaking, Observable.just() not a complete analog of our original code, but I will explain why this happens only in the third part of the article.
⁴ Okay, not so infinite, because at some point I will rest on the restrictions imposed by iron, but you understand what I wanted to say.