12. SharedFlow & StateFlow

Written by Filip Babić

Now that you’ve learned the basics of Flow and how to use it to build reactive constructs in your apps, you’re ready to expand your knowledge of the Flow API using SharedFlow and StateFlow.

These two types of Flow cater to two common use cases:

• Implementing a broadcast mechanism, which takes any data sent to a Flow and shares it with all collectors simultaneously.
• Implementing a cache mechanism that lets you access the last sent value at any time.

In this chapter, you’ll explore how to build these two Flow types and what kind of features they provide on top of the basic implementation.

Getting Started

To begin the project for this chapter, open the starter project using IntelliJ. Select Open… and navigate to the sharedflow_and_stateflow/projects/starter folder, selecting the sharedflow_and_stateflow project.

Once the project loads, find Main.kt and open it. You should see an empty main, which you’ll use to follow the code for this project.

Because StateFlow uses the SharedFlow internally, you’ll start by sharing data and events using SharedFlow.

Sharing a Flow

Sharing data with layers and services is quite common in large applications. Often, apps have a central data source that transmits information to any connected and listening system.

In Android, similar mechanisms are called broadcasts and broadcast receivers. But in reality, it’s a simple fan-out approach. This is a data communication approach where one source of truth sends events to many observers.

Create a SharedFlow and see how it works. Add the following code to main in the project you opened:

val sharedFlow = MutableSharedFlow<Int>(replay = 2)

This simple snippet creates a MutableSharedFlow with the replay count as 2. That means this SharedFlow can be mutated by sending new events to it and that, in the case of “late” subscribers, it can replay the last two events it processed and emitted to the rest of the subscribers.

Open the MutableSharedFlow signature to see more about what this builder function does internally:

public fun <T> MutableSharedFlow(
    replay: Int = 0,
    extraBufferCapacity: Int = 0,
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND
): MutableSharedFlow<T>

The builder can take in the following information:

• replay: Number of events replayed to all subsequent subscribers.
• extraBufferCapacity: Capacity to hold extra values on top of the replay count.
• onBufferOverflow: The strategy to use when the buffer is filled and more events are coming in. By default, this will suspend the emit values until the buffer is free to add more events.

It’s pretty straightforward in what it allows you to customize. Internally, the mechanism is fairly complex. If you want to dive deeper into it, explore the method more and see how it works.

Connect a couple of subscribers to the SharedFlow. Change the code in main to the following:

fun main() {
  val sharedFlow = MutableSharedFlow<Int>(2)

  // 1
  sharedFlow.onEach {
    println("Emitting: $it")
  }.launchIn(GlobalScope) // 2

  sharedFlow.onEach {
    println("Hello: $it")
  }.launchIn(GlobalScope)

  // 3
  sharedFlow.tryEmit(5)
  sharedFlow.tryEmit(3)

  // 4
  Thread.sleep(50)
}

There are a few steps here:

1. You use the originating Flow to transform it into a new one with onEach. This operator creates a Flow that runs the specified lambda function for each item the SharedFlow emits. What’s important here is that the Flow currently won’t execute anything because it hasn’t been consumed yet.

2. To consume the Flow and subscribe to its events, you use launchIn. This launches the Flow operation in a given CoroutineScope and starts observing the events.

3. Once you’ve set up the subscribers, you emit two values using tryEmit. This will try to send new events to the stream, but if the buffer is filled, it will act according to the BufferOverflow strategy.

4. Finally, you sleep the thread for a few milliseconds to let the program run and print the values.

Build and run. You should see something like the following:

Hello: 5
Emitting: 5
Hello: 3
Emitting: 3

Note: The order of subscribers emitting values isn’t guaranteed because it’s based on how the system schedules these events using threads.

If you run the previous code multiple times, you could get the output messages in a different order, but the first message will always be the value 5. This might be a bit confusing, but this is what’s happening:

1. When you invoke tryEmit(5), you update the value in the SharedFlow. This happens in the main thread.
2. Now, the value 5 is in the SharedFlow and the two Flows you created in 1 are ready to emit the value 5. But you don’t know which one will be first because it depends on thread scheduling. For this reason, the first message will always be one of Hello: 5 or Emitting: 5.
3. You invoke tryEmit(3), adding a new value in the SharedFlow. Again, the two Flows you created earlier are now ready to emit the value 3. Whatever the order, the value 3 for a given Flow will always be emitted after the related 5. For both the Hello and Emitting prefixes, the message with the value 5 will always be before the one with the value 3. So Hello: 5 will always be before Hello: 3 and Emitting: 5 will always be before Emitting: 3.

