An Introduction to Flow for Android April 29, 2020 | Martin Tøften

Flow is a new feature of Kotlin Coroutines. Whereas suspending functions in Kotlin can only return a single value; Flow can return multiple values.

You can think of

as a stream. Values flow downstream like water in a river. Values can be created from asynchronous actions such as network requests or database calls. If you are familiar with - behaves similarly to and . is built on top of coroutines. We love coroutines because they bring structured concurrency to Kotlin.

In this article we will look into the basics of

and in which scenarios we can use it.

Example 1:

val streamOfInts = flow {
  println("Start emitting")
  for (i in 0..1) {
    delay(100)
    println("Sending $i")
    emit(i)
  }
  println("Finished emitting")
}

streamOfInts.collect {
  println("Receiving $it")
}

println("Flow completed")

This prints:

Start emitting
Sending 0
Receiving 0
Sending 1
Receiving 1
Finished emitting
Flow completed

The first thing we see in the code above is the flow builder. Inside the flow builder you can do any asynchronous work and use

to send some kind of value. A 100 millisecond delay is used to simulate a network request before emitting a value. is a handy suspend function which delays for a specified time before resuming where it left off. Two values are emitted before the flow completes. is also a suspend function and it is used to start collecting values from the flow. Flows are cold streams so the flow will not start emitting before is called. is similar to in . The flow builder is the producer and the collector is the consumer.

Operators

can be used in a reactive programming style. It has a lot of different operators that can transform a flow with functional operators. Operators such as , , and many more. supports suspending functions on most operators which means that you can execute sequential asynchronous tasks in operators like .

Let's get some data from our database and then from the network:

Example 2:

val stream = flow {
  println("Computed on $thread")
  val dataFromDatabase = database.getData()
  emit(dataFromDatabase)
  val dataFromNetwork = network.getData()
  emit(dataFromNetwork)
}

stream
  .map { it.plus(" mapped on $thread") }
  .flowOn(Dispatchers.IO)
  .collect { println(it.plus(" collected on $thread")) }

This prints:

Computed on worker-1
Database data! mapped on worker-1 collected on main
Network data! mapped on worker-1 collected on main

The code above will first get data from the database and emit it to the collector. It then fetches updated data from the network and emits it. For simplicity we skip the caching of network data to the database.

The view can listen to this stream of data and update the UI when new data is collected. We have used the

operator to transform which thread the code is running in. You can see that we introduced a new operator named which will switch the upstream context in the flow. If you want to execute an operation off the main thread you can set which kind of dispatcher you want to use (, , , ). This is similar to in .

You can find all operators here

Custom Operators

Creating your own custom operators is easily achieved. Here is a custom operator that transforms all strings to uppercase.

Example 3:

fun Flow<String>.uppercase(): Flow<String> {
  return flow {
    collect { emit(it.toUpperCase()) }
  }
}

flowOf("No", "soup", "for", "you!")
  .uppercase()
  .collect { println(it) }

This prints:

NO
SOUP
FOR
YOU!

Data flow in Android

In Android we have to think about how technology works in the context of

, and . There are multiple ways of doing this. Personally I use in repositories and in which I will come back to later. Here's an example of how this might look.

Example 4:

class WeatherRepository {
  fun getWeather(): Flow<Weather> {
    return flow {
      emit(Weather("Rainy :("))
      emit(Weather("Cloudy :|"))
      emit(Weather("Sunny :)"))
    }
  }
}

class WeatherViewModel(
  private val weatherRepository: WeatherRepository
) : ViewModel() {
  val weather = MutableLiveData<Weather>()

  init { getWeather() }

  private fun getWeather() {
    viewModelScope.launch {
      weatherRepository.getWeather()
        .flowOn(Dispatchers.IO)
        .collect { weather.value = it }
      }
    }
  }

class WeatherFragment : Fragment() {
  private fun initObservers() {
    viewModel.weather.observe(this, Observer { println(it) })
  }
}

This will print:

Rainy :(
Cloudy :|
Sunny :)

exposes to . In the init block we start to collect the weather flow. In the fragment we start observing the weather . If the orientation changes and the fragment is recreated the console will print because caches the last item.

If we try and rewrite the same code without

there are some consequences.

class WeatherViewModel(
  private val weatherRepository: WeatherRepository
) : ViewModel() {
  val weatherFlow = weatherRepository.getWeather().flowOn(Dispatchers.IO)
}

class WeatherFragment : Fragment() {
  private fun initFlows() {
    lifecycleScope.launchWhenStarted {
      viewModel.weatherFlow.collect { println(it) }
    }
  }
}

Instead of using

to communicate with we are using . The code mostly works in the same way. If the orientation changes however, will be recreated and attempt to collect the weather flow. This will cause the flow to emit all its values again. For this reason I like to use .

Jetbrains are currently working on something called

which will have similar behaviour as . You can find more info about

DataFlow

here

Testing

For testing flows you can use the

operator to get a fixed amount of items from the flow and as the terminal operator to get the result as a list. suspends until it receives the number of items set in the operator or it waits until the flow ends. You can still use instead of but personally I prefer using because you end up with all elements so you have more control of the result.

Example 5:

val ints = flowOf(0, 1, 2)

@Test
fun `confirm 3 ints from flow`() = runBlockingTest {
  val result = ints.toList()
  assertEquals(result.size, 3)
  assertEquals(result[0], 0)
  assertEquals(result[1], 1)
  assertEquals(result[2], 2)
}

The flow should emit three values in total. In the test above we wait for the flow to emit all values and then confirm the length and the values received. Be careful when using only the

operator. If the flow emits an infinite amount of values the function would suspend forever. To prevent this from happening we have the operator to set the number of values that we want to test.

@Test
fun `confirm 2 ints from flow`() = runBlockingTest {
  val result = ints.take(2).toList()
  assertEquals(result.size, 2)
  assertEquals(result[0], 0)
  assertEquals(result[1], 1)
}

Perfect!

Error handling

If you do not handle errors in your flow your app will crash like any other suspend function. You can use the traditional way of wrapping the flow with a

or you can use the operator to catch exceptions thrown upstream. One thing to note is that it doesn't handle exceptions downstream. If the block throws an exception, your app will crash. To get around this we can use and together to collect all upstream exceptions.

Example 6:

scope.launch {
  dataFlow
    .onEach { /* Handle emitted value */ }
    .catch { logError(it) /* Handle upstream exceptions */ }
    .collect()
}

If an error is thrown the flow will still stop emitting new values. If it should continue after an exception you need to handle this case manually.

Conclusion

This was a short intro to

. The API is lightweight and makes it hard to leak subscriptions, which is a problem in . We get this for free with structured concurrency. No more silent resource leaks! and already have support for making it easy to get started. If you want to read more about you can find more resources below.

References:

https://medium.com/@elizarov/reactive-streams-and-kotlin-flows-bfd12772cda4 https://kotlinlang.org/docs/reference/coroutines/flow.html#asynchronous-flow https://medium.com/@elizarov/reactive-streams-and-kotlin-flows-bfd12772cda4 https://medium.com/@elizarov/cold-flows-hot-channels-d74769805f9 https://medium.com/@elizarov/exceptions-in-kotlin-flows-b59643c940fb https://medium.com/@elizarov/execution-context-of-kotlin-flows-b8c151c9309b https://medium.com/@elizarov/simple-design-of-kotlin-flow-4725e7398c4c https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/index.html