RxJava2: execute an async function for every item in a list and wait for callback

Question

I'm struggling with RxJava2. I want to perform a function on each item of a list. This function :

public void function(final Result result) {

    FirebaseFirestore.getInstance().collection(COLLECTION_NAME).document(result.getId()).get().addOnSuccessListener(new OnSuccessListener<DocumentSnapshot>() {
        @Override
        public void onSuccess(DocumentSnapshot documentSnapshot) {
            // do some operation
        }
    });
}

This function is async and use FirebaseFirestore.

So I tried to use RxJava2 on my list to call the function for every item:

Observable.fromIterable(resultList)
                    .concatMap(result -> Observable.fromCallable(new Callable<String>() {
                        @Override
                        public String call() throws Exception {
                            function(result);
                            return "ok";
                        }
                    }))
                    .subscribe(r -> {
                        // do some operation when all firebase async tasks are done
                    });

The concatMap works and the function is called for every item of the list. The problem is that I need a callback when all firebase async tasks are done.

Any help would be much appreciated.

@AnthonyFillion-Maillet: When all tasks are done onComplete should be called? — Amit Vikram Singh
– Amit Vikram Singh, Commented Jul 30, 2018 at 5:29
@Lino Thank you for your answer, I tried it but unfortunately it seems like onComplete is not called — Yogi Bear
– Yogi Bear, Commented Jul 30, 2018 at 14:15
@AnthonyFillion-Maillet make sense. I updated the answer trying to embed into a custom object the information related to the last request — Lino
– Lino, Commented Jul 30, 2018 at 14:47

Lino · Accepted Answer · 2018-07-30 14:45:56Z

I'll try to draw a possible solution:

public class Callback implements OnSuccessListener<DocumentSnapshot> {
     private final ObservableEmitter<DocumentSnapshot> emitter;
     private final boolean last;

     public Callback(boolean lastvalue, ObservableEmitter<DocumentSnapshot> e) {
        this.last = lastvalue;
        this.emitter = e;
     }

     @Override
     public void onSuccess(DocumentSnapshot value) {
         emitter.onNext(value);
         if (last) {
             emitter.onComplete();
         }
     }
}


Observable<DocumentSnapshot> observable = Observable.create(new ObservableOnSubscribe<DocumentSnapshot>() {
        @Override
        public void subscribe(ObservableEmitter<DocumentSnapshot> e) throws Exception {
            int i = 1;
            for (Result result : resultList) {
                /* callback object now knows which is the last request so it can emit the onComplete */
                Callback callbackInstance = new Callback(resultList.size() == i, e);
                i++;
                FirebaseFirestore.getInstance().collection(COLLECTION_NAME)
                            .document(result.getId()).get().addOnSuccessListener(callbackInstance);
                }
            }
        });

then when the subscriber's onComplete action is hit all the requests to Firebase should be completed.

Collectives™ on Stack Overflow

RxJava2: execute an async function for every item in a list and wait for callback

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related