Implementing a cache for Room with BehaviorSubject

Knowledge recommended for the best understanding of the post: Room, Reactive Programming and Kotlin.

After adding Room to a pet project I decided to change Dao returns to return Observables and remove all AsyncTasks from the code.

When I read this article I liked of the form that Room work when the return of the Dao is a Flowable. With the intention to learn more about Reactive Programming I decided to implement this behavior manually.

To implement something more than just copying the behavior from room's flowable, I decided to add a cache layer together. With this, the objectives of the implementation are:

• Return an object where all subscribers can subscribe and keep listening for updates in the model.
• Implement a cache that will permit that new subscribers can receive the actual data without the necessity of accessing the database.

The first step was to find a better way to implement this with Reactive Programming and my choice was to use BehaviorSubject.

If you don't know Behavior Subject, I couldn't write a better explanation than the one provided by the reactivex website, please read it here.

Now that we have the subject defined we can start working in the Room implementation.

Model code:

Dao code:

To abstract the access to the database we can write a DataSource, as following:

With all the boilerplate code created, we can now finally start the implementation of the subscription and cache.

Let's name the class that will be responsible for implementing the behavior as DatabaseObservable:

Going through parts:

• The constructor receives two lambdas, the generateData is responsible for returning the data that are in the database at the first execution and the mergeData will be called with the actual cache and the new data and has to return an updated cache.
• The cache method returns the actual BehaviorSubject, creating it if necessary. The call to replay is to keep only the last cache in the subject and the call to autoConnect is to emit automatically at each new subscribe.
• The method newData is called every time a new data is saved at the database, it will call mergeData to update the cache and emit it again. The code bs!!.blockingFirst() is just to get the actual cache in a synchronous way.

With this we have the code necessary to implement the desired behavior. Now, let's use it.

Explaining the code again:

• The method generateData just return the data actually in the table.
• The method mergeData search for the actual element in the array to update it, if the element is not found we return the cache with the new element at the end.
• The method all just return an Observable with the value from the cache.
• The method save to the same work to save, the change is that now we have to notify the DatabaseObservable that a data update has occurred.

Summarizing, it is a complex solution to a simple problem. I would just use it in a real project if I need a special control of the cache, like updating it when a change in other model occurs. If you just need of an Observable to a table, the Flowable implementation from Room is a perfect solution.

All code provided in this post is available at this repository.