Differences in Creating Observables in RxJava

There are many ways to create observables in RxJava. The most popular creation mechanism are factories from the Observable class. Today, we going to focus on:  Observable.just  and  Observable.fromCallable  and see the differences between those factories. 

 Observable.just(value)  is the easiest way to create an observable. Observable.just  accepts any object (value) to be emitted, and it will immediately fetch the value before emitting it. It also blocks when evaluating the value. Let us check out an example:

Say we have a method that generates random UUID — let's mimic it as a time-consuming operation:

  public static String getRandomString(int seconds) {

        try {
            TimeUnit.SECONDS.sleep(seconds);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("getRandomString on " + Thread.currentThread().getName() + " finished " + LocalDateTime.now());
        return UUID.randomUUID().toString();

    }

Next, let's create an Observable that emits a string from the above method:

public static void main(String [] args) {

  System.out.println("going to create observable "+LocalDateTime.now());
  Observable<String> myObservable = Observable.just(getRandomString(3));
  System.out.println("After creating observable "+LocalDateTime.now());
}

After running the code snippet above, the output is:

going to create observable 2018-09-15T14:47:09.710168500
getRandomString on main finished 2018-09-15T14:47:12.730101300
After creating observable 2018-09-15T14:47:13.043256100

We can immediately see that Observable.just  is executing the getRandomString  immediately and is blocking call. Why? Because the output of line five is approximately three seconds after the output of line three — the time took to generate a random string.  So, we can conclude that Observable.just is immediate and blocking. 

 Observable.just is executed in the thread in which it has been called and cannot be switched to another thread.

If we try to switch emission on the different thread using the  subscribeOn operator like this:

Observable<String> myObservable = Observable.just(getRandomString(3)).subscribeOn(Schedulers.computation());

We will still get outputs as before, meaning that the just operator is blocking one. So, we will use the Observable.just for converting the fast and simple operation to Observable or need to convert a value to Observable — the blocking way

The other way one can create an Observable in a non-blocking way is using a Observable.fromCallable that accepts a Supplier<T>. Also, the execution of the given supplier is not executed until subscription happens, so it also not immediately:

System.out.println("going to create observable "+LocalDateTime.now());
Observable<String> myObservable = Observable.fromCallable(()-> getRandomString(3)).subscribeOn(Schedulers.computation());
System.out.println("After creating observable "+LocalDateTime.now());

The output of the above code is :

going to create observable 2018-09-16T06:41:09.559444100
After creating observable 2018-09-16T06:41:09.660174600

First, we see that the  Obserable.fromCallable is non-blocking since the output of line three is approximately at the same time as the output of line one. That is because the supplier inside Observable.fromCallable has never been called since no observer has been subscribed to it (as opposed to  Observable.just).

Now, let an observer subscribe to our observable:

 System.out.println("going to create observable "+LocalDateTime.now());
 Observable<String> myObservable = Observable.fromCallable(()-> getRandomString(3)).subscribeOn(Schedulers.computation());
 myObservable.subscribe(x->System.out.println("got "+x+" at time : "+LocalDateTime.now()));
 System.out.println("After creating observable "+LocalDateTime.now());

Again the output of the above code is:

going to create observable 2018-09-16T06:51:36.211351700
After creating observable 2018-09-16T06:51:36.340007900
getRandomString on RxComputationThreadPool-1 finished 2018-09-16T06:51:39.341881700
got 43848c7f-9343-45e5-b451-04486476802f at time : 2018-09-16T06:51:39.563289300

We see that the output of lines one and four are at the same time, and the end of the getRandomString method happened three seconds later after the line one execution in a different thread (line three printed output) So,  Observable.fromCallable can evaluate its object of emission in a different thread. 

Use Observable.fromCallable when you want fresh data for each subscription, because the supplier will be triggered for each new observer in a different thread if you need to, as opposed to Observable.just that acts like a caching source.