How to cancel/unsubscribe from coroutines Flow

Question

I notice a strange behavior when trying to prematurely cancel from a Flow. Take a look at the following example.

This is a simple flow that emits integer values

  private fun createFlow() = flow {
        repeat(10000) {
            emit(it)
        }
    }

Then I call the createFlow function using this code

  CoroutineScope(Dispatchers.Main).launch {
            createFlow().collect {

                Log.i("Main", "$it isActive $isActive")
                if (it == 2) {
                    cancel()
                }
            }
        }

This is what is printed out

0 isActive true
1 isActive true
2 isActive true
3 isActive false
4 isActive false
etc...etc

Now I would expect that the flow should stop emitting integers once it reaches the value of 2 but instead it actually switches the isActive flag to false and keeps emitting without otherwise stopping.

When I add a delay between emissions the flow behaves as I would expect.

private fun createFlow() = flow {
    repeat(10000) {
        delay(500) //add a delay
        emit(it)
    }
}

This is what is printed out after calling the flow again (which is the expected behaviour).

0 isActive true
1 isActive true
2 isActive true

What can I do to cancel the flow emission exactly at the specified value without adding delay?

This open issue seems to describe the same thing, and I think this issue tracks the progress of this fix. — Marko Topolnik
– Marko Topolnik, Commented Jan 10, 2020 at 11:20

ThanosFisherman · Accepted Answer · 2020-01-10 12:36:49Z

18

I came across a workaround in this related issue

I have replaced every single collect with a safeCollect function in my project:

/**
 * Only proceed with the given action if the coroutine has not been cancelled.
 * Necessary because Flow.collect receives items even after coroutine was cancelled
 * https://github.com/Kotlin/kotlinx.coroutines/issues/1265
 */
suspend inline fun <T> Flow<T>.safeCollect(crossinline action: suspend (T) -> Unit) {
  collect {
    coroutineContext.ensureActive()
    action(it)
  }
}

answered Jan 10, 2020 at 12:36

ThanosFisherman

5,88812 gold badges41 silver badges70 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Ivan Morgillo Over a year ago

I believe you can add .cancellable() to your Flow: > This operator provides a shortcut for .onEach { currentCoroutineContext().ensureActive() }.

i30mb1 · Accepted Answer · 2020-12-03 07:38:32Z

14

I want to add that in 1.3.7 version emissions from flow builder now check cancellation status and are properly cancellable. So the code in question will work as expected

answered Dec 3, 2020 at 7:38

i30mb1

4,9063 gold badges22 silver badges36 bronze badges

Comments

Cruces · Accepted Answer · 2020-11-30 22:21:20Z

1

I came up with this recently

it seems that it will only actually cancel if it reaches a suspending point and in your code that emits there is no such point

to solve this either add yield() between emissions or some other suspending function like delay(100)

answered Nov 30, 2020 at 22:21

Cruces

3,1593 gold badges32 silver badges58 bronze badges

Collectives™ on Stack Overflow

How to cancel/unsubscribe from coroutines Flow

3 Answers 3

1 Comment

Comments

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

3 Answers 3

1 Comment

Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related