MANUEL VIVO: Suspend
functions in Coroutines are meant for one-shot calls
that could return a result. But how could you
create a stream of data or return multiple
results over time? With Flow. Yes, you nailed it. This video builds up on "The
ABC of Coroutines" episode. If you're a beginner,
I would recommend watching that one
first, as today we are going with the Flow. We'll cover the differences
between suspend functions and Flow, the different entities
involved in streams of data, and all the async
possibilities that unfold when you use this powerful API. If you learn something
new, like the video and subscribe to the channel,
but only if you think we've earned it. [MUSIC PLAYING] In Coroutines, a
Flow is a type that can emit multiple
values sequentially, as opposed to suspend
functions that return only a single value. For example, you can use a
Flow to receive live updates from a database. Here, we have the suspend
function loadData that returns an object of type Data. In the UI layer, we can call
that function from a Coroutine, for example, created
using a UI scope. When loadData returns, we
update the UI with its result. This is a one-shot call. You call a suspend function,
you get the result back. That's it. But a Flow is different. Flow is built on
top of Coroutines. And since a Flow can emit
multiple values sequentially, it is conceptually a
stream of data whose values can be computed asynchronously. Here, we have a function that
returns a Flow of type Data, meaning that this Flow is
able to emit data objects. Notice that this function is
no longer a suspend function. To trigger the Flow,
we call collect, which is suspend
function and therefore needs to be called
from a Coroutine. In the lambda, we specify
what to do when we receive an element from the Flow. When a new element is
emitted to the Flow, updateUi will be called
with the new value. New values will be
processed until there are no more elements to emit to
the Flow, or the UI goes away. What we've seen is
a stream of data. There are three
entities involved here. The producer produces data
that is added to the stream. Thanks to Coroutines,
Flows can also produce data asynchronously. The consumer consumes the
values from the stream. In the previous example,
updateUi was the consumer. But there can also
be intermediaries that can modify
each value emitted into the stream or
the stream itself. For example, this
intermediary changes the type of the emitted elements. The consumer, in this case,
will receive a different type. In Android, a data
source or repository is typically a
producer of UI data that has the view model,
or view, as the consumer. Other times, the view layer is
a producer of user input events and other layers of the
hierarchy consume them. Layers in between the
producer and consumer usually act as intermediaries
that modify the stream of data to adjust it to the requirements
of the following layer. Now let's start
with the producer and see how to create a Flow. The builder function Flow
creates a new Flow where you can manually emit new
values into a stream of data using the Emit function. In the following
example, we'll see an app that fetches the
latest news periodically. Here, NewsRemoteDataSource
has their property latestNews that returns a flow of
Lists of ArticleHeadline. As a single suspend function
cannot return multiple consecutive values, the
data source needs to create and return a Flow in order
to fulfill the requirement of having to notify all the
latest news every so often. In this case, the data
source acts as the producer. NewsRemoteDataSource
takes NewsApi as a dependency, which
is the class that ultimately makes the
network request exposing a suspend function. Cool. So what's the implementation
of the latestNews property? We'll use the builder function
Flow to create a new Flow. As the block of code
will be executed in the context of a Coroutine,
it can call suspend functions. To repeatedly make
network requests, we create a while loop. Inside it, we call the API
to get the latest news. And with its result, we
call emit to add that object to the Flow. Flows are sequential. And as the producer
is in a Coroutine, when calling a suspend
function, the producer suspends until the
suspend function returns. In this case, the Coroutine
suspends until fetchLatestNews returns the response. Only then emit is called. To make requests on
a fixed interval, we can call delay with a refresh
interval passed as a parameter. Delay is a suspend
function that suspends the Coroutine for some time. After that time, the
Coroutine will resume, and another iteration in
the while loop will happen. Something to keep in mind is
that with a Flow Coroutine builder the producer cannot
emit values from a different Coroutine context. Therefore, don't call
emit in your Coroutines or in with context
blocks of code. You can use other Flow
builders, such as Callback Flow that we'll cover
letter, for these cases. Intermediaries can use
intermediate operators to modify the stream of data
without consuming the values. These operators are
functions that, when applied to a stream
of data, set up a change of
operations that aren't executed until the values
are consumed in the future. Continuing our example, we
have this NewsRepository. In its constructor, it takes
the NewsRemoteDataSource class that we've seen
before, as well as UserData to know
more information about the logged in user. It exposes this
favouriteLatestNews latest news property of type Flow of
a List of ArticleHeadline. With this, we want to expose
just the news articles that are relevant to the user. This function accesses
the latestNews property from newsRemoteDataSource. And then it applies
the map operator to modify the stream of data. We use filter on
the list that has been emitted to choose those
articles whose topic the user is interested in. The transformation
happens on the Flow. And the consumer will
see the filtered list instead of the original one. Also, we use onEach
as a side effect to save the favorite
user news in the cache. Intermediate operators can be
applied one after the other, forming a chain
of operations that are executed lazily when an
item is emitted into the Flow. Note that simply applying
an intermediate operator to a stream does not
start the Flow collection. To trigger the Flow and
start listening for values, use a terminal operator. With collect, you get all
values at the time they are emitted into the stream. Now, in our view model-- in this case, the
LatestNewsViewModel-- we want to consume that Flow
to get notified of the news and update the UI accordingly. In there, we can call collect
that triggers the Flow and starts using it for values. The lambda will be executed
on every new value received. But, as we said, collect
is a suspend function. Therefore, it needs to be
executed within a Coroutine that we can create with the
built-in ViewModelScope. So what's really happening here? When the ViewModel
is created, we create a new Coroutine
to collect the results from favouriteLatestNews. This triggers the Flow
in the data source layer, which will start
fetching the latest news from the network. All emissions are modified
by the map operator in the repository layer to grab
the user's favorite topics. After that, the repository will
save that info in the cache. And the view model will get the
latest filtered information. As the producer remains always
active with the while loop, the stream of data
will be closed when the ViewModel is cleared
and ViewModelScope is canceled. There are two ways Flow
collection can stop. One way is when the producer
finishes emitting items. The stream of data is closed. And the Coroutine that called
collect will resume executing. Or alternatively, the
Coroutine that collects is canceled, as in our example. This will also stop the
underlying producer. Flows are cold and
lazy unless specified with other
intermediate operators. This means that
the producer code will be executed
each time a terminal operator is called on the Flow. In the example, multiple
collectors of the Flow makes the data source to fetch
the latest news multiple times on different fixed intervals. See the shareIn operator
to optimize and share the Flow when multiple consumers
collect at the same time. The implementation
of the producer can come from a
third-party library. And as such, it could throw
unexpected exceptions. To handle these
exceptions, use the catch intermediate operator. Again, in the ViewModel layer,
to catch unexpected exceptions, we can use the catch operator
to handle them and show the right message to the user. As catch is an
intermediate operator, it needs to be called
before collect. But catch can also
emit items to the Flow. If we wanted to handle
those unexpected exceptions in a repository layer, we
can use the catch operator and emit to the Flow using the
emit function with the latest cached news. If we talk about
Android specifically, Flow is integrated in
many Jetpack libraries. And it's popular among Android
third-party libraries as well. Flow is a great fit
for data updates. For example, you can
use Flow with Room to be notified of
changes in your database. As shown in the code
snippet, in the Dao, return a Flow type
to get live updates. Every time there is a
change in the Example table, a new List is emitted with
the items in the database. At this point, you pretty
much know everything you need about Flows. However, there is
another Flow builder I want to talk about, as
it is used quite often. And that is callbackFlow
that lets you convert callback-based APIs into flows. As an example, the Firebase
Firestore Android APIs use callbacks. Let's see how to convert those
callbacks to flow and listen for Firestore database updates. Here, we have a
FirestoreUserEventsDataSource, whose getUserEvents method
returns a Flow of UserEvents. As you can see, we take an
instance of FirebaseFirestore as a dependency. To create the flow, we
use the callbackFlow API. As with the Flow
builder API, here, we are in the context
of a Coroutine. But, unlike the Flow
builder, channelFlow allows values to be emitted from
a different Coroutine context or outside a Coroutine with
the offer method, as we'll see. The first thing to do is
initialize in Firebase and getting the
eventsCollection from it. Therefore, we write
something like this code. However, this could
potentially throw an exception if it fails getting
the eventsCollection. If that happens, or Firebase
cannot be initialized, we need to close the Flow. If Firebase can be
initialized and we're able to get the
eventsCollection, we need to add a callback
using addSnapshotListener. The callback lambda
will be executed every time there is a
change to eventsCollection. In there, we check if the
snapshot of the collection that we receive is new or not. And if not, we call offer
to emit an item to the flow. Offer is not a suspend function. That's why we can call it
from inside a callback. So far, we initialized Firestore
and added the subscription. What's next? Now we want to keep
the communication open with the consumer of the
flow so that it can receive all events sent by Firestore. For that, we use the
awaitClose method, which will wait until the
flow is closed or canceled. When that happens, the callback
inside awaitClose gets called. In our case, we remove the
subscription from Firebase. Yay, you made it to the end. Hope you learned a
lot in this video. Namely, what
problems Flow solve, how you can create
Flows and observe them, and how powerful
they can be with the intermediate operators. You can learn more about Flow
in our developer.android.com documentation. Thanks for watching. And go write better
Android apps with Kotlin. [MUSIC PLAYING]
