android - RxJava2:对列表中的每个项目执行异步功能，然后等待回调

我在与rxjava2抗争。我想对列表中的每一项执行一个函数。此函数：

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
        }
    });
}

此函数是异步的，使用firebaseforestore。
所以我尝试使用列表中的rxjava2为每个项调用函数：

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
                    });

concatmap可以工作，并为列表中的每个项调用该函数。问题是，当所有firebase异步任务完成时，我需要回调。
任何帮助都将不胜感激。

最佳答案

我试着画一个可能的解决方案：

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);
                }
            }
        });

当用户的onComplete操作被点击时，所有对firebase的请求都应该完成。

关于android - RxJava2:对列表中的每个项目执行异步功能，然后等待回调，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/51572018/