Firebase Realtime Database powered by RxJava for Android

Before I start explaining what this post is really about, let me start with description of the terms used:

Firebase Realtime Database

The Firebase Realtime Database is a cloud-hosted database. Data is stored as JSON and synchronized in realtime to every connected client. Realtime Database is used to sync our db in realtime across all the platforms-android, iOS, web.

Key features of Firebase Realtime Database are:

Realtime: When you connect your app to Firebase, you’re not connecting through normal HTTP. You’re connecting through a WebSocket. WebSockets are much, much faster than HTTP. You don’t have to make individual WebSocket calls, because one socket connection is enough. All of your data syncs automatically through that single WebSocket as fast as your client’s network can carry it.

Offline: Firebase apps remain responsive even when offline because the Firebase Realtime Database SDK persists your data to disk. Once connectivity is re-established, the client’s device receives any changes it has missed, synchronizing it with the current server state.

Accessible from Client Devices: The Firebase Realtime Database can be accessed directly from a mobile device or web browser; there’s no need for an application server.

Rules: Security and data validation are available through the Firebase Realtime Database Security Rules, expression-based rules that are executed when data is read or written.

Scale across multiple databases: Multiple databases can be used in the same Firebase project as per your data requirements. Streamline authentication with Firebase Authentication on your project and authenticate users across your database instances. Control access to the data in each database with custom rules for each database instance.

How to start using Firebase Realtime Database on Android?

Get Started: Add the dependency for Firebase Realtime Database to your app-level build.gradle file:

compile 'com.google.firebase:firebase-database:12.0.0'

Read Data: There are three ways by which you can listen to your data:

• ChildEventListener: Child events are triggered in response to specific operations that happen to the children of a node from an operation such as a new child added through the push() method or a child being updated through the updateChildren() method.

ChildEventListener childEventListener = new ChildEventListener() {
    @Override
    public void onChildAdded(DataSnapshot dataSnapshot, String previousChildName) {
        Log.d(TAG, " --> onChildAdded:" + dataSnapshot.getKey());
        T childNodeObject = dataSnapshot.getValue(exampleClass);
    }
    @Override
    public void onChildChanged(DataSnapshot dataSnapshot, String previousChildName) {
        Log.d(TAG, " --> onChildChanged:" + dataSnapshot.getKey());
        T childNodeObject = dataSnapshot.getValue(exampleClass);
    }
    @Override
    public void onChildRemoved(DataSnapshot dataSnapshot) {
        Log.d(TAG, " --> onChildRemoved:" + dataSnapshot.getKey());
        T childNodeObject = dataSnapshot.getValue(exampleClass);
    }
    @Override
    public void onChildMoved(DataSnapshot dataSnapshot, String previousChildName) {
        Log.d(TAG, " --> onChildMoved:" + dataSnapshot.getKey());
        T childNodeObject = dataSnapshot.getValue(exampleClass);
    }
    @Override
    public void onCancelled(DatabaseError databaseError) {
    }
};
queryDatabase.addChildEventListener(childEventListener);
/* T childNodeObject = dataSnapshot.getValue(exampleClass)
<T>: Class type
<exampleClass>: A .class object
onChildAdded(DataSnapshot dataSnapshot, String previousChildName): 
This callback is triggered once for each existing child and then again every time a new child is added to the specified path. The DataSnapshot passed to the listener contains the the new child's data.
onChildChanged(DataSnapshot dataSnapshot, String previousChildName): This event fires every time a child node is modified, including any modifications to descendants of the child node. The DataSnapshot passed to the event listener contains the updated data for the child.
onChildRemoved(DataSnapshot dataSnapshot): Listen for items being removed from a list. The DataSnapshot passed to the event callback contains the data for the removed child.
onChildMoved(DataSnapshot dataSnapshot, String previousChildName): This event is triggered whenever the onChildChanged() callback is triggered by an update that causes reordering of the child. It is used with data that is ordered with orderByChild() or orderByValue().
onCancelled(DatabaseError databaseError): This event is triggered whenever there is some DatabaseError while fetching data from db. */

We can use the dataSnapshot fetched from server and match it with whatever class required using reflection, before doing this make sure the class has a default constructor that get’s called on dataSnapshot.getValue(exampleClass).

