Concurrency with Kotlin Flow

In this demo, you’ll use what you learned in the previous lesson. Start Android Studio and open the 02-flow-fundamentals/Starter folder.

In the previous demo, you used MovieRepository in HomeViewModel. You’ll upgrade that repository and make it reactive. Just like in the previous lesson, you’ll start with the flow builder.

Open MovieRepository.kt. Right now, fetchMoviesByCategorySuspending() fetches movies by category. As you can see, that function isn’t reactive.

Replace that function with the following:

fun fetchMoviesByCategory(): Flow<Map<String, List<Movie>>> = flow {
  val moviesByCategory = movieService.fetchMoviesByCategory()
  emit(moviesByCategory)
}

To make it work, you also need to add this import:

import kotlinx.coroutines.flow.*

If you take a closer look, you’ll notice that you just implemented a similar function to what you saw in the previous lesson. You’re basically creating a flow, asynchronously fetching movies, and then emitting that result to the reactive stream.

Next, imagine you want to have a predefined set of movie categories in your app. Since you’re learning about Kotlin Flow, you’ll expose them in a reactive way.

In MovieRepository.kt, add the following code:

fun categories(): Flow<MovieCategory> = categoriesDummyData.asFlow()

As stated in the previous lesson, you use the asFlow builder to convert existing collections or sequences into a flow. Here, you used it to convert a previously defined list of movie categories that your app supports.

You’ll use the flowOf builder next. Replace the favoriteCategories() function in the MovieRepository.kt with following:

fun fetchFavoriteCategories(): Flow<List<MovieCategory>> = flowOf(favoriteCategoriesDummyData)

The flowOf builder is often used to emit a small number of values. In this example, you’re using it to reactively expose favorite movie categories.

The last example will show the callbackFlow builder. Suppose you have a rating service in your app that notifies you when a new rating for a movie is submitted, and assume you can only communicate with this service through a callback. It’s good that you know how to bridge the callback with the reactive world.

Add the following code at the end of MovieRepository.kt, before the closing bracket:

fun movieRatings(): Flow<Pair<String, Float>> = callbackFlow {
  // 1
  val listener = object : MovieRatingListener {
    override fun onRatingUpdate(movieName: String, newRating: Float) {
      trySend(movieName to newRating)
    }
  }

  // 2
  movieRatingService.addRatingListener(listener)

  // 3
  awaitClose {
    movieRatingService.removeRatingListener(listener)
  }
}

Here’s how the code above works:

1. You created a MovieRatingListener. Whenever onRatingUpdate() gets called, you want to emit data.
2. You added the listener to movieRatingService.
3. Finally, whenever you no longer need to listen to these events, you remove the listener from movieRatingService.

Here, you used the callbackFlow builder to convert MovieRatingListener into a flow.

The last thing to do is to update HomeViewModel to work with the upgraded MovieRepository. Open the HomeViewModel.kt and update fetchMoviesByCategories() and fetchFavoriteCategories() with the following code:

private fun fetchMoviesByCategories() {
  viewModelScope.launch {
    movieRepository.fetchMoviesByCategory()
      .collect {
        _moviesByCategories.emit(it)
      }
  }
}

private fun fetchFavoriteCategories() {
  viewModelScope.launch {
    movieRepository.fetchFavoriteCategories()
      .collect { _categories.emit(it) } 
  }
}

The logic remains the same as before, but in this case, you’re collecting the results from MovieRepository by calling collect() on the flows you previously exposed.

Excellent! Run the app again — you’ll see it behaves the same way as before, but your ViewModel and Repository are implemented in a reactive way.

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