14. Beginning with Coroutine Flow

Written by Filip Babić

Coroutines are amazing when it comes to bridging the synchronous and asynchronous worlds, to return values and communicate between threads. Most of the time that’s what you want and need, but sometimes, computer systems require you to consume multiple values over a period of time.

And there are two different ways you can do this - using sequences and streams. However there are certain limitations to both approaches. You’ve already learned about sequences, but they force you to block the calling thread when you’re observing values. So let’s see what streams have to offer, and how they behave in code.

Streams of data

One of the key similarities between sequences and streams is that both constructs can generate an infinite amount of elements. Sequences usually do this by defining an operation which you run behind the scenes to build a value.

This is also the key difference between streams and sequences, as you usually build streams using a function or their constructor. You then have an interface between the provider of values and a consumer, exposing a different part of the interface to each side.

Take this snippet for example, which uses the Reactive Extensions, or Rx, version of observable streams of data:

val subject = BehaviorSubject.create<String>()

subject.subscribe(observer)
subject.onNext("one")
subject.onNext("two")
subject.onNext("three")

You create a Subject, which implements both sides of the stream interface. The provider can use functions such as offer(), onNext(), and send(), to fill the queue for the stream with values to consume. In this case it’s using onNext() from Rx.

Every Observer which subscribes to this stream will receive all its events, from the moment they subscribed, until they unsubscribe, or the stream closes. The observer in Rx will look like this:

val observer = object: Observer<String> {
  
  override fun onNext(value: String) {
    // consume the value
  }
  
  override fun onError(throwable: Throwable) {
    // handle the error
  }
  
  override fun onComplete() {
    // the stream completed
  }
}

Every time you send any of the events to the Observable side of the Subject, it will send all those events to all of its Observers. It acts as a relay of data from one central point to multiple observing nodes. This is the general idea of streams. Being observable and sending the events to every single Observer which is listening to its data.

But, depending on the implementation of streams, you might have a different setup. One of the things each stream mechanism and implementation shares is the type of streams and when their values are propagated. As such, there are hot and cold streams of data. Let’s consume them one at a time.

Hot streams

Hot streams behave just like TV channels, or radio stations. They keep sending events, and emitting their data, even though no one may be listening or watching the show. It’s why they are called hot. As they don’t care if there are any observers, they will keep on working and computing no matter what, from the moment you create them, until they close.

This is really good when you want values computed in the background fast, preparing them for multiple observers you already have waiting. But if you’re going to be adding observers after the fact, then you could lose the data a hot stream might emit between the computation and the observers starting to listen to events.

Additionally, if the producer of values is hot, it can keep on producing values, even though there are no consumers. This effectively wastes resources, and you have to manually close the stream if you stop using it.

If you were to use coroutines to build such hot streams, you’d use the Channel API. They are hot by default and also support coroutines, making them a bit less leak-prone. But even if you used structured concurrency and coroutines within the Channels, you could potentially leak some resources until the CoroutineScope cancels.

This is why the idea of having a cold stream is important.

Cold streams

It makes sense that, if hot streams are computed right away, and work even without any observers, cold streams do the opposite. They are like the builder pattern, where you define a set of behaviors for a construct, upfront, and only when you call a finalizing function, does it become live and active.

Given that, cold streams are like a social event. You can prepare everything upfront, think of each specific detail you have to fulfill and organize, and only when you’re certain that people are coming, does the event happen. Following the analogy, cold streams won’t produce or send values, until they have an active Observer, to whom they can emit the events.

This is much better, because if there are no observers, there is no need to execute a potentailly heavy operation to produce a value. But if there is at least one observer, then you will compute the value and pass it down the stream, to any amount of consumers.

It sounds too good to be true, and there’s a reason why hot and cold streams aren’t used everywhere, and for every occasion. It’s because streams have a lot of internal limitations, and there are a lot of features a good stream should support to be versatile.

Limitations of streams

In every-day programming, there are certain limitations to the way things should operate for optimal use. You don’t want to waste resources, freeze the UI, lose data and so on. Some of these concepts apply to streams, as well. These limitations revolve around the same problem - the speed of producing and consuming the values.

If your producer is sending out too many values, and the consumer cannot process them fast enough, then you’re bound to lose some data. To effectively process the values, you have to apply backpresure. This is the technical term for eliminating the bottleneck in the producer-consumer pair.