• ValueEventListener: Attaching a ValueEventListener to a list of data will return the entire list of data as a single DataSnapshot, which you can then loop over to access individual children.

ValueEventListener nodeEventListener = new ValueEventListener() {
    @Override
    public void onDataChange(DataSnapshot dataSnapshot) {
        Log.d(TAG, " --> onDataChanged:" + dataSnapshot.getKey() 
+ " --- " + dataSnapshot.getChildrenCount());
        List<T> listOfEntity = new ArrayList<T>();
        for (DataSnapshot singleDataSnapshot : dataSnapshot.getChildren()) {
            T rootNodeObject = singleDataSnapshot.getValue(exampleClass);
            listOfEntity.add(rootNodeObject);
        }
    }
    @Override
    public void onCancelled(DatabaseError databaseError) {
    Log.d(TAG, " --> onCancelled Single", databaseError.toException());    
    }
};
queryDatabase.addValueEventListener(nodeEventListener);
/* onDataChange(DataSnapshot dataSnapshot): Read and listen for changes to the entire contents of a path.
onCancelled(DatabaseError databaseError): This event is triggered whenever there is some error while fetching data from db. */

• ListenerForSingleValueEvent: Same as ValueEventListener but is triggered only once and then doesn’t trigger again.

ValueEventListener nodeSingleEventListener = new ValueEventListener() {
    @Override
    public void onDataChange(DataSnapshot dataSnapshot) {
        Log.d(TAG, " --> onDataChanged Single:" + dataSnapshot.getKey() + " --- " + dataSnapshot.getChildrenCount());
        List<T> listOfEntity = new ArrayList<T>();
        for (DataSnapshot singleDataSnapshot : dataSnapshot.getChildren()) {
            T rootNodeObject = singleDataSnapshot.getValue(exampleClass);
            listOfEntity.add(rootNodeObject);
        }
    }
    @Override
    public void onCancelled(DatabaseError databaseError) {
        Log.d(TAG, " --> onCancelled Single", databaseError.toException());   
   }
};
queryDatabase.addListenerForSingleValueEvent(nodeSingleEventListener);
/* onDataChange(DataSnapshot dataSnapshot): Read and listen for changes to the entire contents of a path.
onCancelled(DatabaseError databaseError): This event is triggered whenever there is some error while fetching data from db. */

After data is fetched the callback is removed until triggered again manually.

Write Data: There are two ways by which you can write data to your db:

• Updating value: You can use setValue(..) to save data to a specified reference, replacing any existing data at that path or pass a custom Java object.

queryDatabase.child("<node_name>").child(<node_id>).child("<field_name>").setValue(<field_value>);
or
queryDatabase.child("<node_name>").child(<node_id>).setValue(<new_object>);

• Update specific children simultaneously: To simultaneously write to specific children of a node without overwriting other child nodes, use the updateChildren(..) method.

String key = mDatabase.child("<node_name>").push().getKey();
Map<String, Object> valueMap = new HashMap<>();
Map<String, Object> childUpdates = new HashMap<>();
childUpdates.put("/<node_name>/" + key, valueMap);
childUpdates.put("/<other_node_name>/" + <other_node_id> + "/" + key, valueMap);
queryDatabase.child.updateChildren(childUpdates);

This example uses push() to create a post in the node containing posts for all users at /<node_name>/key and simultaneously retrieve the key with getKey(). The key can then be used to create a second entry in the user's posts at /<other_node_name>/<other_node_id>/key.

Delete data: The simplest way to delete data is to call removeValue() on a reference to the location of that data. You can also delete by specifying null as the value for another write operation such as setValue(..) or updateChildren(..). You can use this technique with updateChildren(..) to delete multiple children in a single API call.

Detach Listeners: Callbacks are removed by calling the removeListener() method on your Firebase database reference. Calling removeListener() on a parent listener does not automatically remove listeners registered on its child nodes; removeListener() must also be called on any child listeners to remove the callback.

Write data offline: Firebase apps automatically handle temporary network interruptions. Cached data is available while offline and Firebase resends any writes when network connectivity is restored but is not available when user restarts the app.

When you enable disk persistence, your app writes the data locally to the device so your app can maintain state while offline, even if the user or operating system restarts the app. The Firebase Realtime Database client automatically keeps a queue of all write operations that are performed while your app is offline, this queue is also persisted to disk so all of your writes are available when the user or operating system restarts the app. When the app regains connectivity, all of the operations are sent to the Firebase Realtime Database server.