Each subscriber sequentially consumes and prints the values. If you were to add more subscribers, they’d all print the values you send, which is pretty cool.

If you have a single source of truth in your app, this is a great way to ensure all app modules are up to date with the latest information.

Replaying Values

Now that you’ve created the basic relationship between the Flow and its subscribers, it’s time to take advantage of the internal mechanisms the SharedFlow provides to cache and replay the events.

Change the main code to the following:

fun main() {
  val sharedFlow = MutableSharedFlow<Int>(2)

  // Emitting events before subscribing
  sharedFlow.tryEmit(5)
  sharedFlow.tryEmit(3)

  sharedFlow.onEach {
    println("Emitting: $it")
  }.launchIn(GlobalScope)

  sharedFlow.onEach {
    println("Hello: $it")
  }.launchIn(GlobalScope)

  Thread.sleep(50)
}

In this scenario, the setup is mostly the same, with one major difference: You subscribe to events after you send a couple of them.

Build and run. One of the possible outputs is the following, with the same earlier consideration about the order:

Emitting: 5
Hello: 5
Emitting: 3
Hello: 3

The reason is that you used replay = 2 when building the Flow. This means it’ll cache the last two events at all times for all subscribers that come after the fact. In your case, it lets the two subscribers you create consume the last two events and achieve the same output as if you subscribed before emitting them.

To prove this logic works for the last two events, change the code snippet to the following:

fun main() {
  val sharedFlow = MutableSharedFlow<Int>(2)

  sharedFlow.tryEmit(5)
  sharedFlow.tryEmit(3)
  // Add a third event
  sharedFlow.tryEmit(1)

  sharedFlow.onEach {
    println("Emitting: $it")
  }.launchIn(GlobalScope)

  sharedFlow.onEach {
    println("Hello: $it")
  }.launchIn(GlobalScope)

  Thread.sleep(50)
}

In this scenario, you’re adding a third value to emit so you can prove the last two values are cached. Build and run. The output should now feature the values 3 and 1 (with the possibility of a different order).

Hello: 3
Emitting: 3
Hello: 1
Emitting: 1

Hot Streams

A significant difference from regular Flows is that a SharedFlow is hot by default. This means that when you create the Flow, it immediately starts working. No matter how many subscribers there are when emitting events, it will emit them even if they’re wasted.

Prove this by changing the code snippet. Change main to the following:

fun main() {
  val sharedFlow = MutableSharedFlow<Int>() // remove the replay count

  sharedFlow.tryEmit(5)
  sharedFlow.tryEmit(3)
  sharedFlow.tryEmit(1)

  sharedFlow.onEach {
    println("Emitting: $it")
  }.launchIn(GlobalScope)

  sharedFlow.onEach {
    println("Hello: $it")
  }.launchIn(GlobalScope)

  Thread.sleep(50)
}

In this scenario, you’ve removed the replay count from the Flow builder, meaning it won’t cache any events it receives for future subscribers. Build and run. Observe the output:

Process finished with exit code 0

Or, rather, a lack thereof. In this case, there are no print statements because the values you emit happen before you subscribe to the Flow. Change the code one more time, to the following:

fun main() {
  // 1
  val coroutineScope = CoroutineScope(Dispatchers.Default)
  val sharedFlow = MutableSharedFlow<Int>()

  sharedFlow.onEach {
    println("Emitting: $it")
  }.launchIn(coroutineScope)

  // 2
  coroutineScope.launch {
    sharedFlow.emit(5)
    sharedFlow.emit(3)
    sharedFlow.emit(1)

     // 3
    coroutineScope.cancel()
  }

  // 4
  while (coroutineScope.isActive) {

  }
}

This code snippet is quite different from the previous examples, so dive into it, one step at a time:

1. To keep the program alive and avoid GlobalScope, you create a custom CoroutineScope instead.
2. You launch a new coroutine to emit three values. emit suspends, making sure the values are sent when the consumers are available to receive them, based on the default buffer overflow strategy.
3. Once the values emit, you cancel the scope to let the program know it can finish after the events are sent and consumed.
4. You keep the program running while coroutineScope.isActive.

Build and run. You’ll go back to the default implementation, where the three values are emitted and consumed. The output should be the following:

Emitting: 5
Emitting: 3
Emitting: 1

So be very mindful when creating and subscribing to SharedFlows because the order of events and the program’s lifecycle is quite important. If you don’t use a replay count and your subscribers don’t subscribe right away, you might lose events that no one can consume.

