When should one use RxJava Observable and when simple Callback on Android?

I'm working on networking for my app. So I decided to try out Square's Retrofit. I see that they support simple Callback

@GET("/user/{id}/photo")
void getUserPhoto(@Path("id") int id, Callback<Photo> cb);

and RxJava's Observable

@GET("/user/{id}/photo")
Observable<Photo> getUserPhoto(@Path("id") int id);

Both look pretty similar at first glance, but when it gets to implementation it gets interesting...

While with simple callback implementation would look similar to this:

api.getUserPhoto(photoId, new Callback<Photo>() {
    @Override
    public void onSuccess() {
    }
});

which is quite simple and straightforward. And with Observable it quickly gets verbose and quite complicated.

public Observable<Photo> getUserPhoto(final int photoId) {
    return Observable.create(new Observable.OnSubscribeFunc<Photo>() {
        @Override
        public Subscription onSubscribe(Observer<? super Photo> observer) {
            try {
                observer.onNext(api.getUserPhoto(photoId));
                observer.onCompleted();
            } catch (Exception e) {
                observer.onError(e);
            }

            return Subscriptions.empty();
        }
    }).subscribeOn(Schedulers.threadPoolForIO());
}

And that is not it. You still have to do something like this:

Observable.from(photoIdArray)
        .mapMany(new Func1<String, Observable<Photo>>() {
            @Override
            public Observable<Photo> call(Integer s) {
                return getUserPhoto(s);
            }
        })
        .subscribeOn(Schedulers.threadPoolForIO())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Action1<Photo>() {
            @Override
            public void call(Photo photo) {
                //save photo?
            }
        });

Am I missing something here? Or is this a wrong case to use Observables? When would/should one prefer Observable over simple Callback?

Update

Using retrofit is much simpler than example above as @Niels showed in his answer or in Jake Wharton's example project U2020. But essentially the question stays the same - when should one use one way or the other?

For simple networking stuff, the advantages of RxJava over Callback is very limited. The simple getUserPhoto example:

RxJava:

api.getUserPhoto(photoId)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Action1<Photo>() {
            @Override
            public void call(Photo photo) {
               // do some stuff with your photo 
            }
     });

Callback:

api.getUserPhoto(photoId, new Callback<Photo>() {
    @Override
    public void onSuccess(Photo photo, Response response) {
    }
});

The RxJava variant is not much better than the Callback variant. For now, let's ignore the error handling. Let's take a list of photos:

RxJava:

api.getUserPhotos(userId)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.flatMap(new Func1<List<Photo>, Observable<Photo>>() {
    @Override
    public Observable<Photo> call(List<Photo> photos) {
         return Observable.from(photos);
    }
})
.filter(new Func1<Photo, Boolean>() {
    @Override
    public Boolean call(Photo photo) {
         return photo.isPNG();
    }
})
.subscribe(
    new Action1<Photo>() {
    @Override
        public void call(Photo photo) {
            list.add(photo)
        }
    });

Callback:

api.getUserPhotos(userId, new Callback<List<Photo>>() {
    @Override
    public void onSuccess(List<Photo> photos, Response response) {
    	List<Photo> filteredPhotos = new ArrayList<Photo>();
    	for(Photo photo: photos) {
    		if(photo.isPNG()) {
    			filteredList.add(photo);
    		}
    	}
    }
});

Now, the RxJava variant still isn't smaller, although with Lambdas it would be getter closer to the Callback variant. Furthermore, if you have access to the JSON feed, it would be kind of weird to retrieve all photos when you're only displaying the PNGs. Just adjust the feed to it only displays PNGs.

First conclusion

It doesn't make your codebase smaller when you're loading a simple JSON that you prepared to be in the right format.

Now, let's make things a bit more interesting. Let's say you not only want to retrieve the userPhoto, but you have an Instagram-clone, and you want to retrieve 2 JSONs:

1. getUserDetails()
2. getUserPhotos()

You want to load these two JSONs in parallel, and when both are loaded, the page should be displayed. The callback variant will become a bit more difficult: you have to create 2 callbacks, store the data in the activity, and if all the data is loaded, display the page:

Callback:

api.getUserDetails(userId, new Callback<UserDetails>() {
    @Override
    public void onSuccess(UserDetails details, Response response) {
    	this.details = details;
    	if(this.photos != null) {
    		displayPage();
    	}
    }
});

