Room with Kotlin-coroutines observe db changes

Question

So, I recently started experimentation with coroutines, I switched from Rxjava2 to coroutines, I haven't got a grasp of it yet but still, I ran into a condition where I needed to observe my database change and update the UI corresponding to that.

RxJava used to provide me with Flowables, Completeable etc. using that I would be able to observe changes in Db.

    abstract fun insert(data: SomeData): Long

    @Query("SELECT * FROM somedata_table")
    abstract fun getData(): Flowable<List<SomeData>>

So here now I used to subscribe to getData and always used to observe changes

Now Enter coroutines, I am using a suspended function with a deferred result to return my responses

    @Insert(onConflict = OnConflictStrategy.IGNORE)
    abstract fun insert(data: SomeData): Long

    @Query("SELECT * FROM somedata_table")
    abstract fun getData(): List<SomeData>

suspend fun getAllSomeData():Deferred<List<SomeData>>{
        return GlobalScope.async (context= coroutineContext){
            database.myDao().getData()
        }
    }

Now I have no way to listen for updates, Channels in coroutines might be the right answer? but I am not sure how to use it with Room.

Solution 1

Use Room 2.2.0 Flows and kotlin coroutines. It's contentious but I dislike LiveData as it gives you results on the UI thread. If you have to do any data parsing you'll have to push everything back to another IO thread. It's also cleaner than using channels directly as you have to do extra openSubscription().consumeEach { .. } calls every time you want to listen to events.

Flow approach Requires the following versions:

// this version uses coroutines and flows in their non-experimental version

org.jetbrains.kotlinx:kotlinx-coroutines-android:1.3.2
androidx.room:room-runtime:2.2.0
androidx.room:room-compiler:2.2.0

Dao:

@Dao
interface MyDao {
     @Query("SELECT * FROM somedata_table")
     fun getData(): Flow<List<SomeData>>
}

class to do observation:

launch {
   dao.getData().collect { data ->
    //handle data here
   }
}

if your calling class is not itself a CoroutineScope you'd have to call launch with the context of something that is. That can be GlobalScope or some other class you create.

GlobalScope.launch {
   dao.getData().collect { data ->
    //handle data here
   }
}

the collect lambda will receive every udpate to the table much like an Rx onNext call.

Solution 2

Currently, there are two different ways of doing that. The first is to use a liveData builder function. To make this work, you need to update lifecycle to androidx.lifecycle:*:2.2.0-alpha01 or any newer version. The LiveData builder function will be used to call getData() asynchronously, and then use emit() to emit the result. Using this method, you will modify your Room getData() function to a suspend function and make the return type wrapped as a LiveData, replacing the Flowable used before.

@Query("SELECT * FROM somedata_table")
abstract suspend fun getData(): LiveData<List<SomeData>>

In your viewmodel you create a liveData which references your Room database

val someData: LiveData<SomeData> = liveData {
    val data = database.myDao().getData() 
    emit(data)
}

The second approach is to get data from our DB as Flow. To use this, you need to update Room to androidx.room:room-*:2.2.0-alpha02 (currently the latest) or a newer version. This update enables @Query DAO methods to be of return type Flow The returned Flow will re-emit a new set of values if the observing tables in the query are invalidated. Declaring a DAO function with a Channel return type is an error

@Query("SELECT * FROM somedata_table")
abstract fun getData(): Flow<List<SomeData>?>

The return type is a flow of a nullable list. The list is nullable because Room will return null when the query has no data fetched.

To fetch data from the flow we will use the terminal operator collect{ } in our Presenter/ViewModel. It is preferable to do this in the ViewModel since it comes with a ViewModelScope. The solution given below assumes we are doing this in a ViewModel where we have a provided viewModelScope.

    fun loadData(){
        viewModelScope.launch {
            database.myDao()
               .getData()
               .distinctUntilChanged().
               .collect{
                        it?.let{ /** Update your obsevable data here **/
               }
    }

Solution 3

Gradle dependencies:

dependencies {
    compile group: 'org.jetbrains.kotlinx', name: 'kotlinx-coroutines-reactive', version: '1.1.1'
}

Room Dao

@Dao
interface HistoryDao : BaseDao<HistoryEntity> {

    @Query("select * from History order by time desc")
    fun observe(): Flowable<List<HistoryEntity>>

    ...
}

Interactor (browserHistoryInteractor below) (layer between dao and Fragment/Presenter)

// To get channel of List<HistoryEntity>:
import kotlinx.coroutines.reactive.openSubscription

fun observe() = historyDao.observe().openSubscription() // convert list to Coroutines channel

Presenter/Fragment/Activity (end point (in my case it is lifecycle-aware presenter))

import kotlinx.coroutines.Job
import kotlinx.coroutines.launch

private val compositeJob = Job() // somewhat equivalent "compositeDisposable" in rx

override fun onCreate() {
    super.onCreate()

    launch(compositeJob) { // start coroutine
        val channel = browserHistoryInteractor.observe() 
        for (items in channel) {  // waits for next list of items (suspended)
            showInView { view?.setItems(items) }
        }
    }
}

override fun onDestroy() {
    compositeJob.cancel() // as in rx you need to cancel all jobs
    super.onDestroy()
}

https://www.youtube.com/watch?v=lh2Vqt4DpHU&list=PLdb5m83JnoaBqMWF-qqhZY_01SNEhG5Qs&index=5 at 29:25