When the producer queue fills up, and the consumer can’t process the values fast enough, you become bottlenecked from the consumer side. If however, the consumer is eating the values too fast, and it keeps waiting for more to be produced, you’re bottlenecked on the producer side. Either way, one side has to halt - block, until the pair is balanced again.

Supporting backpressure

As you’ve learned, if one side of the producer-consumer pair is too fast or too slow in its job, you will lose data, or end up blocking the non-bottlenecked side. Unless you add backpressure support.

Backpressure can be achieved in different ways, such as using buffered underlying streams with a fixed capacity. This is the easiest solution, but also the most error prone, because you can easily use up a lot of computer memory, or even overflow the buffer. This, again, will cause a bottleneck, and you’ll lose data. You could have the capacitiy of unlimited, but then you risk overflowing the memory.

Another way is to build a synchronization mechanism, where you’d pause and resume threads as bottlenecks occur, but this may be even worse, as you could be freezing threads for a long time, which is a waste of resources in the end. This is why it’s important to avoid blocking threads, when building streams with backpressure. Because of this design requirement, the Flow API is a fresh new take on streams.

A new approach to streams

Having the best of both worlds, the Flow API supports cold, asynchronously-built streams of values, where the thread communication and backpressure support is implemented through the use of coroutines. When you think about it, it’s the perfect combination.

Having coroutines at its foundation allows for backpressure-by-design. If your producer is overflowing the consumer, then you can suspend the producer, until you free up the queue of events you need to process. On the other hand, if your consumer is really fast, and you need to slow it down - introduce a delay, or a debounce period, all you have to do is apply the same logic - suspend the consumer, until it meets your conditions.

The other happy coincidence of coroutines is the built-in context switching. By abstracting away threading and dispatching, through the use of CoroutineContexts, you can easily switch the consumption of events from one thread to another, by passing in a different CoroutineContext from a Dispatchers. And it’s performant, because you don’t have to worry about thread allocation, since coroutines use predefined thread pools.

This seems a bit too good to be true, right? It feels as if the API will be very complicated, because it has to handle all those details a regular stream cannot intrinsically implement.

Well that’s where the fun kicks in. Weirdly enough, Flow works based on only two interfaces - the Flow and the FlowCollector. For the sake of comparison, if you’re coming from a reactive-driven world, the Flow would be like an Observable, whereas the FlowCollector would be something similar to an Observer - a subscriber to events.

Let’s examine how they work.

Building Flows

To create a Flow, just like with standard coroutines, you have to use a builder. But first open up Main.kt, in the starter project, which you can find by navigating to the project files, the starter folder, and opening the beginning_with_coroutines_flow folder.

Next, find main(). It should be empty, but you’re about to add the code it needs to build a Flow. Add the following snippet to main(), so it doesn’t look so empty. Conveniently enough, Flow’s builder function is called flow:

val flowOfStrings = flow {
  for (number in 0..100) {
    emit("Emitting: $number")
  }
}

This snippet of code will build a Flow<String>, which calls emit() a hundred times, sending a String value to every observer which starts listening to the data. And to do that, you must call collect(). Add the following snippet under the flow() call:

GlobalScope.launch {
  flowOfStrings.collect { value ->
    println(value)
  }
}

Thread.sleep(1000)

collect() is a suspending function, and as such needs to be called from a coroutine, or another suspending function. From within, you have access to every single value you emit from within the Flow builder. In this case, you’re consuming each value by printing it out.

Now, to understand how Flows work from within, check the builder definition:

public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> 
  = SafeFlow(block)

You create a Flow<T>, with BuilderInference, meaning that, just like with producers and actors, you devise the generic type from within the function constructor. Futhermore, the lambda block is of the type FlowCollector<T>.() -> Unit, meaning that the internal scope of the lambda will be a FlowCollector. This is great as you can both create a Flow, and emit() values directly to the collector. You have the entire API connected in one place, making it very simple and clean to use.

In one of the previous snippets, you collected the values from a Flow, but there’s much more you can do with the Flow, before you consume the data.

Collecting and transforming values

Once you build a Flow, you can do many things with the stream, before the values reach the FlowCollector. Just like with Rx, or with collections in Kotlin, you can transform the values using operators like map(), flatMap(), reduce() and much more. Additionally, you can use operators like debounce(), delayFlow() and delayEach() to apply backpressure or delays manually for each item, or the entire Flow.