Now that you’ve learned how to create and consume SharedFlows the typical way, it’s time to learn how to create them using transformations.

Transforming a Flow to a SharedFlow

An alternative to creating a MutableSharedFlow is to start with a regular Flow and use shareIn to transform it and allow the fan-out behavior. Change the main snippet to the following:

fun main() {
  val coroutineScope = CoroutineScope(Dispatchers.Default)
  // 1
  val sharedFlow = flow {
    emit(5)
    emit(3)
    emit(1)

    Thread.sleep(50)
    coroutineScope.cancel()
  }.shareIn(coroutineScope, started = SharingStarted.Lazily) // 2

  sharedFlow.onEach {
    println("Emitting: $it")
  }.launchIn(coroutineScope)

  while (coroutineScope.isActive) {

  }
}

This snippet uses shareIn to transform a basic Flow into a SharedFlow. There are two key differences here in how this snippet works compared with the previous implementation:

1. Instead of using MutableSharedFlow() and building it manually, then emitting values, you create a basic Flow here using the default builder. You also emit the values from within the Flow rather than through a launch block.
2. Using shareIn, you transform this Flow into a SharedFlow. The operator takes in two parameters: the CoroutineScope in which you’ll share the data and events, and the SharingStarted parameter that defines whether the SharedFlow will start emitting values right away or wait for its first subscriber. You can use Eagerly to start emitting immediately rather than waiting for a subscriber.

Build and run. You should see the following output:

Emitting: 5
Emitting: 3
Emitting: 1

Process finished with exit code 0

Everything works like before, with a slightly different setup. You can use this approach if you have a factory function that provides a source of truth for your data. But you need to share its values with multiple subscribers.

SharedFlow Notes

There are a few last things to learn about the SharedFlow.

One key aspect of a SharedFlow is that it never completes. Because it represents a possibly infinite stream of new information shared across multiple subscribers, it’s understandable that completing such a Flow wouldn’t make sense.

It represents an uncommon scenario where you don’t expect a finite set of events that end with a completion signal but rather a continuous stream that feeds information until the subscribers no longer need it.

For this reason, it’s quite powerful. But you also have to make sure you close the Flow as soon as you don’t need it!

Because the SharedFlow never completes, it’s important to know that using some operators has no effect. These operators are usually the ones that change the dispatcher or context of the Flow, like flowOn, cancellable or by creating a new Flow using buffered. All these will have no effect with a SharedFlow.

The SharedFlow was also envisioned as a replacement for the BroadcastChannel because it’s built in a safer, more configurable and clear way, unlike the Channel API. Because of this, it’s still being worked on and expanded. It’s not suitable for inheritance because the interface will probably change in the future, so be mindful of that if you decide to build a custom implementation of the SharedFlow.

Finally, subscribing to the SharedFlow doesn’t affect the performance or add any overhead. But having multiple subscribers means data emitting events will have a worse case O(N) effect because every subscriber will have to receive and consume the event.

So be mindful of having too many subscribers that consume events in an operation-heavy way, such as doing data transformations, loading more data when the events arrive or similar operations.

You’ve learned a lot about SharedFlow, which is important because it forms the foundation for how StateFlow works. You’ll explore that next!

Building a StateFlow

An even more advanced version of a SharedFlow is the StateFlow. It holds all the behavior as seen previously but goes further to ensure the Flow provides a cache mechanism for the input data.

This means you can access the last sent and stored data through a value accessor. First, see how to create a StateFlow. Replace the code in main with the following:

fun main() {
  val coroutineScope = CoroutineScope(Dispatchers.Default)

  val stateFlow = MutableStateFlow("Author: Filip") // here
  
  while (coroutineScope.isActive) {

  }
}

As before, you’re keeping a CoroutineScope around to keep the program running while you test the Flow. What’s important is the way you’re creating the StateFlow. You use a simple builder called MutableStateFlow(initialValue: T), which lets you define the value you start the Flow with.

Because StateFlow keeps the last value cached, you can define which value you want to start with or choose to start with null.

The next step is to observe the value and subscribe to the Flow. Add the following code after creating the stateFlow:

println(stateFlow.value) // 1

coroutineScope.launch {
  stateFlow.collect { // 2
    println(it)
  }
}

In this snippet, you access the StateFlow in two ways:

1. Directly accessing value gives you the data stored in the StateFlow at any given moment. This is an easy way to access values without subscribing to changes. But it’s much safer to subscribe to the Flow instead because the values can change quickly based on your data source.
2. Subscribing to value changes using collect and consuming each value the same way. This is the recommended way of accessing data from a StateFlow because you’ll receive constant updates from your data source.

