RxJava需要有关如何将Firebase实现为远程数据源的指导

Question

我是RxJava的新手和一般的MVP结构。我有更多以传统方式使用Firebase RTDB的经验，现在我想知道如何最好地将其作为RemoteDataSource（ex TaskRemoteDataSource）进行调整。在其他MVP示例中，我将使用callback.onTaskLoaded（task），但是示例契约需要返回Flowable。

来自Firebase查询和写入的回调世界。我想知道使用RxJava处理Firebase回调的最佳做法是什么。在搜索github时，有N个RxFirebase库似乎可以工作，但我不确定将哪个库与我的项目耦合。谢谢。

Answer 1

所以我使用RxJava包围Firebase RTD或Firestore的方式，在这种情况下，Firestore的方式如下。假设您想要从Firestore中检索数据（这些想法与RTD几乎相同，只需使用实时数据库）

    /**
     * Retrieves a document snapshot of a particular document within the Firestore database.
     * @param collection
     * @param document
     * @return
     */
    @Override
    public Observable<DocumentSnapshot> retrieveByDocument$(String collection, String document) {
        //Here I create the Observable and pass in an emitter as its lambda paramater. 
        //An emitter simply allows me to call onNext, if new data comes in, as well as catch errors or call oncomplete. 
        ///It gives me control over the incoming data.

        Observable<DocumentSnapshot> observable = Observable.create((ObservableEmitter<DocumentSnapshot> emitter) -> {
            //Here I wrap the usual way of retrieving data from Firestore, by passing in the collection(i.e table name) and document(i.e item id) and 
        //I add a snapshot listener to listen to changes to that particular location.

          mFirebaseFirestore.collection(collection).document(document)
                    .addSnapshotListener((DocumentSnapshot documentSnapshot, FirebaseFirestoreException e) -> {
                        if(e != null) {
                     //If there is an error thrown, I emit the error using the emitter
                            emitter.onError(e);
                        }else{
                     //If there is a value presented by the firestore stream, I use onNext to emit it to the observer, so when I subscribe I receive the incoming stream
                            emitter.onNext(documentSnapshot);
                        }
                    });
            });
            //I finally return the observable, so that I can subscribe to it later.
            return observable;
        }