Converting Synchronous and Asynchronous APIs to Observables using RxJava2

1. Overview

In this tutorial, we’re going to learn how to transform traditional synchronous and asynchronous APIs into Observables using RxJava2 operators.

We’ll create a few simple functions that will help us discuss these operators in detail.

2. Maven Dependencies

First, we have to add RxJava2 and RxJava2Extensions as Maven dependencies:

<dependency>
    <groupId>io.reactivex.rxjava2</groupId>
    <artifactId>rxjava</artifactId>
    <version>2.2.2</version>
</dependency>
<dependency>
    <groupId>com.github.akarnokd</groupId>
    <artifactId>rxjava2-extensions</artifactId>
    <version>0.20.4</version>
</dependency>

3. The Operators

RxJava2 defines a whole lot of operators for various use cases of reactive programming.

But we’ll be discussing only a few operators that are commonly used for converting synchronous or asynchronous methods into Observables based on their nature. These operators take functions as arguments and emit the value returned from that function.

Along with the normal operators, RxJava2 defines a few more operators for extended functionalities.

Let’s explore how we can make use of these operators to convert synchronous and asynchronous methods.

4. Synchronous Method Conversion

4.1. Using  fromCallable()

This operator returns an Observable that, when a subscriber subscribes to it, invokes the function passed as the argument and then emits the value returned from that function. Let’s create a function that returns an integer and transform it:

AtomicInteger counter = new AtomicInteger();
Callable<Integer> callable = () -> counter.incrementAndGet();

Now, let’s transform it into an Observable and test it by subscribing to it:

Observable<Integer> source = Observable.fromCallable(callable);

for (int i = 1; i < 5; i++) {
    source.test()
      .awaitDone(5, TimeUnit.SECONDS)
      .assertResult(i);
    assertEquals(i, counter.get());
}

The fromCallable() operator executes the specified function lazily each time when the wrapped Observable gets subscribed. To test this behavior, we’ve created multiple subscribers using a loop.

Since reactive streams are asynchronous by default, the subscriber will return immediately. In most of the practical scenarios, the callable function will have some kind of delay to complete its execution. So, we’ve added a maximum wait time of five seconds before testing the result of our callable function.

Note also that we’ve used Observable‘s test() method. This method is handy when testing Observables. It creates a TestObserver and subscribes to our Observable.

4.2. Using start()

The start() operator is part of the RxJava2Extension module. It will invoke the specified function asynchronously and returns an Observable that emits the result:

Observable<Integer> source = AsyncObservable.start(callable);

for (int i = 1; i < 5; i++) {
    source.test()
      .awaitDone(5, TimeUnit.SECONDS)
      .assertResult(1);
    assertEquals(1, counter.get());
}

The function is called immediately, not whenever a subscriber subscribes to the resulting Observable. Multiple subscriptions to this observable observe the same return value.

5. Asynchronous Method Conversion

5.1. Using  fromFuture()

As we know, the most common way of creating an asynchronous method in Java is using the Future implementation. The fromFuture method takes a Future as its argument and emits the value obtained from the Future.get() method.

First, let’s make the function which we’ve created earlier asynchronous:

ExecutorService executor = Executors.newSingleThreadExecutor();
Future<Integer> future = executor.submit(callable);

Next, let’s do the testing by transforming it:

Observable<Integer> source = Observable.fromFuture(future);

for (int i = 1; i < 5; i++) {
    source.test()
      .awaitDone(5, TimeUnit.SECONDS)
      .assertResult(1);
    assertEquals(1, counter.get());
}
executor.shutdown();

And notice once again that each subscription observes the same return value.

Now, the dispose() method of Observable is really helpful when it comes to memory leak prevention. But in this case, it will not cancel the future because of the blocking nature of Future.get().

So, we can make sure to cancel the future by combining the doOnDispose() function of the source observable and the cancel method on future:

source.doOnDispose(() -> future.cancel(true));

5.2. Using startFuture()

As the name depicts, this operator will start the specified Future immediately and emits the return value when a subscriber subscribes it. Unlike the fromFuture operator which caches the result for next use, this operator will execute the asynchronous method each time when it gets subscribed:

ExecutorService executor = Executors.newSingleThreadExecutor();
Observable<Integer> source = AsyncObservable.startFuture(() -> executor.submit(callable));

for (int i = 1; i < 5; i++) {
    source.test()
      .awaitDone(5, TimeUnit.SECONDS)
      .assertResult(i);
    assertEquals(i, counter.get());
}
executor.shutdown();

5.3. Using deferFuture()

This operator aggregates multiple Observables returned from a Future method and returns a stream of return values obtained from each Observable. This will start the passed asynchronous factory function whenever a new subscriber subscribes.

So let’s first create the asynchronous factory function:

List<Integer> list = Arrays.asList(new Integer[] { counter.incrementAndGet(), 
  counter.incrementAndGet(), counter.incrementAndGet() });
ExecutorService exec = Executors.newSingleThreadExecutor();
Callable<Observable<Integer>> callable = () -> Observable.fromIterable(list);

And then we can do a quick test:

Observable<Integer> source = AsyncObservable.deferFuture(() -> exec.submit(callable));
for (int i = 1; i < 4; i++) {
    source.test()
      .awaitDone(5, TimeUnit.SECONDS)
      .assertResult(1,2,3);
}
exec.shutdown();

6. Conclusion

In this tutorial, we’ve learned how to transform synchronous and the asynchronous methods to RxJava2 observables.

Of course, the examples we’ve shown here are the basic implementations. But we can use RxJava2 for more complex applications like video streaming and applications where we need to send large amounts of data in portions.