FirebaseDatabase.getInstance().setPersistenceEnabled(true);

Keeping Data Fresh: The Firebase Realtime Database client automatically downloads the data at these locations and keeps it in sync even if the reference has no active listeners. By default, 10MB of previously synced data is cached. If the cache outgrows its configured size, the Firebase Realtime Database purges data that has been used least recently. Data that is kept in sync is not purged from the cache.

queryDatabase.keepSynced(true);

Query: With Firebase database queries, you can selectively retrieve data based on various factors. To construct a query in your database, you start by specifying how you want your data to be ordered/filtered using one of the functions: orderByChild(..): Order results by the value of a specified child key or nested child path. orderByKey(): Order results by child keys. orderByValue(): Order results by child values. limitToFirst(..): Sets the maximum number of items to return from the beginning of the ordered list of results. limitToLast(..): Sets the maximum number of items to return from the end of the ordered list of results. startAt(..): Return items greater than or equal to the specified key or value depending on the order-by method chosen. endAt(..): Return items less than or equal to the specified key or value depending on the order-by method chosen. equalTo(..): Return items equal to the specified key or value depending on the order-by method chosen

Optional: Configure ProGuard: When using Firebase Realtime Database in your app along with ProGuard you need to consider how your model objects will be serialized and deserialized after obfuscation. If you use DataSnapshot.getValue(Class) or DatabaseReference.setValue(Object) to read and write data you will need to add rules to the proguard-rules.pro file:

# Add this global rule
-keepattributes Signature
# This rule will properly ProGuard all the model classes in
# the package com.yourcompany.models. Modify to fit the structure
# of your app.
-keepclassmembers class com.yourcompany.models.** {
  *;
}

I think after reading all the points mentioned above you’ll able to implement Firebase Realtime Database in your Android app, for details understanding read this https://firebase.google.com/docs/database/.

RxJava

RxJava is a Java VM implementation of Reactive Extensions: a library for composing asynchronous and event-based programs by using observable sequences. It extends the observer pattern to support sequences of data/events and adds operators that allow you to compose sequences together declaratively while abstracting away concerns about things like low-level threading, synchronization, thread-safety and concurrent data structures.

Build blocks for RxJava

Observables: Observables are the sources for the data. Usually they start providing data once a subscriber starts listening. An observable may emit any number of items (including zero items). It can terminate either successfully or with an error.

Subscribers: A observable can have any number of subscribers. If a new item is emitted from the observable, the onNext(..) method is called on each subscriber. If the observable finishes its data flow successfully, onComplete() method is called on each subscriber. Similarly, if observable finishes its data flow with an error, the onError(..) method is called on each subscriber.