Take the following snippet for example:

GlobalScope.launch {
  flowOfStrings
      .map { it.split(" ") }
      .map { it.last() }
      .delayEach(100)
      .collect { value ->
        println(value)
      }
}

If you replace the previous way of consuming the Flow, and run main() again, you’ll now see the values are mapped back to the actual numbers, after the String is split. Furthermore, you print each value with a small delay, ultimately suspending the Flow until the consumer is ready.

Note: All of the operators above are marked with suspend, so you have to call them from within a coroutine or another suspending function. This keeps the API uniform, as Flows are built upon coroutines.

Switching the context

Another thing you can do with Flow events, is switch the context in which you’ll consume them. To do that, you have to call flowOn(context: CoroutineContext), just like this:

GlobalScope.launch {
  flowOfStrings
    .map { it.split(" ") }
    .map { it.last() }
    .flowOn(Dispatchers.IO)
    .delayEach(100)
    .flowOn(Dispatchers.Default)
    .collect { value ->
      println(value)
    }
}

In this snippet, you’re calling flowOn() twice. The first time after defining the mapping operations, and then the second time after delaying every item for a hundred miliseconds. The real power of applying context switch’s is that you can do it as many times as you want, for each operator you’re calling on the Flow. However, whenever you call flowOn(), you’re applying the context switch only on the preceding operators, as the documentation states:

/**
 * Changes the context where this flow is executed to 
 * the given [context]. This operator is composable 
 * and affects only preceding operators that do not have 
 * its own context.
 * This operator is context preserving: [context] **does not** 
 * leak into the downstream flow.
 ...
 **/

Additionally, as the docs state, the context is not leaked into the downstream flow, and the rest of the Flow operators and chained calls do not know about the context switch, nor can they abuse the previous CoroutineContext.

Ultimately, it’s important to know that the final consumption of events can happen only on the original context. This means that no matter how many context switches you apply to the Flow, the last context will be the same as the original one.

So if you create a Flow on the main thread, you’ll have to consume the events on it, as well. This is something you have to be careful about, because otherwise, you’ll get an exception, if you try to produce values in a different context than the one you’re consuming events in.

Flow Constraints

Since Flow is really easy to use as-is, there have to be some constraints in order to keep people from abusing or breaking the API. There are two main things which each Flow should adhere to, and each use case should enforce - preserving the context and being transparent with exceptions.

Preserving the Flow context

As mentioned above, you have to be clean when using CoroutineContexts with the Flow API. The producing and consuming contexts have to be the same. This effectively means that you cannot have concurrent value production, because the Flow itself is not thread safe, and doesn’t allow for such emmisions.

So if you try to run the following snippet:

val flowOfStrings = flow {
  for (number in 0..100) {

    GlobalScope.launch {
      emit("Emitting: $number")
    }
  }
}

GlobalScope.launch {
  flowOfStrings.collect()
}

You will receive an exception, saying you can’t change the Flow concurrently.

If you want coroutines to be synchronized, and have the ability to concurrently produce values in the Flow, you can use channelFlow() instead, and offer() or send() to emit the values to the FlowCollector. Changing the code to the following snippet will work:

val flowOfStrings = channelFlow {
  for (number in 0..100) {

    withContext(Dispatchers.IO) {
      offer("Emitting: $number")
    }
  }
}

GlobalScope.launch {
  flowOfStrings.collect()
}

If not, then you should create the Flow values in a non-concurrent way, and then use flowOn(), to switch the Flow to any CoroutineContext you want, if you want to avoid using channelFlow().

Additionally, the Flow‘s CoroutineContext cannot be bound to a Job, and as such you shouldn’t combine any Jobs with the context you’re trying to switch the Flow to. This is because the Flow shouldn’t be something that’s lifecycle-aware, and something which can be cancelled. Especially because you can effectively mix multiple CoroutineContexts using flowOn(), and introducing a Job can only break things, or make them unsafe.

Being transparent with exceptions

When dealing with exceptions in coroutines, it’s relatively easy to bury them down. For example, by using async(), you could effectively receive an exception, but if you never call await(), you’re not going to throw it for the coroutines to catch. Additionally, if you add a CoroutineExceptionHandler, when exceptions occur in coroutines they get propagated to it, ending the coroutine.

