Getting Started with RxJava

In this blog, we will learn about the basics of RxJava and its building blocks. RxJava is the open-source implementation of ReactiveX in Java.

What is RX?

RX stands for the Reactive Extension which is used for creating the asynchronous and event-based programs. It is a powerful tool that provides the various benefits. Some of the benefits are it is Composable (RX operators can be combined to produce more complicated operations) and Transformative (RX operator can transform one type of data into another) in nature and even make Error Handling easy.  The basic three constructs of RXJava are:-
1. Observable
2. Subscriber
3. Operator

Observable
An object that emits a stream of data or events. It starts providing data once the subscriber subscribes to it. It is basically of two type:- Blocking and NonBlocking

Subscriber
An object that acts upon the data emitted by Observable. Whenever the Observable emits a new item, the onNext() method is invoked and if the Observable stops emitting the value, the onComplete() method is invoked. And on emitting items, if any error encounters, the onError() method is invoked.

Operator
It helps to transform the data send by the Observable. And there are lots of operators that can be found here in alphabetical order. These operators operate on Observable and simply return Observable and that’s why it can be applied in chaining like one after the other.

Let’s implement some basic example.

Example 1: Creating an Observable using just() operator() and then subscribing it.

xxxxxxxxxx

//Creating an Observable of Integer Type.

Observable observable = Observable.just(1,2,3,4,5);

//Subscribing observable

observable.subscribe(

(value) -> System.out.println("Value is "+ value ),

(error) -> System.out.println("oops! Something went wrong " + error.getMessage()),

() -> System.out.println("The Observable is completed")

);

Output:

xxxxxxxxxx

Value is 1

Value is 2

Value is 3

Value is 4

Value is 5

The Observable is completed

Example 2: Creating an Observable from an existing data structure using from() operator and then subscribing it.

xxxxxxxxxx

List numbers = Arrays.asList(1,2,3,4);

//Create an Observable from existing List

Observable observable1 = Observable.from(numbers);

observable1.subscribe(

(value) -> System.out.println("Value is "+ value ),

(error) -> System.out.println("oops! Something went wrong " + ((Throwable)error).getMessage()),

() -> System.out.println("The Observable is completed")

);

Output:

xxxxxxxxxx

Value is 1

Value is 2

Value is 3

Value is 4

The Observable is completed

Example 3: Creating an Observable and transforming it using different operators and then subscribing to it.

xxxxxxxxxx

Observable.just(1,2,3,4)    // Creating an Observable

        .map(value -> value + 1) //Incrementing the value emitted by Observable

        .filter(value -> value % 2 == 0) //Filtering the even numbers

        .count() //Computing the length of Observable

        .subscribe(

        (value) -> System.out.println("length of Observable is "+ value ),

        (error) -> System.out.println("Something went wrong " + error.getMessage()),

        () -> System.out.println("The Observable is completed")

);

Output:

xxxxxxxxxx

length of Observable is 2

The Observable is completed

Conclusion:
So this is pretty straightforward, we create Observable which emits data and it is activated when we will subscribe to it. And if we want to transform the Observable we can use operators.

Hope this is helpful and give you the basic understanding Of RxJava and its basic constructs. Please feel free to provide your suggestions

References:
https://instil.co/2014/08/05/rxjava-in-different-flavours-of-java/
https://github.com/ReactiveX/RxJava/wiki/Alphabetical-List-of-Observable-Operators