LiveData implementation that is based on Kotlin coroutines

MainViewModel.kt

class MainViewModel : DisposableViewModel() {
    val title = ChannelLiveData("New Title 0")
    val isReady = ChannelLiveData(false)

    init {
        disposableLaunch {
            (1..Int.MAX_VALUE).forEach { time ->
                delay(1, TimeUnit.SECONDS)
                title.send("New Title $time")
            }
        }
        disposableLaunch {
            title                    
                    .openSubscription()
                    .map { it.contains("3") }
                    .consumeEach { isReady.send(it) }
        }
    }

}

open class DisposableViewModel : ViewModel() {
    protected val compositeDisposable = mutableListOf<Job>()

    override fun onCleared() {
        super.onCleared()
        runBlocking { compositeDisposable.forEach { it.cancelAndJoin() } }
    }

    protected fun disposableLaunch(block: suspend CoroutineScope.() -> Unit) {
        compositeDisposable += launch(block = block)
    }
}

class ChannelLiveData<T>(value: T) : LiveData<T>(), BroadcastChannel<T> {
    private val channel = ConflatedBroadcastChannel(value)
    private var subscription: Closeable? = null

    override fun onActive() {
        super.onActive()
        subscription = channel.openSubscription().apply {
            launch(CommonPool) { consumeEach(::postValue) }
        }
    }

    override fun onInactive() {
        super.onInactive()
        subscription?.close()
    }

    override val isClosedForSend: Boolean
        get() = channel.isClosedForSend

    override val isFull: Boolean
        get() = channel.isFull

    override val onSend: SelectClause2<T, SendChannel<T>>
        get() = channel.onSend

    override fun close(cause: Throwable?) = channel.close()

    override fun offer(element: T) = channel.offer(element)

    override fun openSubscription() = channel.openSubscription()

    override suspend fun send(element: T) = channel.send(element)
}