RxJava, side-effects, and methods that return void

Question

I have written a repository using RxJava and am having trouble figuring out the best way to handle writes and deletes to the repo.

When writing this, my goal was to make sure that any write or delete methods would be able to be called from RxJava side-effect methods. This means that the write / delete methods have to run synchronously and don't return an Observable. This way I could do things like this:

repository
    .fetchData()
    .doOnNext(s -> if (s.equals("something") {
                       repository.writeData(s);
                   }
    )
    .subscribe();

This is a trivial example, but it shows that I call a repository method that returns an Observable, subscribe to that Observable and when the Observable's onNext() method gets called I can perform my side-effect which is a synchronous write to a memory cache or database.

Ok, so that example works great, but now I have an issue where I only need to write or delete data from my memory cache or database. That is, I want to use RxJava to run synchronous code where the method returns void.

The way my Repository's API is written I would want to do something like call repository.deleteData(), however I want to do this off the main thread. One way to approach this would be to ditch RxJava and just create a thread to execute the task on. This would work, but I do everything else with RxJava so I'd like to use it here as well.

I found 2 potential solutions but am not sure how correct they are:

1) Use Observable.just(null) to kick off an observable and then use the doOnNext() side-effect method to call my repository's method from. This works it just seems hacky but it lets me use RxJava and lets me specify the threads to subscribe on / observe on which makes threading easy and allows the repository's API to stay the same so that it still returns void.

2) I can also do something like Observable.defer() and then from defer's call method I can return Observable.just(repository.deleteData()). If I do it this way I do have to change the repository's API so that the deleteData method has a Void return type instead of void (and also the method has return null; added to fulfill the contract of the method signature). This also allows me to use RxJava, specify which threads I want to observe and subscribe on and keep using an Observable even though what I'm really doing is a side-effect.

Is there a way for RxJava to create an Observable which calls a method with a void return type which runs synchronous code outside of a side-effect method?

This link may help you. Consider the difference between rxjava and rxjava2, I think your first solution is the common usage. — Dean Xu
so you want to construct Observables from your synchronous methods? — yosriz

Tassos Bassoukos Tassos Bassoukos · Accepted Answer · 2017-08-10T09:36:33

Well, why not something like this?

public static <T> Observable<T> observeCompletion(Runnable r) {
   return Observable.defer(() -> {
         r.run();
         return Observable.empty();
   };
}

Now you can do the thread hopping that you require via onSubscribe, and you keep everything in Rx.

RxJava, side-effects, and methods that return void

1 Answers