Advantages of Using RXJava Over Java 8 Stream

Java 8 stream has made my programming life simpler as a software engineer. However there are a lot more that can be improved by incorporating the RXJava library. RXJava contains tons of functionalities to process streams or observables.

In Java 9 there is a similar concept called

Flow

In this article I will highlight a subset of RXJava features that in my opinion will be most useful in my codes. I will not cover all the functionalities that RXJava has because that will be too long. For more coverage in RXJava functionalities, please read [1].

Easily Combine Two Streams with Zip Operator

Sometimes there are cases that you wanted to iterate together between 2 Collections. Say you have two lists of integers and you want to "combine" each element with the same index in list 1 with each element in list 2. Then you would probably think code like this.

List<Integer> list1 = IntStream.range(0, 10)
        .boxed()
        .collect(Collectors.toList());
List<Integer> list2 = IntStream.range(1, 11)
        .boxed()
        .collect(Collectors.toList());

for (int i = 0; i < list1.size(); i++) {
  System.out.println(list1.get(i) + list2.get(i));
}

In python, there is an operator called

zip

zip

Observable.zip(
            Observable.range(0, 10), 
            Observable.range(1, 11), 
            (x, y) -> x + y)
        .subscribe(System.out::println);

Easy Reuse Stream / Observable

Java stream is designed to be used one time. If you’re wondering why there’s a great explanation of that at stackoverflow [2], but I will not get into that here. 

Now, consider this example

Stream<Integer> intStream = IntStream.range(0, 10).boxed();
intStream.forEach(System.out::println);
intStream.forEach(System.out::println);

This code will print integer from 0 to 10, but then will the following exception.

java.lang.IllegalStateException: stream has already been operated upon or closed

Java 8 stream is not meant to be used more than once. The reason behind this behavior is, in my understanding, a unification behavior between iterator for non-one-shot source and one-shot source [2]. In one-shot source, like reading lines from a file, it is hard to determine the case of if we iterate this twice what should happen?

With Java 8 stream, you have to re-initiated the stream to make this works.

Stream<Integer> intStream = IntStream.range(0, 10).boxed();
intStream.forEach(System.out::println);
intStream = IntStream.range(0, 10).boxed();
intStream.forEach(System.out::println);

RXJava programming concept assumption is different from Java 8 stream, thus allowing us to re-use the stream in

Collection

Observable<Integer> intObservable =  Observable.range(0, 10);
intObservable.subscribe(System.out::println);
intObservable.subscribe(System.out::println);

Easy Error Handling

Consider another example. You wanted to do division operation and use integer from -4 to 5 as the denominator.

IntStream.range(-4, 5)
        .map(i -> 1 / i)
        .forEach(System.out::println);

you will notice that there's a case where in the middle of the stream this will calculate

1/0

java.lang.ArithmeticException: / by zero

Now you have two choices

1. Change the stream value to manually exclude 0. you can use

   filter(i != 0)

   stream operation for this.
2. Change the

   map

   function, maybe to use

   try... catch...

   . Okay, I will be honest with you, this option is sucks.

In this case you can easily go with option 1, but not in real-world cases. In real-world cases you would very much likely go with option 2. You do not even know what are the values that caused the error. Or if you know the values causing the error you need to consider how many values you need to hardcode into the filter operation. Maybe it is just easier for you to place an error handling function.

RXJava, fortunately, will help you to handle that error case without writing a try-catch block! Here are some ways you can handle unwanted value causing an error, without

filter

Add On Error Subscription

subscribe()

Observable.range(-4, 5)
        .map(i -> 1 / i)
        .subscribe(
                System.out::println,
                t -> System.out.println("some error happened"));

This block of code will print "some error happened" because it will still try to execute

1/0

Substitute Value When Error

Use

onErrorReturnItem()

  Observable.range(-4, 5)
        .map(i -> 1 / i)
        .onErrorReturnItem(-1)
        .subscribe(
                System.out::println, 
                t -> System.out.println("some error happened"));

You need a little bit more careful when using

onErrorReturnItem()

-1

Conclusion

RXJava has many ways to simplify your stream codes, but that's not the only function that it has. This article has not yet discussed other exciting functionalities of RXJava, such as how to handles "hot" or "cold" observables, how to handle backpressure, or how should we parallelize the operation. I encourage readers to explore more about the reactive concept and dig into the RXJava library to experience and thinking about making your applications to be reactive.

References

[1] Samoylov, N., Nield, T. 2020. Learning RxJava 2nd Edition.

[2] https://stackoverflow.com/questions/28459498/why-are-java-streams-once-off/28513908#28513908

[3] Java 9 Reactive Streams. https://www.baeldung.com/java-9-reactive-streams