Concurrency with Kotlin Flow

This will be another short demo. You’ll see how to handle slow responses in flows using the timeout operator.

Open CategoryViewModel.kt. In the previous lesson, you used the catch operator to handle errors. Now, you’ll add a timeout to the flow:

@OptIn(ExperimentalCoroutinesApi::class, FlowPreview::class) // HERE
private fun moviesForCategory(categoryId: String): Flow<List<MovieViewState>> =
  movieRepository.categories()
    .filter { it.id == categoryId }
    .flatMapLatest { movieCategory -> movieRepository.fetchMoviesForCategory(movieCategory.name) }
    .map { movies ->
      movies.map { movie ->
        val imageRes = movieRepository.fetchMovieImage(movieId = movie.id)
        MovieViewState(movie.id, movie.title, "", imageRes)
      }
    }
    .timeout(3000.milliseconds) // HERE   
    .catch { throwable ->                                         
      Log.e("CategoryViewModel", "Error happened", throwable)
      emit(emptyList())
    }

Add the following imports as well:

import kotlin.time.Duration.Companion.milliseconds

You added the timeout operator to the flow. This operator cancels the flow if it doesn’t emit any items within the specified time. In this case, the flow will be canceled if it doesn’t emit any items within three seconds. To test this, add a delay to fetchMoviesForCategory() in MovieRepository.kt:

fun fetchMoviesForCategory(categoryName: String): Flow<List<Movie>> = flow {
  delay(5000) // Add a delay of 5 seconds
  val movies = movieService.fetchMoviesForCategory(categoryName)
  emit(movies)
}

Run the app and navigate to a category. You’ll see that the flow is canceled after three seconds, and an empty list is emitted. You’ll also see the error message in the log.

Make sure to remove the delay from fetchMoviesForCategory() in MovieRepository.kt.

That ends this demo. Continue with the lesson for a summary.