psteiger / flow-lifecycle-observer Goto Github PK

Hi all, and thank you for the awesome lib 💪🏻

Would it be unreasonable to have a LICENSE file that specifies with license the library is distributed with?

We are tidying up the licenses' situation in one of our apps, and we hit a blocker with flow-lifecycle-observer.

Thanks a lot 😍

I've found that most of the time in UI code you really want collectLatest rather than collect. If your UI code has gone away rendering a thumbnail or doing a bunch of heavy computations, you need to be able to interrupt that when the latest value of whatever flow you were watching is delivered (which may invalidate that work). Assuming the block of code that you pass to collectLatest employs cooperative cancellations (e.g. calling ensureActive() or yield() relatively often) then this all just works.

At the moment this library uses collect only.

I'm not exactly sure what I'd want the API to look like to support collectLatest. I ended up writing the following extension function instead of using your library. The following extension function simply launches and cancels, it's up to you what you want to put inside, whether it be collect, collectLatest, combine or whatever. This constitutes a solution for me, so this issue is by no means urgent, and I guess it's become more of a suggestion...

/**
 * Like [LifecycleCoroutineScope.launchWhenStarted], except instead of suspending the execution
 * of the block whilst not started, it is cancelled instead and relaunched.
 *
 * One use case for this is if the given block collects from a [SharedFlow] produced by [shareIn] or
 * [stateIn] using [SharingStarted.WhileSubscribed]. Suspending execution of the given block is not
 * sufficient to remove the block from the subscribers of the [SharedFlow], but cancelling it is. It
 * can be important to remove oneself from the subscribers to allow expensive upstream flows to stop
 * when the subscriber count reaches 0.
 *
 * For each purpose, this must be called at most once per lifecycle, since it re-launches the
 * given block of code again whenever the lifecycle gets into the required state. Calling this
 * from e.g. [Fragment.onStart] would cause job duplication after the fragment is stopped and
 * started a few times.
 *
 * If this is a view lifecycle ([Fragment.getViewLifecycleOwner]), then I suggest you run it in
 * [Fragment.onViewCreated]. If this is a regular lifecycle ([Fragment] or [Activity]), then I
 * suggest you run it in [Fragment.onCreate] or [Activity.onCreate].
 */
fun LifecycleOwner.relaunchWhenStarted(block: suspend CoroutineScope.() -> Unit) {
    lifecycle.addObserver(object : DefaultLifecycleObserver {
        private var job: Job? = null

        override fun onStart(owner: LifecycleOwner) {
            job = owner.lifecycleScope.launchWhenStarted(block)
        }

        override fun onStop(owner: LifecycleOwner) {
            job?.cancel()
            job = null
        }
    })
}

Usually ongoing operations should not be interrupted due to rotation. Suggest to add a delay before cancelling the job like the androidx livedata { } builder.

In their version they have an optional timeout parameter which means that observers in the newly created activity after rotation have time to subscribe before the observers in the old activity unsubscribe. Thereby keeping any upstream shared flow alive.

I recently came across an issue that was due to the fact that the flow collection started with observe operator is actually performed right before the lifecycle reaches the STARTED state.

Checking the source code, we can indeed see that:

/**
  * Started state for a LifecycleOwner. For an {@link android.app.Activity}, this state
  * is reached in two cases:
  * <ul>
  *     <li>after {@link android.app.Activity#onStart() onStart} call;
  *     <li><b>right before</b> {@link android.app.Activity#onPause() onPause} call.
  * </ul>
  */
  STARTED,

Well, I can at least tell that onStart() on a lifecycle observer is indeed invoked right before an androidx.Fragment reaches the STARTED state.

It may all sound not really important, but it turned out to matter today for some particular case. I could easy fix that with using launchWhenStarted instead of launch, inside the operator implementation.

This what I'm using in my project (it also has different naming, which I find more explicit. It also has a collectWhileResumed variant):

@file:Suppress("unused")

package com.peterlaurence.trekme.util

import androidx.lifecycle.DefaultLifecycleObserver
import androidx.lifecycle.LifecycleOwner
import androidx.lifecycle.lifecycleScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect

/**
 * Utility extension function to start collecting a flow when the lifecycle is started,
 * and *cancel* the collection on stop, with a custom collector.
 * This is different from `lifecycleScope.launchWhenStarted{ flow.collect{..} }`, in which case the
 * coroutine is just suspended on stop.
 */
inline fun <reified T> Flow<T>.collectWhileStarted(
        lifecycleOwner: LifecycleOwner,
        noinline collector: suspend (T) -> Unit
) {
    ObserverWhileStartedImpl(lifecycleOwner, this, collector)
}

/**
 * Utility extension function on [Flow] to start collecting a flow when the lifecycle is started,
 * and *cancel* the collection on stop.
 */
inline fun <reified T> Flow<T>.collectWhileStartedIn(
        lifecycleOwner: LifecycleOwner
) {
    ObserverWhileStartedImpl(lifecycleOwner, this, {})
}

class ObserverWhileStartedImpl<T>(
        lifecycleOwner: LifecycleOwner,
        private val flow: Flow<T>,
        private val collector: suspend (T) -> Unit
) : DefaultLifecycleObserver {

    private var job: Job? = null

    override fun onStart(owner: LifecycleOwner) {
        job = owner.lifecycleScope.launchWhenStarted {
            flow.collect {
                collector(it)
            }
        }
    }

    override fun onStop(owner: LifecycleOwner) {
        job?.cancel()
        job = null
    }

    init {
        lifecycleOwner.lifecycle.addObserver(this)
    }
}

inline fun <reified T> Flow<T>.collectWhileResumed(
        lifecycleOwner: LifecycleOwner,
        noinline collector: suspend (T) -> Unit
) {
    ObserverWhileResumedImpl(lifecycleOwner, this, collector)
}

class ObserverWhileResumedImpl<T>(
        lifecycleOwner: LifecycleOwner,
        private val flow: Flow<T>,
        private val collector: suspend (T) -> Unit
) : DefaultLifecycleObserver {

    private var job: Job? = null

    override fun onResume(owner: LifecycleOwner) {
        job = owner.lifecycleScope.launchWhenResumed {
            flow.collect {
                collector(it)
            }
        }
    }

    override fun onPause(owner: LifecycleOwner) {
        job?.cancel()
        job = null
    }

    init {
        lifecycleOwner.lifecycle.addObserver(this)
    }
}

Hi @psteiger,

Jose Alcérreca has posted an article about Flows and LiveData, and at the end he describes the rationale behind repeatOnLifecycle, which is exactly the purpose of what we've done here.

repeatOnLifecycle is part of lifecycle-runtime-ktx 2.4.0-alpha01 (so still alpha for the moment). We could mention that in the readme, but it's your call.

The problem with using Flow or SharedFlow as a replacement for LiveData with Jetpack Compose is that Flow.collectAsState() require to pass as argument the initial value of the Flow. StateFlow.collectAsState() doesn't have this requirement as it use the initial value of the StateFlow. Architecture wise it seems problematic to specify the initial value from within the View, it should be up to the ViewModel (which contain this Flow) to define the initial value.

I've found your library after reading the blog posts you linked in its description, and was hopping you got a solution for this.