api.getUserPhotos(userId, new Callback<List<Photo>>() {
    @Override
    public void onSuccess(List<Photo> photos, Response response) {
    	this.photos = photos;
    	if(this.details != null) {
    		displayPage();
    	}
    }
});

RxJava:

private class Combined {
	UserDetails details;
	List<Photo> photos;
}


Observable.zip(api.getUserDetails(userId), api.getUserPhotos(userId), new Func2<UserDetails, List<Photo>, Combined>() {
            @Override
            public Combined call(UserDetails details, List<Photo> photos) {
                Combined r = new Combined();
                r.details = details;
                r.photos = photos;
                return r;
            }
        }).subscribe(new Action1<Combined>() {
            @Override
            public void call(Combined combined) {
            }
        });

We are getting somewhere! The code of RxJava is now as big as the callback option. The RxJava code is more robust; Think of what would happen if we needed a third JSON to be loaded (like the latest Videos)? The RxJava would only need a tiny adjustment, while the Callback variant needs to be adjusted in multiple places (on each callback we need to check if all data is retrieved).

Another example; we want to create an autocomplete field, which loads data using Retrofit. We don't want to do a webcall every time an EditText has a TextChangedEvent. When typing fast, only the last element should trigger the call. On RxJava we can use the debounce operator:

inputObservable.debounce(1, TimeUnit.SECONDS).subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
            	// use Retrofit to create autocompletedata
            }
        });

I won't create the Callback variant but you will understand this is much more work.

Conclusion: RxJava is exceptionally good when data is sent as a stream. The Retrofit Observable pushes all elements on the stream at the same time. This isn't particularly useful in itself compared to Callback. But when there are multiple elements pushed on the stream and different times, and you need to do timing-related stuff, RxJava makes the code a lot more maintainable.

The Observable stuff is already done in Retrofit, so the code could be this:

api.getUserPhoto(photoId)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Action1<Photo>() {
         @Override
            public void call(Photo photo) {
                //save photo?
            }
     });

In the case of getUserPhoto() the advantages for RxJava aren't great. But let's take another example when you'll get all the photos for a user, but only when the image is PNG, and you don't have access to the JSON to do the filtering on the serverside.

api.getUserPhotos(userId)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.flatMap(new Func1<List<Photo>, Observable<Photo>>() {
    @Override
    public Observable<Photo> call(List<Photo> photos) {
         return Observable.from(photos);
    }
})
.filter(new Func1<Photo, Boolean>() {
    @Override
    public Boolean call(Photo photo) {
         return photo.isPNG();
    }
})
.subscribe(
    new Action1<Photo>() {
    @Override
        public void call(Photo photo) {
            // on main thread; callback for each photo, add them to a list or something.
            list.add(photo)
        }
    }, 
    new Action1<Throwable>() {
    @Override
        public void call(Throwable throwable) {
            // on main thread; something went wrong
            System.out.println("Error! " + throwable);
        }
    }, 
    new Action0() {
        @Override
        public void call() {
            // on main thread; all photo's loaded, time to show the list or something.
        }
    });

Now the JSON returns a list of Photo's. We'll flatMap them to individual items. By doing so, we'll be able to use the filter method to ignore photos which are not PNG. After that, we'll subscribe, and get a callback for each individual photo, an errorHandler, and a callback when all rows have been completed.

TLDR Point here being; the callback only returns you a callback for succes and failure; the RxJava Observable allows you to do map, reduce, filter and many more stuff.

With rxjava you can do more things with less code.

Let´s assume that you want to implement instant search in your app. With callbacks you have worried about unsubscribing the previous request and subscribe to the new one, handle orientation change yourself... I think it´s a lot of code and too verbose.

With rxjava is very simple.

public class PhotoModel{
  BehaviorSubject<Observable<Photo>> subject = BehaviorSubject.create(...);
  
  public void setUserId(String id){
   subject.onNext(Api.getUserPhoto(photoId));
  }
  
  public Observable<Photo> subscribeToPhoto(){
    return Observable.switchOnNext(subject);
  }
}

if you want to implement instant search you only have to listen for TextChangeListener and call to photoModel.setUserId(EditText.getText());

In onCreate method of Fragment or activity you subscribe to the Observable that returns photoModel.subscribeToPhoto(), it returns an Observable that always emit the items emited by the latest Observable(request).

