Room with Kotlin-coroutines observe db changes

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 **/
               }
    }

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

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. Here I'm using lifecycleScope assuming we're in an Activity class.

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

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