Cancellations in Kotlin Flow

Cancellations in Kotlin Flow

In this lesson, you’ll learn about handling cancellations in Kotlin Flow. This is important for managing your app’s resources efficiently and ensuring it isn’t doing any unnecessary work.

Cancellation and Coroutine Scope

Cancellation in Kotlin Flow is closely tied to its coroutine context. A flow built within a coroutine is automatically canceled when the coroutine’s scope is canceled. This behavior ensures that flow-related resources are released when they’re no longer needed, preventing potential memory leaks and performance issues.

Examine how flow cancellation relates to coroutine scope:

val factoryScope = CoroutineScope(Dispatchers.Default)

fun carrots(): Flow<Carrot> = fetchNewBatches().asFlow()
  .flatMapLatest { batch -> batch.carrots.asFlow() }

fun processCarrots() {
  factoryScope.launch {
    val firstProcessingLine = scope.launch {
      carrots()
        .map { carrot -> processCarrot() }
        .collect { carrot ->
          println("Processing carrots on Line 1: $type")
        }
    }

    val secondProcessingLine = scope.launch {
      carrots()
        .map { carrot -> processCarrot() }
        .collect { carrot ->
          println("Processing carrots on Line 2: $type")
        }
    }
    
    delay(5000) // During this period you notice that the second line is not needed
    
    secondProcessingLine.cancel() // Cancel the second processing line
  }
}

fun shutdownFactory() {
  factoryScope.cancel() // Cancel the factory scope
}

In this example, processCarrots() creates two processing lines that consume the flow of carrots. When you notice the second line isn’t needed, you cancel it by calling secondProcessingLine.cancel(). This stops the second line from processing any more carrots.

When you’re done with the factory, you cancel the factoryScope by calling factoryScope.cancel(). This cancels all the coroutines running within the scope, including the processing lines.

The timeout Operator

The timeout operator limits the time allowed for an operation within a flow. If the specified timeout is exceeded, the operator throws a TimeoutCancellationException, which can be handled like any other exception in coroutines.

Suppose you want to add a timeout to the carrot processing lines. If a carrot takes longer than one second to process, you can cancel the operation and check what’s causing the delay:

val factoryScope = CoroutineScope(Dispatchers.Default)

fun carrots(): Flow<Carrot> = fetchNewBatches().asFlow()
  .flatMapLatest { batch -> batch.carrots.asFlow() }

fun processCarrots() {
  factoryScope.launch {
    val firstProcessingLine = scope.launch {
      carrots()
        .map { carrot -> processCarrot() }
        .timeout(1000) {                                // Set a timeout of 1 second
          throw TimeoutCancellationException("Carrot processing timed out. Check processing line.")
        }
        .collect { carrot ->
          println("Processing carrots on Line 1: $type")
        }
        .catch { e -> 
          // Check processing line
        }
    }

    val secondProcessingLine = scope.launch {
      carrots()
        .map { carrot -> processCarrot() }
        .timeout(1000) {                                // Set a timeout of 1 second
          throw TimeoutCancellationException("Carrot processing timed out. Check processing line.")
        }
        .collect { carrot ->
          println("Processing carrots on Line 2: $type")
        }
        .catch { e ->
          // Check processing line
        }
    }

    delay(5000) // During this period you notice that the second line is not needed

    secondProcessingLine.cancel() // Cancel the second processing line
  }
}

fun shutdownFactory() {
  factoryScope.cancel() // Cancel the factory scope
}

We’ve added a timeout operator to each processing line. If processing a carrot takes longer than one second, the timeout operator throws a TimeoutCancellationException. You’ll catch this exception using the catch operator and handle it accordingly.

Wrap-Up

In this lesson, you’ve learned about cancellations in Kotlin Flow, particularly how cancellation is managed through coroutine scopes and how the timeout operator can be used to enforce time limits on flow operations.

In the following demo, you’ll see how to apply this in your streaming app.