AndroidObservable.bindFragment(this, photoModel.subscribeToPhoto())
                 .subscribe(new Action1<Photo>(Photo photo){
      //Here you always receive the response of the latest query to the server.
                  });

Also, if PhotoModel is a Singleton, for instance, you don't need to worry about orientation changes, because BehaviorSubject emits the last server response, regardless of when you subscribe.

With this lines of code we have implemented an instant search and handle orientation changes. Do you think that you can implement this with callbacks with less code? I doubt it.

We usually go with the following logic:

1. If it's a simple one-response call, then Callback or Future is better.
2. If it's a call with multiple responses (stream), or when there are complex interaction between different calls (see @Niels' answer), then Observables are better.

By the samples and conclusions in other answers, I think there is no big difference for simple one or two-steps tasks. However, Callback is simple and straightforward. RxJava is more complicated and too big for simple task. There is the third solution by: AbacusUtil. Let me implement above use cases with all three solutions: Callback, RxJava, CompletableFuture(AbacusUtil) with Retrolambda:

Fetch photo from network and save/display on device:

// By Callback
api.getUserPhoto(userId, new Callback<Photo>() {
    @Override
    public void onResponse(Call<Photo> call, Response<Photo> response) {
        save(response.body()); // or update view on UI thread.
    }

    @Override
    public void onFailure(Call<Photo> call, Throwable t) {
        // show error message on UI or do something else.
    }
});

// By RxJava
api.getUserPhoto2(userId) //
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(photo -> {
            save(photo); // or update view on UI thread.
        }, error -> {
            // show error message on UI or do something else.
        });

// By Thread pool executor and CompletableFuture.
TPExecutor.execute(() -> api.getUserPhoto(userId))
        .thenRunOnUI((photo, error) -> {
            if (error != null) {
                // show error message on UI or do something else.
            } else {
                save(photo); // or update view on UI thread.
            }
        });

Load user details and photo in parallel

// By Callback
// ignored because it's little complicated

// By RxJava
Observable.zip(api.getUserDetails2(userId), api.getUserPhoto2(userId), (details, photo) -> Pair.of(details, photo))
        .subscribe(p -> {
            // Do your task.
        });

// By Thread pool executor and CompletableFuture.
TPExecutor.execute(() -> api.getUserDetails(userId))
          .runOnUIAfterBoth(TPExecutor.execute(() -> api.getUserPhoto(userId)), p -> {
    // Do your task
});

I personally prefer to use Rx to get api responses in cases that I have to do filter, map or something like that on the data Or in cases that I have to do another api calls based on previous calls responses

It looks like you're reinventing the wheel, what you're doing is already implemented in retrofit.

For an example, you could look at retrofit's RestAdapterTest.java, where they define an interface with Observable as return type, and then use it.

When you are creating an app for fun, a pet project, or a POC or the 1st prototype you use simple core android/java classes like callbacks, async tasks, loopers, threads, etc. They are simple to use and do not require any 3rd party lib integration. A big library integration just to build a small-time unchangeable project is illogical when something similar can be done right of the bat.

However, these are like a very sharp knife. Using these in a production environment is always cool but also they'll have consequences. It's difficult to write safe concurrent code if you are not well versed with Clean coding and SOLID principles. You'll have to maintain a proper architecture to facilitate future changes and improve team productivity.

On the other hand, concurrency libraries like RxJava, Co-routines etc are tried and tested over a billion times to help write production-ready concurrent code. Now again, it's not that using these libraries you are not writing concurrent code or abstracting away all the concurrency logic away. You still are. But now, it's visible and enforces a clear pattern for writing concurrent code throughout the codebase and more importantly throughout your development team.

This is a major benefit of using a concurrency framework instead of the plain old core classes that deal with raw concurrency. However, don't misunderstand me. I'm a big believer in limiting the dependencies of the external library but in this specific case, you'll have to build a custom framework for your code-base which is a time-consuming task and can be done only after advance experience. Hence, concurrency frameworks are preferred over using plain classes like callbacks, etc.

---

TL'DR

If you are already using RxJava for concurrent coding throughout the code base, simply use RxJava Observable/Flowable. A wise question to ask then is should I use Observables to Flowables. If not then, go ahead and use a callable.