Build and run. You should see the following output in the terminal:

Author: Filip
Author: Filip

Make sure to close the program in the terminal after seeing the output by clicking the Stop button, which looks like a big red square.

Now that you’ve learned how to access these values, it’s time to see how to change them and what happens with the output. Add the following snippet of code after launch and before the while loop:

stateFlow.value = "Author: Luka" // 1

stateFlow.tryEmit("FPE: Max") // 2

coroutineScope.launch {
  stateFlow.emit("TE: Godfred") // 3
}

Thread.sleep(50)
coroutineScope.cancel()

A few things are happening here. You update the value of the StateFlow in three different ways before sleeping the thread for a few milliseconds and canceling the CoroutineScope. Before you learn the three methods of updating state here, build and run. You should see the following output:

Author: Filip
FPE: Max
TE: Godfred

There seems to be an issue, right? You have three value changes and one initial value, but only three items are being printed. The reason for this lies in the way you’re updating the values.

To update the value, you can:

1. Change the value of the StateFlow directly. This approach doesn’t guarantee the value change or emission, but you can use it when you’re sure there’s only one place you update the state from.
2. Use tryEmit from anywhere and try to send a new value to the Flow without blocking or suspending. This is the safest way to update the value. But if there are multiple value changes, it might not emit the new piece of data.
3. Use emit within a coroutine or another suspend function, suspending if there’s a buffer overload. This approach is more certain to emit a value in the end due to the suspending mechanism.

Because of these three rules and ways to update the data, you can see the “Author: Luka” value doesn’t get emitted. tryEmit and emit take precedence.

Now, you know how to deal with advanced variants of the Flow construct. You can use this knowledge to power various types of software, such as UI-based apps, broadcast or event systems, database communication and much more.

StateFlow Notes

Like SharedFlow has a few important notes about it, so does StateFlow, given that it’s largely built on top of the SharedFlow API.

SharedFlow can’t fail or complete because it serves the same broadcast type of communication. It keeps sending information until you’re ready to cancel the flow or the CoroutineScope. So make sure to clean up the Flow when you’re ready to stop the events.

Additionally, StateFlow was based on the Channel API, specifically the ConflatedBroadcastChannel, which is now obsolete. In that sense, it provides a value you can safely read and update in multiple ways.

Further, some operators don’t affect a StateFlow, such as distinctUntilChanged, flowOn and buffer with specific strategies.

Finally, the StateFlow API and signature might change in the future to add more behavior, so inheriting and building a custom implementations of the API isn’t recommended.

Key Points

• You can use Flows for non-finite use cases where they represent a constant stream of events.

• Most common stream use cases are fan-out or broadcast mechanisms that may or may not have cached data built-in.

• Advanced Flows that represent these use cases are SharedFlow and StateFlow.

• SharedFlow is an alternative to Flow that allows multiple subscribers. Each subscriber receives events simultaneously from the moment they subscribe.

• SharedFlows are hot by default, so they don’t wait for subscribers to start emitting data.

• You can create a SharedFlow using the constructor MutableSharedFlow() or by transforming an existing Flow using the shareIn operator.

• A SharedFlow will never complete, as part of its design choice.

• Closing a SharedFlow is recommended for optimization when you no longer need to emit and consume events.

• Some operators don’t affect a SharedFlow, such as buffer, flowOn and cancellable.

• Be mindful of the number of subscribers or their subscribe blocks when using a SharedFlow. Adding more subscribers doesn’t create any overhead in itself. But each subscriber has to process every event sequentially.

• StateFlow is largely based on the SharedFlow API and provides similar functionality.

• One key aspect of a StateFlow is it always caches the last value it receives. You can use this value through the stateFlow.value accessor.

• It’s recommended to change the state of the StateFlow using functions like emit and tryEmit rather than using the value accessor directly.

• Almost the same rules and notes apply to a StateFlow as they do to SharedFlow regarding the base behavior and implementation.

Where to Go From Here?

You’ve learned everything you need to know about the Flow API, an excellent implementation of the observable/observer pattern and reactive streams!

The reason why Flows are so good is that they use coroutines under the hood, unlike RxJava. This allows the system to suspend both the subscriber and the producer of values to not lose values or cause too much processing overhead.

If you’re keen on learning more about these two types of Flows, specifically in Android, check out the official Android documentation.

You’ll also learn about these concepts in more depth in the last section of the book so keep reading to learn more about Kotlin Coroutines and how to use them in Android!