Unsubscribe from Kotlin Flow without cancelling current scope-CodePudding

I am subscribing to a Kotlin SharedFlow of booleans within a repeatOnLifecycle block in Android. I want to subscribe until I receive the first true boolean and act on it.

As soon as the first true boolean is received I need to unsubscribe and run the processing funtion within the lifecyce scope so it gets cancelled when the lifecycle transitions to an invalid state.

When I call cancel on the current scope and embed the processing code in a NonCancellable context it will not be cancelled on lifecycle events.

I think I would want something like a takeWhile inlcuding the first element that did not match the predicate.

Below are some sample flows where I want to collect all elements until the $ sign:

true $ ...
false, true $ ...
false, false, true $ ...

Sample code:

viewLifecycleOwner.lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        flowOfInterest.collectLatest {
            if (it) {
                stopCollection()
                doOnTrue()
            } else {
                doOnFalse()
            }
        }
    }
}

What is the correct/simplest way to achieve this behavior?

Thanks!

CodePudding user response：

Answering my own question here. What I needed was something like a takeWhile operator that includes the first non-matching element. Such an opeator can be created using the transformWhile operator like this:

viewLifecycleOwner.lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        flowOfInterest.transformWhile {
            emit(it)
            !it
        }.collectLatest {
            if (it) {
                doOnTrue()
            } else {
                doOnFalse()
            }
        }
    }
}

This is not as nice and compact as I had hoped, but it works.

CodePudding user response：

You can use the first function to continue collecting until the given predicate returns true.

viewLifecycleOwner.lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        flowOfInterest.first { 
            if(it) doOnTrue() else doOnFalse()
            it 
        }
    }
}

CodePudding user response：

You can launch a new coroutine inside repeatOnLifecycle function scope, because launch function returns a Job so you can use it to cancel your job later.

Moreover, if you want to listen multiple flows inside repeatOnLifecycle scope, you'll need to launch a child coroutine to let them run in parallel. So they won't block each other and canceling a flow will not affect other flows.

Example code:

lateinit var job: Job

viewLifecycleOwner.lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        job = launch {
            flowOfInterest.collectLatest {
                if (it) {
                    stopCollection()
                    doOnTrue()
                } else {
                    doOnFalse()
                }
            }
        }
    }
}

func stopCollection() {
    job.cancel()
}

CodePudding user response：

You can also use takeWhile to define condition which controls collecting from flow.

 flowOfInterest.takeWhile { !it }.collect {
     //execute smth
 }

If your flow is "cold" after reaching the state, when the condition is not met, the flow will stop emitting at all if there is no more observers. Otherwise, it will continue (e.g. if you use shared flow), but your current observer (which you applied to the question) will stop collecting.

As a workaround to your specific case, if your flow is not "cold", you can call

flowOfInterest.take(1).collect { ... }

to receive the element, on which the predicate returns false.

You can also use additional boolean value that you will modify in collect block when you will reach some condition, which will be the predicate for takeWhile.