How to synchronise asynchronous methods in RxJava? Async Waterfall in RxJava

After a couple of days of trial and error, the solution I used for this question is as follows. Any suggestions on improvement are most welcome!

private Observable<File> makeObservable(final String filePath){
    return Observable.create(new ObservableOnSubscribe<File>() {
        @Override
        public void subscribe(final ObservableEmitter<File> e) throws Exception {
            // someAsyncOperation will call the Callback.onResult after
            // having finished the asynchronous operation.
            // Callback<File> is an abstract code I put together for this example
            SomeDatabase.someAsyncOperation(filePath, new Callback<File>(){
                @Override
                public void onResult(Error error, File file){
                    if (error != null){
                        e.onError(error);
                    } else {
                        e.onNext(file);
                        e.onComplete();
                    }
                }
            })
        }
    });
}

private void processFilePaths(ArrayList<String> dataArray){
    ArrayList<Observable<File>> observables = new ArrayList<>();
    for (String filePath : dataArray){
        observables.add(makeObservable(filePath));
    }
    Observable.concat(observables)
            .toList()
            .subscribe(new Consumer<List<File>>() {
                @Override
                public void accept(@NonNull List<File> files) throws Exception {
                    // process results.
                    // Files are now sequentially processed using the asynchronous code.
                }
            });
}

In short what's happening:

1. Turn dataArray into an array of Observable's
2. Each Observable performs the asynchronous operation at the time of subscription and feeds the data to onNext after the async operation finishes
3. Use Observable.concat() on the array of Observable's to ensure sequential execution
4. Use .toList() to combine the results of Observable's into one List

While this code does what I require, I'm not entirely satisfied with this solution for a few reasons:

1. I'm not sure if executing the asynchronous code within the Observable.create()...subscribe(){ is the right way to use Observable's.
2. I read that using Observable.create should be used with caution (ref docs)
3. I'm not sure the behaviour I'm trying to achieve requires creating a new Observable for each item in the dataArray. It seems more natural to me to have one observable that is capable of releasing data sequentially - but again, that may be just my old style of thinking.
4. I've read that using concatMap should do the trick, but I could never figure out how to apply it to this example, and found concat doing just the trick. Anyone caring to explain the difference between the two?

I also tried using .zip function on the array of Observable's before, and while I managed to get the List of results at the end, the async operations were executed in parallel not sequentially.

One very simple and easy way to perform asynchronous operations (calculations etc.) is to use the RxJava Async Utils library (not written by me, just used in our code).

• Easiest way to run a method in a background thread is to use Async.start(this::methodToCall, Schedulers.io()) which returns an Observable<T> that produces the return value of the passed method after the method call completes.
• There are many other alternative methods that allow you to e.g. report intermediate results via an Observer<T>

If you want to use Observable.from() and concatMap() (or similar ways) remember to first move the processing to a background scheduler with .observerOn(Schedulers.io()) or any other scheduler you want.

To call a function over an array of objects sequentially, you need not any async facilities. Just make a for-loop over the array. If you need that loop to execute asynchronously, then please describe what asynchrony you need: either to start the loop after array is asynchronously filled, or you want to react asynchronously to the result of the computation, or both. In either case, class CompletableFuture may help.