In case you haven’t heard about RxJava, it’s a library for composing asynchronous and event-based programs using observable sequences for Java and Android apps.
Making it excellent in conjunction with Firestore for things like say, responding to a change in the database and fetching the updated data while handling it on the right thread to keep your app running smoothly.
Not to mention, it makes your app super modular and concise making it easy to manage and write tests for.
This is going to be an introduction to RxJava by using it with Firestore and the RxFirestoreKt library as well, so I hope you’re not too rusty with your Kotlin.
We’ll start by going over a very basic use-case scenario, then go over a few scenarios you’ll almost definitely run into where I had some trouble with myself.
Definitions
There a few things in RxJava (or ReactiveX in general) that we need to know about in order to use them effectively.
First we have the Observable which is an object that emits data. The emitted data is received by an Observer, otherwise known as a Subscriber which is where we can write what happens whenever data is received.
There are several different types of Observables, one is the Flowable which works just like a regular observable but with Backpressure Strategy, a way of handling data when our observable emits more than our Subscriber can handle. We’ll be using Flowable’s a lot in this guide.
Then we have the Single which either completes with a single value or an error, the Maybe which completes with or without a single value, and a Completable which doesn’t emit a fetched value but instead, emits a signal whether or not it’s completed an operation.
While Observables and Flowables emit continuous streams of data until you stop them, you can say a Single emits one value, a Maybe emits one or zero values, and a completable only emits zero values, assuming none of these run into errors, otherwise, you won’t be getting any values from anything.
Downloading the Libraries
implementation 'com.google.firebase:firebase-firestore:17.1.2' implementation 'io.reactivex.rxjava2:rxjava:2.2.3' implementation 'de.aaronoe:rxfirestorekt:0.1.1'
Before we get to work, we need a few libraries. First you need to connect your app with Firebase and have the Firestore library imported, if you don’t know how to connect your app to Firebase, here’s a link on how to do that.
Then we’ll need to RxJava and RxFirestoreKt libraries as well.
For the sake of the video, I’m using a Shopping List app I’m currently developing with an MVVM architecture. If you don’t know what it is, I split my classes into three packages, the Model where my objects’ data structure is defined, the View Model that fetches all the data, and the View which presents the data from the View Model in the User Interface,
1. Fetching Documents
fun getUsers(): Flowable<List<User>> {
val usersRef = firestore.collection("users")
return usersRef.getFlowable(BackpressureStrategy.LATEST)
}
Now all we need to do is call getFlowable and return the result of that method, hence the return type of my method should be a Flowable of Lists of ShoppingLists which is a class defined in my Model.
Another thing to note is the parameter passed into getFlowable, BackpressureStrategy.LATEST. While Backpressure Strategies can be a whole article of their own, LATEST is a safe go-to strategy if you want your subscriber to receive the latest emissions safely after being overloaded.
private fun getUsers() {
val getUsersSubscriber = listService.getUsers()
.observeOn(Schedulers.io())
.subscribe {
doStuffWithUsers(it)
}
// Prepare the disposable for cleanup
compositeDisposable.add(getUsersSubscriber)
}
Now in my View, the activity, I’m going to create a method, get the Flowable we returned earlier, and subscribe to it. What this does is kick off the operation, and every time new incoming data is received, the program executes whatever we put inside these braces.
The best part of this is that if our document’s structure matches that of our model, RxFirestoreKt automatically maps the document to our model so things like my id and name are automatically set.
We can also go one step further and assign the operation onto a thread of our choosing. While multithreading can be a whole lecture of its own, I tend to use just 2 the most. That’s Schedulers.io, a background thread dedicated to network calls and database operations, and AndroidSchedulers.mainThread which is obviously enough, the main thread which is the only one that has access to the UI, and the one that will cause the app to freeze if overloaded.
Since our Flowable will keep on emitting values indefinitely, we want to stop it as we leave the activity, so each time I create a new subscriber, I add it to a CompositeDisposable.
override fun onDestroy() {
super.onDestroy()
compositeDisposable.dispose()
}
Like this, I can stop all subscribers in the activity by ‘disposing’ them.
2. Inserting Documents and Adding a new document’s id inside itself
fun insertListItem(itemName: String): Single<DocumentReference> {
// Build the Map for the Collection
val item = HashMap<String, Any>()
item["name"] = itemName
item["count"] = defaultCount
// Add the data and return a 'Single' observable
val itemsRef = listRef!!.collection("items")
return itemsRef.addDocumentSingle(item)
}
While inserting a document isn’t exactly a function you’d expect to receive a value back from, you certainly can use it as an Observable to your advantage. RxFirestoreKt allows us to insert a document to return a Single of the DocumentReference it’s been inserted into.
private fun insertList(listTitle: String) {
// Add the List to the Database
val insertListObserver = listService?.insertList(listTitle)
?.observeOn(Schedulers.io()) // Background thread
?.subscribe({ ref ->
val map = HashMap<String, Any>()
map["id"] = ref.id
ref.set(map, SetOptions.merge())
}, { exception -> handleError(exception) })
// Prepare disposable for cleanup
insertListObserver?.let { compositeDisposable.add(insertListObserver) }
}
Don’t let the ‘?’s confuse you. That’s just Kotlin null handling
In this case, I’m inserting a list item and as soon as it’s inserted and has its id generated by Firestore, I get its ID from the reference and perform just a regular set operation from vanilla Firestore so that I can take advantage of the SetOptions.merge.
Now you might be thinking, why stick the id in the document? Remember the automatic mapping from earlier? Now we can take advantage of that and download each List Item’s id along with the rest of its data. Useful for when we want to open a details activity.
One more thing to note here. On the subscriber, I also have a lambda handling an onError callback. Just another thing we can do with subscribers.
3. Performing OR Queries
fun getListMembers(): Observable<Any> {
// Query from users collection
val userObservables = ArrayList<Observable<List<User>>>()
for (i in 0 until users!!.size) {
userObservables.add(firestore.collection("users").whereEqualTo("id", users!![i]).getObservable())
}
return Observable.combineLatest(userObservables) { args -> return@combineLatest args }
}
The Firestore docs themselves state that chaining query expressions work via logic AND and if we want to perform queries via logic OR, we have to do that within the app instead. Luckily for us, RxJava makes that easy.
I’m using a special case myself. I have an array of user ids. I want to query all the users in this array from the users collection, but I don’t want to have to download the whole collection, because if my app becomes big, then that will be a huge download.
To do this, I prepare an Array List of observables, and for each user id in my array, I add a query of each single user from the user’s collection.
To finish, I use Observable.combineLatest which combines all these different observables into a single stream so that my subscriber can handle them all like a single List of Users.
So that’s my little introduction to RxJava using RxFirestoreKt, by no means am I an RxJava expert but I do hope you found this useful. If you did, why not be a subscriber to the blog and I do promise to emit some data that you will find very useful.