Hey guys, still new to RX, when I try to use obser...

# rx

[Nikola Milovic]

Hey guys, still new to RX, when I try to use observables to get result from firestore

override fun loadDrillTypes(): Observable<ArrayList<DrillsType>> {

        val list = ArrayList<DrillsType>()

        firebaseFirestore.collection("drilltypes")
            .get()
            .addOnSuccessListener { documents ->
                for (document in documents) {
                    val doc = document.toObject(DrillsType::class.java)
                    list.add(doc)
                }
            }
            .addOnFailureListener { exception ->
                Log.w("TAG", "Error getting documents: ", exception)
            }
        return Observable.just(list)   // delay(2, TimeUnit.SECONDS)
    }

I get an empty ArrayList

compositeDisposable += repository.loadDrillTypes()
            .subscribeWith(object : DisposableObserver<ArrayList<DrillsType>>() {

            override fun onError(e: Throwable) {
                //if some error happens in our data layer our app will not crash, we will
                // get error here
            }

            override fun onNext(data: ArrayList<DrillsType>) {
                Log.d("TAG", data.toString())
            }

            override fun onComplete() {
                Log.d("TAG", "COMPLETE")
            }
        })
    }

But if i add the delay(2, seconds) then I get the normal result that I need... I am 100% sure there is a way around this without having it to be delayed. Or is the delay a standard practice?

[kioba]

Just immediately fires the list and does not wait for the firestore to finish.
OnSuccess/onFaliure is a callback from firestore. It doesn't block the execution of the function

[kioba]

You might be ble tuse a library ehich does the work for you https://github.com/FrangSierra/RxFirebase

[Nikola Milovic]

Thanks. But I'd like to stay within rxkotlin library. I'll look into emiting the data somehow or something along those lines

[kioba]

so just to clarify RXFirebase is just a wrapper library around FIrebase. You can still use RxKotlin 🙂

if you want to implement it manually I would suggest to do something like the following:

Single.create { emitter -> 
val storageTask = firebaseFirestore.collection("drilltypes")
            .get()
            .addOnSuccessListener { documents ->
                val success = documents.map { document.toObject(DrillsType::class.java) }
                emitter.onSuccess(success)
            }
            .addOnFailureListener { exception ->
                if (emitter.isDisposed().not())
                    emitter.onError(exception)
                Log.w("TAG", "Error getting documents: ", exception)
            }

           emitter.setCancellable{ 
               storageTask.cancel()
           }
}

[kioba]

@Nikola Milovic please note that I could not compile the code above, but if you clean it up a little it will work and also handles the cancelation even if the stream is unsubscribed

[Nikola Milovic]

Thanks @kioba, i got the gist of it. I was searching for something like this!

[kioba]

let me know if you need anything else 👍