RxJava with Firestore Realtime data

Question

I have repository classes. In these classes I make simple collection("..").get() like this.

override fun getTestCollectionItems(): Observable<TestModel> {

    return Observable.create { subscriber ->

        firebaseFirestore.collection(TEST_COLLECTION)
                .get()
                .addOnCompleteListener { task ->

                    if (task.isSuccessful()) {
                        for (document in task.getResult()) {
                            if (document.exists()) {
                                val documentModel = document.toObject(TestModel::class.java)
                                subscriber.onNext(documentModel)
                            }
                        }
                        subscriber.onComplete()
                    } else {
                        subscriber.onError(task.exception!!)
                    }
                }
    }
}

But I found the Real time Firecloud option. If I move the Listener to the Repository then is it good meaning?

I tried the next one:

override fun getRealTimeCollection() : Observable<TestModel> {

    return Observable.create { subscriber ->

        firebaseFirestore.collection(TEST_COLLECTION).document("3lPtYZEEhPdfvZ1wfHIP")
            .addSnapshotListener(EventListener<DocumentSnapshot> { snapshot, e ->
                if (e != null) {
                    Log.w("test", "Listen failed.", e)
                    subscriber.onError(e)
                    return@EventListener
                }

                if (snapshot != null && snapshot.exists()) {
                    Log.d("test", "Current data: " + snapshot.data)
                    val documentModel = snapshot.toObject(TestModel::class.java)
                    subscriber.onNext(documentModel)
                } else {
                    Log.d("test", "Current data: null")

                }
            })
    }
}

with DisposableObservable. But when I disposed it, then the Firebase still sent new datas. It will be memory leak. How can I use the RxJava for this situation? Is it correct to move Real Time data to Repository?

Thank you!

score 10 · Accepted Answer · answered Mar 13 '18 at 07:16

When you create Observable, using the Observable.create method, you get actually ObservableEmitter<T>, with this emitter you should add Cancellable or Disposable using setCancellable()/setDisposable. (you can read about the difference here)
These callbacks will be triggered when you'll dispose your Observable and there you should add the proper un-registration logic of firestore.

override fun getRealTimeCollection(): Observable<TestModel> {

    return Observable.create { emitter ->

        val listenerRegistration = firebaseFirestore.collection(TEST_COLLECTION).document("3lPtYZEEhPdfvZ1wfHIP")
                .addSnapshotListener(EventListener<DocumentSnapshot> { snapshot, e ->
                    if (e != null) {
                        Log.w("test", "Listen failed.", e)
                        emitter.onError(e)
                        return@EventListener
                    }

                    if (snapshot != null && snapshot.exists()) {
                        Log.d("test", "Current data: " + snapshot.data)
                        val documentModel = snapshot.toObject(TestModel::class.java)
                        emitter.onNext(documentModel)
                    } else {
                        Log.d("test", "Current data: null")

                    }
                })
        emitter.setCancellable { listenerRegistration.remove() }
    }
}

score 0 · Answer 2 · answered Dec 19 '19 at 01:45

I believe you shouldn't wrap firebase functions that way as it will not respect schedulers you use when you subscribe, the firebase's callbacks are executed on the main thread.

So if you do:

wrappedFirestore.flatMap{ networkCall } 
        .subscribeOn(ioScheduler)
        .subscribe()

It will still fail with NetworkOnMainThreadException.

RxJava with Firestore Realtime data

2 Answers2