Calling an RxJava Single In Kotlin Lambda

Question

I am trying to tinker with the new Room library in pairing it with RxJava.

I've found a way to use a Single to insert items on the background thread like this, inside of an activity:

Single.fromCallable { AppDatabase.getInMemoryDatabase(this).taskDao().insertAll(task) }
                    .subscribeOn(Schedulers.newThread())
                    .subscribe()

Now, I have a RecyclerView with tasks that has a checkbox you can use to mark an item as complete or not. What I want to do is update the item each time it is checked/unchecked. I'll paste the whole ViewHolder for completion, but note specifically the lambda in bindTask():

inner class TaskViewHolder(view: View?) : RecyclerView.ViewHolder(view) {
        val descriptionTextView = view?.findViewById(R.id.task_description) as? TextView
        val completedCheckBox = view?.findViewById(R.id.task_completed) as? CheckBox

        fun bindTask(task: Task) {
            descriptionTextView?.text = task.description
            completedCheckBox?.isChecked = task.completed

            completedCheckBox?.setOnCheckedChangeListener { _, isChecked ->
                tasks[adapterPosition].completed = isChecked

                Single.fromCallable { itemView.context.taskDao().update(tasks[adapterPosition]) }
                        .subscribeOn(Schedulers.newThread())
                        .subscribe()
            }
        }
    }

This works for the first item I check, but after that I'm unable to click any other checkboxes. I thought the Single would destroy itself, but perhaps I can't do this inside the lambda? Do I need to pull the Single outside of it somehow?

please don't use [`inner class TaskViewHolder`](https://stackoverflow.com/a/3107626/1016472) just [`class TaskViewHolder`](https://kotlinlang.org/docs/reference/nested-classes.html#inner-classes) to [prevent memory leaks](https://stackoverflow.com/a/3107626/1016472). Maybe you need a [WeakReference](https://developer.android.com/reference/java/lang/ref/WeakReference.html) to access the outer class — Ralph Bergmann, Jun 10 '17 at 20:44
I would create member variable emitter using Observable.create(ObservableOnSubscribe{ e-> emitter = e}).yourRxLogic and inside setOnCheckedChangeListener i would use emitter.onNext(someValue) to pass it rx stream. — Tuby, Jun 10 '17 at 21:37

score 3 · Accepted Answer · edited Jun 10 '17 at 22:28

I would create Observable using Observable.create, save that emitter using lambda, and pass next items inside setOnCheckedChangeListener using emitter.onNext()

class TaskViewHolder(view: View) : RecyclerView.ViewHolder(view)
{
    private lateinit var emitter: ObservableEmitter<Task>
    private val disposable: Disposable = Observable.create(ObservableOnSubscribe<Task> { e -> emitter = e })
            .subscribeOn(Schedulers.newThread())
            .observeOn(Schedulers.newThread())
            .subscribe({ itemView.context.taskDao().update(it) })
    val descriptionTextView = view?.findViewById(R.id.task_description) as? TextView
    val completedCheckBox = view?.findViewById(R.id.task_completed) as? CheckBox

    fun bindTask(task: Task) {
        descriptionTextView?.text = task.description
        completedCheckBox?.isChecked = task.completed

        completedCheckBox?.setOnCheckedChangeListener { _, isChecked ->
            tasks[adapterPosition].completed = isChecked
            emitter.onNext(tasks[adapterPosition])
        }
    }
}

Can you give an example on how to add the Rx logic for the update code? I tried putting `.subscribeOn(Schedulers.newThread()).subscribe({ itemView.context.taskDao().update(it) })` on the disposable but I continue to get the "cannot access database on main thread" error. — AdamMc331, Jun 10 '17 at 22:02
Well, your subscribe is fine, try using `.observeOn(Schedulers.newThread())` instead of `subscribeOn`, you are actually subscribing on main UI thread. — Tuby, Jun 10 '17 at 22:24
That worked for me! I will update the answer with my exact code to help any future readers. Thank you! :) — AdamMc331, Jun 10 '17 at 22:27

Ralph Bergmann · Answer 2 · 2017-06-10T20:58:57.847

2

I have not tested it but this should work

class TaskViewHolder(view: View?) : RecyclerView.ViewHolder(view) {

    val descriptionTextView: TextView? = null
    val completedCheckBox: CheckBox? = null

    fun bindTask(task: Task) {
        descriptionTextView?.text = task.description
        completedCheckBox?.isChecked = task.completed

        completedCheckBox?.setOnCheckedChangeListener { _, isChecked ->
            tasks[adapterPosition].completed = isChecked

            itemView.context.taskDao().update(tasks[adapterPosition])
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe({ changeCount -> Timber.i("%,d item(s) updated", changeCount) },
                           { error -> Timber.e(error, "update failed") })
        }
    }
}

interface TaskDao {
    fun update(task: Task): Flowable<Int>
}

Instead of creating a new Single I use the RxJava functionality or Room

edited Jun 10 '17 at 20:58

answered Jun 10 '17 at 20:36

Ralph Bergmann

3,015
4
30
61

Hmm Room is telling me that update methods must return void or int. Is it possible `Flowable` can only be used on queries? – AdamMc331 Jun 10 '17 at 20:41
I played with Room and with RxJava but not in combination :-( Have you added `android.arch.persistence.room:rxjava2` to your build.gradle – Ralph Bergmann Jun 10 '17 at 20:48
Yes I have it. I think it is only used on queries. For Insert, Update, and Delete I'll have to create the Flowables/Singles/WhateverRxComponent to do it on a new thread. I thought that's what I was doing here, but something is happening behind the scenes that I don't fully understand – AdamMc331 Jun 10 '17 at 20:52
I suppose I could access the list whenever `onPause()` is called in the activity, and have it loop through and update the entire list? Would be nice to solve this issue, but I think that's what I'll attempt in the mean time. – AdamMc331 Jun 10 '17 at 21:31

Calling an RxJava Single In Kotlin Lambda

2 Answers2