Replace callbacks with observables from RxJava

Question

Im using listeners as callbacks to observe asynchronous operations with Android, but I think that could be great replacing this listeners with RxJava, Im new using this library but I really like it and Im always using it with Android projects.

Here is my code to refactor:

public void getData( final OnResponseListener listener ){    if(data!=null && !data.isEmpty()){        listener.onSuccess();    }    else{        listener.onError();    } } 

A simple callback:

public interface OnResponseListener {    public void onSuccess();    public void onError();  } 

And the "observer":

object.getData( new OnResponseListener() {     @Override     public void onSuccess() {        Log.w(TAG," on success");     }      @Override     public void onError() {        Log.e(TAG," on error");     } }); 

Thanks!

Answer 1

For example you can use Observable.fromCallable to create observable with your data.

public Observable<Data> getData(){     return Observable.fromCallable(() -> {         Data result = null;         //do something, get your Data object         return result;     }); } 

then use your data

 getData().subscribeOn(Schedulers.io())             .observeOn(AndroidSchedulers.mainThread())             .subscribe(data -> {                 //do something with your data             }, error -> {                 //do something on error             }); 

Used rxjava 1.x and lambda expressions.

edit:

if I understand you well, you wanted to replace that listener, not wrap it into observable. I added other example in reference to your comment. Oh.. also you should use Single if you are expecting only one item.

public Single<Data> getData() {         return Single.create(singleSubscriber -> {             Data result = object.getData();             if(result == null){                 singleSubscriber.onError(new Exception("no data"));             } else {                 singleSubscriber.onSuccess(result);             }         });     }  getData().subscribeOn(Schedulers.io())             .observeOn(AndroidSchedulers.mainThread())             .subscribe(data -> {                 //do something with your data             }, error -> {                 //do something on error             }); 

Answer 2

You are looking for Completable.create:

Completable: Represents a deferred computation without any value but only indication for completion or exception. The class follows a similar event pattern as Reactive-Streams: onSubscribe (onError|onComplete)?

Completable.create(subscriber -> {     object.getData(new OnResponseListener() {         @Override         public void onSuccess() {            subscriber.onCompleted();         }          @Override         public void onError() {            subscriber.onError(* put appropriate Throwable here *);         }     } }) ...//apply Schedulers .subscribe((() -> *success*), (throwable -> *error*));