You can find a lot of articles on google that explains RxJava pretty well (https://blog.mindorks.com/a-complete-guide-to-learn-rxjava-b55c0cea3631 — a complete RxJava guide) but let’s focus on what we need to learn to implement it with Firebase Realtime Database.

How to start using RxJava on Android?

Get Started: Add the dependency on your gradle file

implementation 'io.reactivex.rxjava2:rxandroid:1.x.x'
implementation 'io.reactivex.rxjava2:rxjava:1.x.x'
/* rxandroid is used because it provides a Scheduler that schedules on the main thread or any given Looper */

Creating and subscribing:

Observable.create(subscriber -> {
        try {
            //do some time taking operation
            subscriber.onNext(<object>);    
            subscriber.onCompleted(); 
        } catch (Exception e) {
            subscriber.onError(e);        
        }
}).subscribeOn(<a ui/background thread scheduler>)
.observeOn(<a ui/background thread scheduler>)
.subscribe(new Subscriber<Object>(){
 @Override public void onCompleted() {
 }
 @Override public void onError(Throwable e) {
 }
 @Override public void onNext(Object object) {
 }
});
/* First, a observable is created in which you can do any time taking operation similar to AsyncTask doInBackground(). After the operation is completed you can pass <object> using subscriber.onNext(<object>) and if there's some error handle it and pass it using subscriber.onError(e). onNext(<object>), onError(e) and onCompleted() are received on subscriber's onNext(Object object), onError(Throwable e) and onCompleted() respectively. 
Note - Both onError(e) and onCompleted() results in closing the subscriber i.e subscriber won't receive any update until manually called again. */

observeOn(..) and subscribeOn(..) are used to switch thread. Depending on your requirements you can use ui/background thread on whichever method.

observeOn(..) works only downstream i.e the statements written after it will run on the thread changed by observeOn(..). Consequent observeOn(..)s do change the thread.

subscribeOn(..) works downstream and upstream. i.e whatever is written before/after this method they’ll run on the thread specified by subscribeOn(..). Consecutive subscribeOn(..)s do not change the thread.

Thread change by an observeOn(..) cannot be overridden by a subscribeOn(..) i.e once the observeOn(..) changes the thread then subscribeOn(..) written after it has no meaning.

Example Code

I have implemented some custom classes that can used anywhere in your code. Let’s start with the demo of the code.

Note:- Code is written in both kotlin and java

NateshR/FirebaseRx

FirebaseDataRepository<T>: An interface used to bind data and view i.e view can pass it’s requirements to data:

• T can be your pojo of data structure present at Firebase server.
• ListenerType (type: enum, values: NODE, SINGLE_NODE) whether you want to ValueEventListener or ListenerForSingleValueEvent.
• QueryType (type:enum, values: ORDER_BY_CHILD, ORDER_BY_KEY, ORDER_BY_VALUE, START_AT, END_AT, EQUAL_AT, LIMIT_TO_FIRST, LIMIT_TO_LAST, DISCUSSION_ID, TIME_TO_LIVE, FILTER_ID, DIALOG_ID) which passed as LinkedHashMap<FirebaseDataRepository.QueryType, String> and query is parsed in parseQuery(..) method present in FirebaseDataStoreFactory.
• Implement respective methods as per your requirements: childData(..), nodeChildList(..), singleNodeChildList(..).

FirebaseDataStoreFactory<T>: A custom class that contains implementation: fetching data from Realtime Database and passing it using RxJava.

• Observable<T> data(Class<T> exampleClass, Query queryDatabase): exampleClass: pojo of data structure reflected using dataSnapshot.getValue(exampleClass), queryDatabase: final query after parsed using parseQuery(..). ChildEventListener wrapped around Observable.create(..), so whenever a child is added, changed, removed, moved or cancelled we’ll receive a callback on the respective method and will get pass through onNext(..) or onError(..). onCompleted() is not called since we want to listent to this continuously.
• Observable<List<T>> dataList(FirebaseDataRepository.ListenerType listenerType, Class<T> exampleClass, Query queryDatabase): listenerType:specifies type of listener i.e NODE or SINGLE_NODE, exampleClass: pojo of data structure reflected using dataSnapshot.getValue(exampleClass), queryDatabase: final query after parsed using parseQuery(..). ValueEventListener or ListenerForSingleValueEvent wrapped around Observable.create(..), so whenever there’s a change in list of data i.e added, changed etc we’ll receive a callback on the respective method and will get pass through onNext(..) or onError(..). If listenerType is NODE, onCompleted() is not called since we want to listent to this continuously but if it is SINGLE_NODE onCompleted() is called since we want to close the subscriber as once data is fetched in case of ListenerForSingleValueEvent there is no callback after so if we didn’t call onCompleted(), a no use subscriber will live throughout.
• Query parseQuery(Query finalQuery, LinkedHashMap<FirebaseDataRepository.QueryType, String> queryMap): finalQuery: returning the final query after evaluation it against queryMap, queryMap: a LinkedHashMap containing FirebaseDataRepository.QueryType and it’s respective value. This method is used to parse query on a DatabaseReference object.

AccessCredentialsDataRepository: An example class implementing FirebaseDataRepository<AccessCredentialsEntity> and using FirebaseDataStoreFactory<AccessCredentialsEntity> to fetch data from firebase server by passing a DatabaseReference(containing path information of data).

AccessCredentialsEntity: An example pojo class of data structure present at Firebase server.

If you are following this post thoroughly you’ll understand these terms.(All the details are mentioned in Firebase and RxJava sections above). To get started just add all the required dependencies in your grade and pull FirebaseDataRepository<T>, FirebaseDataStoreFactory<T> in your project. For reference how to use these classes, read: AccessCredentialsDataRepository.