Android LiveData / architecture components integration

[elizarov]

We need an integration module for Android LiveData architecture components that provides read-to-use extensions for their use with coroutine in a similar style to reactive integration modules. See also #232

[LouisCAD]

@GeoffreyMetais We may help this get real by telling our experience using both LiveData and kotlinx.coroutines

[elizarov]

@LouisCAD Please, do.

[LouisCAD]

The first thing I'd like to say is about null safety and generics:

LiveData is not very Kotlin friendly as its value property can be null even if the type parameter is not nullable, and its Observer SAM passes an explicitly @Nullable value.

However, LiveData is not tightly coupled to the rest of the lifecycle part of Android Architecture Components, and it could be rewritten in a more Kotlin friendly way (and in Kotlin BTW) where it could take an initial value so it can be non null.

LiveData is based on Lifecycle, and you can see in the imports that it really only depends on java.util and Lifecycle (plus two internal classes that only depend on java.util that could be copied if needed), and it's pretty lightweight, so there's no roadblocks to do a more Kotlin friendly implementation where you could specify an initial non null value for a non null generic type.

I think this would be the first step, but the issue is that using the class name LiveData could be confusing. So like Future vs Deferred, there should be another one, unless Google refactors LiveData itself to be more Kotlin friendly nullability-wise.

The names I found (for now) are AliveData and LifecycleData, which are IMHO, a bit less nice than LiveData.

If you don't agree with me, or otherwise, if you have other name ideas, please, tell me.

[GeoffreyMetais]

+1 for nullability, I'm starting my own LiveData for convenience because of this.

ViewModel would need the more coroutines treatment.

In my current refactoring, I have to set an actor to queue operations and guarantee data integrity
I'm not sure I'm doing it well but I'd love this to be managed by a helper ViewModel class.

[LouisCAD]

I agree about ViewModel. I have a JobViewModel subclass in an app that has a protected job property that is cancelled in onCleared(), for use as a parent Job. This is really trivial to do.

@GeoffreyMetais Since you've very recently added VLC Android to GitHub, you can now share permalinks of relevant lines of your LiveData usages in this project to feed this issue 😉. Also, feel free to reach me on kotlinlang Slack if you want to talk about the design and implementation your own LiveData you're starting to work on with the -v option.

A part where I see room for improvement is having a coroutines friendly version of MediatorLiveData to listen to several other sources (LiveData or/and channels) and compute a value out of it, with ability to suspend, and with cancellation support when the LiveData becomes inactive. I thought about this several times, after I used Transformations.switchMap(…), but I don't have a proper design idea yet.