This is why Flow exposes a convenient function which behaves similar to flowOn(). You can use catch(), providing a lambda which will catch any exception you produce in the stream and any of its previous operators. Examine the snippet below:

flowOfStrings
  .map { it.split(" ") }
  .map { it[1] }
  .catch { it.printStackTrace() }
  .flowOn(Dispatchers.Default)
  .collect { println(it) }

Instead of mapping to it.last(), you’re using indices. In case you receive an empty string, this will cause an IndexOutOfBoundsException. But because you’re calling catch(), after map(), if an exception occurs, you’ll catch it, and print its stack trace. This way, you’ll be able to handle any exceptions from the original stream and the operators alike.

Change the way you build Flow, to this:

val flowOfStrings = flow {
  emit("")

  for (number in 0..100) {
    emit("Emitting: $number")
  }
}

You will now cause an exception to be thrown, but you’ll see that the program doesn’t crash. This is because catch() will stop the exception from throwing all the way up to cause your app to crash.

You will still get a stack trace from the exception. Add this line of code under collect() and within the coroutine, to be certain the program continues normally:

println("The code still works!")

Run the code once again. You should now see an exception’s stack trace, and right after that The code still works!. This means catch() has stopped the exception from one of the stream operators from breaking the entire program. And the rest of the coroutine still runs and works like a charm.

In case you’d want to continue emitting values if an exception occurs, you have access to the original FlowCollector within catch(), and it’s advised to simply call emitAll(), with the fallback values. Change the GlobalScope.launch() code to the following:

flowOfStrings
.map { it.split(" ") }
.map { it[1] }
.catch {
  it.printStackTrace()
  // send the fallback value or values
  emit("Fallback")
}
.flowOn(Dispatchers.Default)
.collect { println(it) }

println("The code still works!")

You should now see the exception stack trace printed out, as well as Fallback and The code still works!. This shows you that you can catch exceptions in streams, handle them correctly, and continue the stream with some fallback values. You won’t break the outer coroutine, even if an exception occurs and is caught with catch()!

Key Points

• Sometimes you need to build more than one value asynchronously, this is usually done with sequences or streams.

• Sequences are lazy and cold, but blocking when you need to consume events. It’s better to use and suspend coroutines instead.

• If you build streams using Channels, then you have coroutine support and suspendability, but they are hot by default.

• Being cold means the data isn’t computed, until you start observing. As opposed to being cold, being hot means the data is computed right away, with, or without any observers.

• As such, streams have two sides - the producer, or observable construct, and a consumer, or the observer construct.

• The main limitations of streams are they are blocking and use backpressure.

• Blocking happens when a stream needs to produce or consume events.

• Backpressure is when a stream is producing or consuming events too fast, and one side has to be slowed down, to balance the stream.

• Backpressure is usually done through blocking the thread of a producer or a consumer.

• A good stream avoids blocking, supports context switching while still allowing for backpressure.

• The Flow API is built upon coroutines, allowing for suspending.

• Because you can suspend a consumer or a producer, you get intrinsic backpressure support.

• Additionally, you’re avoiding blocking, which is what a good stream should do.

• To create a Flow, simply call flow(), and provide a way to emit values to the FlowCollectors which decide to process the values.

• To attach a FlowCollector to a Flow, you have to call collect(), with a lambda in which you will consume each of the values.

• collect() is a suspending function, so it has to be within a coroutine or another suspending function.

• You have the access to a FlowCollector from within collect(), so you can emit values.

• Flows can be transformed and mutated by various operators like map(), flatMap().

• You can apply manual backpressure using debounce(), delayEach() and delayFlow().

• Switching the context of a Flow allows you to change the threads in which you consume each piece of data, or perform each operator.

• To switch context, call flowOn(context) after the operators you wish to switch the context of.

• The Flow collects the values always in the context of the CoroutineScope it is located in. So if you call collect() on the main thread, you’ll also consume the values there.

• Flows don’t allow you to produce values concurrently. If you try to do that, an exception will occur.

• If you do need to produce values from multiple threads, you can use channelFlow().

• It’s better to use flowOn() to switch contexts of the Flow, than to bury them down in coroutines.

• Flows should be transparent when it comes to exceptions.

• To handle exceptions with Flows, use the catch() operator.

• catch() will intercept any uncaught exceptions, from all the operators you called before catch() itself.