My previous comment reminds me about the fact that LiveData is really all about being Lifecycle aware and is a bit dumb about how it works. For example, when the Activity observing it is restarted (e.g. for a config change like screen density change, locale change, or screen rotation for the many apps that don't handle it other than by restarting their Activities), the LiveData becomes inactive to become active immediately after.

Allowing smarter behaviors could be a design consideration in the coroutines implementation. Lifecycle has multiple states where a (better version of the current)LiveData may want to stay "hot": CREATED, STARTED and RESUMED, but in some cases such as Location updates or other "expensive hotness", it may not want to stay has hot in the CREATED state as it is in the RESUMED state. The current implementation of LiveData that is provided by the Android team has just an "active state", but a coroutine implementation may allow more configuration. It would simply take the highest lifecycle state of the Lifecycles who have an observer.
Another thing that may be interesting would be to add an optional delay before the LiveData steps back a lifecycle state. That way, you could have a LiveData that stays hot for a few seconds before stopping completely, similarly to how some ThreadPools work in Java where some stop unused threads after 60 seconds. This could have some benefits when the user is quickly switching between apps as the "primary" app would have its LiveDatas staying hot if the user switches back quickly enough. The exact behavior I'd use is adding 2 seconds of delay before considering a DESTROYED lifecycle state and stop everything, to accommodate config changes, and adding between 5 seconds and 2 minutes of delay before getting "cold" after all observing lifecycles are back in the CREATED state. I personally don't see where I'd treat STARTED differently from RESUMED since both mean foreground, RESUMED being for an Activity at the top that allows the one behind to stay visible.

I hope I covered useful things for this issue and it's not too verbose for the target folks 🙂

[dmytrodanylyk]

Hi,

Me and Andras Nemeth (@mos8502) from Atlassian are working on integration module. Basically we need extension functions to convert LiveData to ConflatedBroadcastChannel and probably some other channels to imitate cold/hot observable, single, etc.

Let us do and test prototype and share for initial feedback. Stay tuned. 😉

Thanks for feedback @GeoffreyMetais @LouisCAD.

[dmytrodanylyk]

@elizarov

Me and Andras wanted to share some progress on coroutines + arch work. Let’s speak about LiveData first, as it is most used component which is used in Room and ViewModel.

Our initial idea was to wrap LiveData with ConflatedBroadcastChannel which basically act the same. But implementation is error prone and may not reflect all the future changes which google may make.

So we decided to implement it in different way.

import android.arch.lifecycle.Observer
import android.arch.lifecycle.LifecycleObserver

class LiveDataChannel<T>(private val liveData: LiveData<T>)
    : LinkedListChannel<T?>(), SubscriptionReceiveChannel<T?>, Observer<T?>, LifecycleObserver {

    override fun onChanged(t: T?) {
        offer(t)
    }

    override fun afterClose(cause: Throwable?) = liveData.removeObserver(this)

    @OnLifecycleEvent(Lifecycle.Event.ON_DESTROY)
    fun onDestroy() = close()

}

fun <T> LiveData<T>.observeChannel(lifecycleOwner: LifecycleOwner): LiveDataChannel<T> {
    val channel = LiveDataChannel(this)
    observe(lifecycleOwner, channel)
    lifecycleOwner.lifecycle.addObserver(channel)
    return channel
}

fun <T> LiveData<T>.observeChannel(): LiveDataChannel<T> {
    val channel = LiveDataChannel(this)
    observeForever(channel)
    return channel
}

Basically you subscribe to LiveData but as a return type get Channel, not an Observable.

Usage sample:

val channel = liveData.observeChannel(lifecycleOwner)
launch {
    channel.consumeEach { data: List<Data>? ->
        // use data
    }
}

This is a nice way to bridge LiveData - Channels.

[dmytrodanylyk]

Uploaded to github working android sample.

• library
• example-live-data based on developer.android.com

[matejdro]

Also one thing to note is that LiveData is very tightly coupled to the Android's UI thread (you can only set values on the UI thread and observers always return values on UI thread). That makes it less useful for any kind of processing operation. It seems like it was meant more as a mechanism to transmit data to Android UI components.

It looks like to me like ConflatedBroadcastChannel is pretty good starting point. It has list of subscribers, we would just need to expose whether anyone is subscribed or not, similar to onActive and onInactive callbacks of LiveData˛.

[mos8502]

@matejdro why would you want the channel to expose wether anybody is subscribed or duplicate LiveData API like onActive or onInactive ?

ConflatedBroadcastChannel was used in our original approach but IMO it felt more like a reimplementation of LiveData in a kotlin/coroutine way.

The latest proposal is more close to what LiveDataReactiveStreams does for reactive streams: it allows you to consume live data as a channel.

[matejdro]

Well I'm still looking for a good stream/observable implementation with following properties:

1. Stream needs to only be active when there are any active observers (for example if stream provides user's location it would only activate GPS if app's UI would be in foreground, observing the location stream). Additionally, this state needs to be easily propagated to other streams.

   • LiveData does this fantastically, with easily exposed onActive and onInactive methods when extending it and excellent MediatorLiveData that propagates lifecycle to other LiveDatas.
   • RxJava is a bit worse here, you have to mess around with Subjects and doOn[Un]Subscribe() methods, but in the end you can get it to work.
   • Channel on the other hand seems to completely lack any kind of "Hey somebody is observing me, lets turn on" notion (my understanding is Provide abstraction for cold streams #254 is supposed to fix that).

2. Stream needs to be completely independent of the UI thread (UI thread is only meant for drawing the UI. I don't want any data processing operations there)

   • LiveData falls completely on its face here. Every operation returns to the UI thread. Even if you immediately switch to background thread, you would get a lot of wasteful back and forth thread switching
   • Both RxJava and Channel seems to have no such issues

3. Stream operators (map, filter etc.) should be suspend-able to make it easy to chain various suspending tasks.

   • This is not that difficult to do with LiveData, since its map operation is basically pushing a value into another LiveData (putting coroutine in between would be trivial)
   • This is difficult/hacky for RxJava , since its operators need to return value immediately. The best solution I see is using flatMap instead of regular map/filter, combined with publish operator from kotlinx-coroutines-reactive.
   • Channel does this out of the box

Consuming LiveData as a channel fixes the third issue, but it still leaves the "everything happens on the UI thread" issue exposed. Yes, I know Channel will automatically switch threads when consuming, but this still creates wasteful thread switches. It also breaks the propagating part of the first issue. As far as I know there is no way to attach Channel to other Channels and keep active/inactive state in all of them like MediatorLiveData or Subject does. That means that all data generation would have to be done before channel conversion which kinda defeats the point.

ConflatedBroadcastChannel with onActive and onInactive callbacks on the other hand seems to tick all the boxes here. There aren't any operators and no active state propagation, but I feel this would be easier to add than to work around all LiveData problems by wrapping it.

[mos8502]

@matejdro I'm note sure what you're describing here is in scope, but then again it depends on what what approach is preferred.

The approach we took and described by @dmytrodanylyk is a simple extension to LiveData that allows it to be consumed as a channel and yes the value in LiveData is still updated on the main thread but allows consumption on another coroutine context which may mean switching between threads.

Eliminating the middle man (LiveData) is possible (at least in case of the architecture components) if they provide APIs that allow data to be consumed as a channel directly. If this eliminates any unwanted thread switching, still depends on the implementation however.

But you would have the same problem if a 3rd party library decided to expose observable data as LiveData.

It seem what you are looking for is a new subscribable producer (channel) that only starts producing content when there is at least one subscriber.

This is just a very naive implementation and a lot of error cases are not handled (also the name is misleading a bit as OnDemandChannel is not a channel itself)

class OnDemandChannel<out E>(context: CoroutineContext, private val onActive: () -> ReceiveChannel<E>, parent: Job? = null): Closeable {
    private val sendChannel = ConflatedBroadcastChannel<E>()
    private val _actor = actor<Message>(context = context, parent = parent, capacity = Channel.UNLIMITED) {
        while (isActive) {
            require(channel.receive() == Message.OnSubscribe)
            onActivated()
        }
    }

    fun subscribe(): ReceiveChannel<E> {
        require(_actor.offer(Message.OnSubscribe))
        return ActiveChannel(sendChannel.openSubscription())
    }

    override fun close() {
        _actor.close()
    }
    
    private suspend fun ActorScope<Message>.onActivated() {
        var subscribers = 1
        val producer = onActive()
        try {
            while (isActive && subscribers > 0) {
                select<Unit> {
                    onReceive {
                        when (it) {
                            Message.OnSubscribe -> subscribers++
                            Message.OnUnsubscribe -> subscribers--
                        }
                    }
                    producer.onReceive {
                        sendChannel.send(it)
                    }
                }
            }
        } finally {
            producer.cancel()
        }
    }
    
    private enum class Message {
        OnSubscribe, OnUnsubscribe
    }

    private inner class ActiveChannel<out E>(val wrapped: ReceiveChannel<E>) : ReceiveChannel<E> by wrapped {
        override fun cancel(cause: Throwable?): Boolean {
            require(_actor.offer(Message.OnUnsubscribe))
            return wrapped.cancel(cause)
        }
    }
}

[matejdro]

Yes I agree that this is out of scope. My post was mostly replying to @GeoffreyMetais and @LouisCAD that talked about replacing LiveData above.

[elizarov]

I like the approach by @dmytrodanylyk for livedata-to-channel integration. Seems clean to me and solves this particular integration issue. As a side note, I feel and see the need for something like launchConsumeEach, but let's discuss it separately in #328.

@matejdro Is right that missing onActive/onInactive limits usability of ConflatedBroadcastChannel. We shall consider adding them (added as #329)

[dmytrodanylyk]

@elizarov

Should me and @mos8502 proceed with integration module and send PR?

[matejdro]

One important thing to note is that all LiveData.observe... and LiveData.removeObserver calls can only be made on the UI thread (I'm not sure about lifecycle.addObserver).

Code from @dmytrodanylyk above would crash if coroutine with any other dispatcher than UI would try to open channel from LiveData. Maybe it should be wrapped in withContext(UI)? That would make it easier to use that class, but then we would also need a way to disable that somehow for Unit Tests where UI is unavailable.

[dmytrodanylyk]

@matejdro

I think we should realy on how LiveData is implemented.

There must be reasons why google decided to go with this approach. What if in future they allow to observe on non ui thread?

[mos8502]

@matejdro I think I agree with @dmytrodanylyk here. Even though looking at the source it's clear that there are assertions around the current invoking thread, these are not documented. The only case when it's explicitly being called out that you must call the method on the main thread is LiveData::setValue.

Event with the channel "wrapper" you may use LiveData incorrectly and subscribe/unsubscribe from it from the inappropriate thread. Why would the wrapper fix this ?

Maybe just as RxBinding we could make an assertions ourselves to check if those invocations are made from the correct thread, but if that's already done for us by LiveData I don't see the value in it. Also as @dmytrodanylyk says, if removing an observer is going to be allowed in a future release of LiveData we will have unnecessary restrictions in the implementation

[elizarov]

I'm expanding the scope of this issue to include integration with other Android architecture components that are callback-based. In particular, it seems to be useful to be able to suspend coroutine until a lifecycle reaches a certain state. For example, to animate UI coroutine only in resumed state. See also #104 which I'm closing.

[elizarov]

I'm closing it, as this integration is being picked up to Google and will come from their side.

[ashdavies]

@elizarov I'm aware of Google implementing an integration with CoroutineScope and LiveData, but conversion of a LiveData to a Channel (or something similar) would need to be part of the Kotlin library no?

[LouisCAD]

@ashdavies This is likely to become part of livedata-ktx (if a good API can be settled).

[ashdavies]

That'd be cool, I'd just assumed since suggested implementations (I'm referencing @dmytrodanylyk's here), make use of LinkedListChannel, which is now internal to Kotlin, and with Channel's still being experimental/obsolete, it's not clear if the support for integration should come externally (if at all), especially with the existing reactive libraries being Kotlin libs.

[ashdavies]

Not to mention the fact that we still seem to be waiting on a stable release for 2.1.0, but then again, I've always been impatient 😂

[matejdro]

You can make LiveData integration without using internal APIs, you just have to use composition and feed data into channel instead of extending it.

[ashdavies]

@matejdro that's probably true, but @LouisCAD has raised a lot of good points regarding nullability and lifecycle events caveats, that would mean an "opinionated" implementation from either JetBrains, or Google would be more reliable.