相关词汇

Suspend functions :挂起函数。告诉kotlin编译器，这个方法需要执行在一个协程中
Coroutine：协程
Producer:提供者
Intermediary :（中介-可选option）中间的 可以修改发送到数据流的值，或修正数据流本身。
Consumer:消费者（使用者）使用数据流中的值
entity:实体（Java been）
emit（）:发射数据项
context:???????上下文web后端开发或UI线程的context；如需更改数据流的 CoroutineContext，请使用中间运算符 flowOn

Flow:流,用于存放异步存放element数据 官方文档Android 上的 Kotlin 数据流
collect（）:收集
map{...} :中间层操作符 ，注意中间运算符可以接连应用，形成链式运算，在数据项被发送到数据流时延迟执行。请注意，仅将一个中间运算符应用于数据流不会启动数据流收集。
其他中间层操作符：除非使用其他中间运算符指定流，否则数据流始终为冷数据并延迟执行。

---

youtube视频链接

演讲人 ：MANUEL VICENTE VIVO
协程中的挂起函数用于一次调用，可以返回一个结果。但是，如何创建一个数据流来随时间返回多个结果呢?在这个Kotlin词汇的视频中，Manuel会给我们带来Kotlin Flow的好处，还有更多!
Going with the flow - Kotlin Vocabulary

推荐另外一个协程视频The ABC of Coroutines - Kotlin Vocabulary

文章

image.png

android上的kotlin数据流Flow--android开发者

---

介绍

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.
但是如何创建数据流或随时间返回多个结果呢? 使用Flow.

Yes,you nailed ti.
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.
如果你是初学者，我建议你先看这部电影，因为今天我们要讲的是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.

本视频我们将讨论suspend函数和Flow之间的区别、不同的entity涉及stream流数据 ，以及在使用这个强大的API时出现的所有异步可能性。
If you learn something new ,like the video and subscribe to the channel, but only if you think we've earned it.
如果你学到了新东西，喜欢这个视频并订阅这个频道，但前提是你认为这是我们应得的。

In Coroutines ,a Flow is a type that can emit multiple values sequentially,as opposed to suspend functions that return only a singgle value.

在协程中，Flow是一种可以连续发出多个值的类型，而不是只返回单个值的挂起函数。

For example ,you can use a Flow to receive live updates from database.
例如，您可以使用Flow从数据库接收实时更新。

suspend function

suspend fun loadData():Data

Here,we have the suspend function loadData returns an object of type Data.
在这里，我们让挂起函数loadData返回一个Data类型的对象。

uiScope.launch{
  val data = loadData()
  updateUi(data)
}

In hte UI layer,we can call that function from a Coroutine,for example ,created using a UI scope.
在UI层，我们可以从协程调用该函数，例如，使用UI作用域创建的协程。
When loadData returns,we update the UI with its result.

当loadData返回时，我们用它的结果更新UI。
This is a one-shot call.
是一次性调用。
You call a suspend function, you get the result back.That is it.
你调用一个挂起函数，就会得到结果。

---

Flow 函数

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.
但是，Flow是不同的。Flow在协程上构建。而且由于Flow可以连续地产生多个值，因此它在概念上是一个数据流，其值可以异步计算。

fun dataStream():Flow<Data>

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.
这里是一个返回Flow类型的函数，返回值Flow<Data>，这意味着该Flow能够emit(发出)数据对象。
注意，这个函数不再是挂起函数。

Flow的collect是一个挂起函数

uiScope.launch{
  dataStream().collect{
    updateUi(data)
  }  
}

To trigger the Flow,we can call collect,which is suspend function and therefore needs to be called from a Coroutine.

要触发Flow，我们可以调用collect，collect是挂起函数，因此需要从协程调用。
In the lambda,we specify what to do when we receive an element from the Flow.
在lambda表达式中，我们指定从Flow接收元素时要做什么。

image.png

When a new element is emitted to the Flow, updateUi will be called with the new value.
当向Flow发出新元素时，updateUi函数会被调用，参数是新的element数据

New values will be processed until there are no more elements to emit to the Flow, or the UI goes away.
新的值将被处理，直到没有更多的element可以发送到Flow，或者UI消失。

stream数据。这里涉及到三个实体。Producer、Consumer、intermeiaries

What we've seen is a stream of data。
There are three entities involved here .
我们看到的是stream数据。这里涉及到三个实体。Producer、Consumer、intermeiaries

image.png

The produrer produces data that is added to the stream.
Thanks to Coroutines ,Flows can also produce data asynchronously.

produrer生成添加到流中的数据。
多亏了协程，流还可以异步地生成数据。
The Consumer consumes the values from the stream.
In the previous example updateUi was the consumer.
Consumer消费流中的值。在前面的示例中，updateUi函数就是Consumer。

But there can also be intermeiaries that can modify each value emitted into the stream or the stream itself.
但是也可以有intermeiary，它们可以修改发送(emit)到流或流本身的每个值。

image.png

For example ,this intermediaries changes the type of the emitted elements.
The consumer, in this case,will receive a different type.
例如，这个intermediary会更改发出的元素的类型。
在这种情况下，消费者(consumer)将收到不同的类型。

在android中描述生产者和消费者

In Android ,a data source or repository is typically a producer of UI data that has the view model, or view,as the consumer.

在Android中，数据源或repository通常是UI数据的生产者，viewModel或view作为消费者。
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.
其他时候，视图层是用户输入事件的生产者，视图层次结构的其他层consume这些输入事件。
生产者和消费者之间的层通常充当修改数据流的中间体(intermediary)，以使其适应下一层的需求。

通过实例介绍Flow

Flow-retrofit简图

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.

现在让我们从producer开始，看看如何创建一个Flow。
构建器函数Flow创建了一个新的Flow，您可以使用emit函数手动将新值发送到数据流中。
In the following example,we'll see an app that fetches the latest news periodically.
在下面的例子中，我们将看到一个定期获取最新消息的应用程序。

class NewsRemoteDataSource(
 private val newsApi:NewsApi
) {
  val latestNews :Flow<List<ArticleHeadline>>{
    ...
  }
}

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.
在这里，NewsRemoteDataSource有其属性latestNews，它返回一个列表的ArticleHeadline流（文章标题）。
由于单个挂起函数不能返回多个连续的值，数据源需要创建并返回一个Flow，以满足必须经常通知所有最新消息的需求。

In this case ,the data source acts as the producer.
在本例中，data source充当生产者。

interface NewsApi{
  suspend fun fetchLatestNews():List<ArticleHeadline>
}
//=======
class NewsRemoteDataSource(
 private val newsApi:NewsApi
) {
  val latestNews :Flow<List<ArticleHeadline>>{
    ...
  }
}

NewsRemoteDatasource takes NewsApi as a dependency,which is the class that ultimately makes the network request exposing a suspend funciton.

NewsRemoteDatasource将NewsApi作为依赖项，NewsApi类是最终使网络请求暴露挂起函数。

class NewsRemoteDataSource(
 private val newsApi:NewsApi
) {
  val latestNews :Flow<List<ArticleHeadline>> = flow {
     //in a coroutine  -- can suspend
  }
}

So what's the implementation of the latestNews property?
We'll use the builder function Flow to create a new Flow.
那么latestNews属性的实现是什么呢?
我们将使用构建器函数Flow来创建一个新的Flow。
As the block of code will be executed in the context of a Coroutine, it can call suspend funcitons.

由于代码块将在Coroutine的上下文中执行，它可以调用suspend函数。

class NewsRemoteDataSource(
 private val newsApi:NewsApi
) {
  val latestNews :Flow<List<ArticleHeadline>> = flow {
     while(ture){
       //todo 数据请求 和发送
     }
  }
}

To repeatedly make network requests,we create a while loop.
为了重复发出网络请求，我们创建了一个while循环。

class NewsRemoteDataSource(
 private val newsApi:NewsApi
) {
  val latestNews :Flow<List<ArticleHeadline>> = flow {
     while(ture){
       val latestNews = newsApi.fetchLatestNews()
       emit(latestNews) //发射发布
     }
  }
}

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.

在它内部，我们调用Api来获取最新的新闻。
然后调用emit将该latestNews对象添加到Flow中。
Flows are sequential.
And as the producer is in a Corountine,when calling a suspend function,the producer suspend until the suspend function returns.
In this case ,the Coroutine suspends until fetchLatestNews returns the response.
Only the emit is called .
流动是连续的。
由于生产者在Corountine中，当调用suspend函数时，生产者会一直挂起直到suspend函数返回。
在例子中，协程挂起直到fetchLatestNews返回响应。
只有emit被调用。

class NewsRemoteDataSource(
 private val newsApi:NewsApi,
 private val refeshInterValMs:Float = 5000f //默认时间间隔
) {
  val latestNews :Flow<List<ArticleHeadline>> = flow {
     while(ture){
       val latestNews = newsApi.fetchLatestNews()
       emit(latestNews) //发射发布
       delay(refeshInterValMs)
     }
  }
}

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.
要按固定的间隔发出请求，我们可以调用delay，并将刷新间隔作为参数传递。
Delay是一个挂起函数，它将协程挂起一段时间。
在此之后，协程将继续执行，而while循环中的另一次迭代将会发生。

Flow中的CallbackFlow构造器

Something to keep in mind is that with a Flow Coroutine builder the producer cannot emit values form a different Coroutine context.
需要记住的是，在Flow协程构建器中，生成器不能发出形成不同协程上下文的值。
Therefore,don't call emit in your Coroutines or in with context blocks of code.
因此，不要在你的协程或代码的上下文块中调用emit。

You can use other Flow builders ,such as Callback Flow that we'll cover letter,for these cases.
对于这些情况，您可以使用其他流构建器，例如我们将介绍的CallbackFlow。
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.
这些操作符是一些函数，当应用于数据流时，它们会设置操作的更改，这些更改直到将来消费这些值时才会执行。

class NewsRepository(
 private val newsRemoteDataSource: NewsRemoteDataSource,
 private val userData ： UserData //比如用户登陆返回的用户信息
) {
  val favouriteLatestNews : Flow<List<ArticleHeadline>> = 
     newsRemoteDataSource.latestNews
    .map{news -> news.filter {userData.isFavouriteTopic(it) } }
    .onEach{ news -> saveInCache(news) }

}

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.
继续我们的例子，我们有这个NewsRepository。
在它的构造函数中，它接受我们以前见过的NewsRemoteDataSource类和UserData，UserData以了解关于登录用户的更多信息。
It exposes this favouriteLatestNews latest news preperty of type of Flow of a List of ArticleHeadlind.
它暴露了这个favoritelatestnews属性，属于Flow类型的ArticleHeadlind列表的最新新闻。
With this ,we want to expose just the news articles that are relevant to the user.
通过这种方式，我们希望只公开与用户相关的新闻文章。

This function accesses the latestNews property from the 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 filters list instead of the original one.
Also,wo use onEach as a side effect to save the favorite user news in the cache.
此latestNews函数从newsRemoteDataSource访问latestNews属性。
然后它应用map操作符来修改数据流。
我们在已发出的列表上使用过滤器来选择用户感兴趣的主题的文章。
转换发生在流上。
消费者将看到过滤器列表而不是原来的列表。
此外，我们使用onEach作为一个副作用来保存最喜欢的用户新闻在缓存中。

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.

中间操作符可以一个接一个地应用，形成操作链，在将项发送到Flow时惰性执行这些操作。
Note that simply applying an intermediate operator to a stream does not start the Flow colleciotn.
To trigger the Flow and start listening for values,use a terminal operator.
请注意，简单地对流应用中间操作符不会启动Flow colleciotn。
要触发流并开始监听值，请使用终端操作符。
With collect ,you get all values at the time they are emitted into the stream.
使用collect，您可以在所有值被释放到流时获得它们。

class LatestNewsViewModel(
   private val newsRepository: NewsRepository
)  :ViewModel(){
  init{
     viewModelScope.launch {
        newsRepository.favouriteLatestNews.collect{ favouriteNews ->
           //Update View with the latest favourite news 
       }
    }
  }
  
}

Now ,in our view model in this case ,the LatestNewsViewModel，we want to consume the 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.
现在，在我们的视图模型(在本例中是LatestNewsViewModel)中，我们希望消费Flow流以获得新闻通知并相应地更新UI。
在那里，我们可以调用collect来触发Flow并开始使用它来获取值。
lambda表达式将在接收到的每个新值上执行。
But, as we said,collect is a suspend function.
Therefore,it needs to be executed within a Coroutine that can create with the build-in ViewModelScope.
但是，就像我们说的，collect是一个挂起函数。
因此，它需要在一个可以用内置的ViewModelScope创建的协程中执行。
So what's really happening here ?
When the ViewModel is created ,we create a new Coroutine to collect the results from favouriteLatestNews.
那么到底发生了什么呢?
当ViewModel被创建时，我们创建一个新的Coroutine从favoritelatestnews收集结果。

This triggers the Flow in the data source layer,which will start fetching the latest news from the network.
这将触发数据源层中的Flow，它将开始从网络获取最新的新闻。
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.
存储库层中的map操作符修改所有排放，以获取用户最喜欢的主题。
之后，存储库将在缓存中保存该信息。
视图模型将得到最新的过滤信息。

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.
由于生成器在while循环中始终处于活动状态，当ViewModel被清除和ViewModelScope被取消时，数据流将被关闭。

Flow collection停止

There are two ways Flow collection can stop.
One way is when 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.

有两种方法可以停止流收集。
一种方法是当生产者完成输出项目时。数据流关闭。
调用collect的协程会继续执行。
或者，Coroutine的收集被取消，就像我们的例子。
这也将停止潜在的生产者。

通常Flow是冷、懒加载

Flows are cold and lazy unless specified with other intermediate operators.
This means that producer code will be executed each time a terminal operator is called on the Flow.
流是冷的和惰性的，除非用其他中间操作符指定。
这意味着每次在Flow上调用终端操作符时都将执行生成程序代码。
In the example,multiple collectors of the Flow makes the data source to fetch the latest news multiple times on different fixed intervals.

在本例中，Flow的多个收集器使数据源以不同的固定间隔多次获取最新新闻。
See the shareIn operator to optimize and share the Flow when multiple consumers collect at the same time.
当多个消费者同时收集时，请参阅shareIn操作符来优化和共享Flow。

第三方库实现生产者

The implementation of the producer can come from a third-party libary.
生产者的实现可以来自第三方库。
And as such ,it could throw unexpected exceptions.
To handle these exceptions ,use the catch intermediate operator.

因此，它可能会抛出意外异常。
要处理这些异常，请使用catch中间操作符。

class LatestNewsViewModel(
   private val newsRepository: NewsRepository
)  :ViewModel(){
  init{
     viewModelScope.launch {
        newsRepository.favouriteLatestNews
           .catch{ exception -> notifyError(exception) }
           .collect{ favouriteNews ->
           //Update View with the latest favourite news
       }
    }
  }
  
}

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.

同样，在ViewModel层中，为了捕获意外异常，我们可以使用catch操作符来处理它们,并向用户显示正确的消息。
由于catch是一个中间操作符，因此需要在collect之前调用它。

class NewsRepository(
 private val newsRemoteDataSource: NewsRemoteDataSource,
 private val userData ： UserData //比如用户登陆返回的用户信息
) {
  val favouriteLatestNews : Flow<List<ArticleHeadline>> = 
     newsRemoteDataSource.latestNews
    .map{news -> news.filter {userData.isFavouriteTopic(it) } }
    .onEach{ news -> saveInCache(news) }
    //If an error happens,emit the last cached values 如果网络范围错误，那么使用最新的cache数据
    .catch{ exception-> emit(lastCachedNews() }

}

But catch can also emits 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.
但catch也可以发出流。
如果我们想在存储库层处理那些意外的异常，我们可以使用catch操作符，并使用带有最新缓存新闻的emit函数向Flow发出消息。

If we talk about Android specifically Flow is integrated in many JetPack libraries.
And it's popular among Android third-party libraries as well.
如果我们专门讨论Android, Flow集成在许多JetPack库中。
它在Android第三方库中也很受欢迎。

Flow &Room配合使用

Flow is a great fit for data updates.
Flow非常适合数据更新。

您可以查看官方文章使用 Flow with Room 接收有关数据库更改的通知

@Daa
abstract class ExampleDao {
    @Query("SELECT * FROM Example ")
    abstract fun getExamples() :Flow<List<Example>>
}

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.
例如，您可以使用Flow和Room来通知数据库中的更改。
如代码片段所示，在Dao中，返回一个Flow类型以获得实时更新。
每当在Example表中发生更改时，就会触发一个新的List，其中包含数据库中的项目。

callbackFlow构造器

At this point, you pretty much know everything you need about Flows.
在这一点上，您几乎已经了解了关于flow所需的所有内容。
However,there is another Flow builder I want to talk about,as it is used quite ofen.
And that is callbackFlow that lets you convert callback-based APIs into flows.
然而，我想谈谈另一种流构建器，因为它被使用得非常频繁。
这就是callbackFlow，它允许你将基于回调的api转换为流。
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.
例如，Firebase Firestore Android api使用回调。
让我们看看如何将这些回调转换为流并侦听Firestore数据库更新。

class FirestoreUserEventsDataSource(
    private val firestore :FirebaseFirestore
){
    //Method to get user events from the Firestore database
    fun getUserEvents() :Flow<UserEvents> { ... }
}

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.

在这里，我们有一个FirestoreUserEventsDataSource，它的getUserEvents方法返回一个UserEvents流。
如您所见，我们将FirebaseFirestore的一个实例作为依赖项。

class FirestoreUserEventsDataSource(
    private val firestore :FirebaseFirestore
){
    //Method to get user events from the Firestore database
    fun getUserEvents() :Flow<UserEvents> = callbackFlow {
        ...
    }
}

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.

要创建流，我们使用callbackFlow API。
与Flow构建器API一样，在这里，我们处于协程的上下文中。
但是，与Flow构建器不同的是，channelFlow允许从不同的协程上下文中或通过offer方法在协程外部发出值，我们将看到这一点。

image.png

class FirestoreUserEventsDataSource(    ...){
    var eventsCollection ：CollectionReference ?=null
    try{
        eventsCollection = FirebaseFirestore.getInstance()
             .collection("collection")
             .document("app")
    }catch(e:Throwable){
    }
   
}

The first thing to do is initialize in Firebase and getting the eventsCollection from it.
Therefore,we write something like this code.
要做的第一件事是在Firebase中初始化并从中获取eventsCollection。
因此，我们编写了如下代码。

class FirestoreUserEventsDataSource(    ...){
    var eventsCollection ：CollectionReference ?=null
    try{
        eventsCollection = FirebaseFirestore.getInstance()
             .collection("collection")
             .document("app")
    }catch(e:Throwable){
        close(e) //关闭Flow
    }
   
}

However,this could potentially throw an an exception if it fails getting the eventsCollection.
If that happens,or Firebase cannot be initialized,we need to closed the Flow.
但是，如果获取eventsCollection失败，可能会抛出异常。
如果发生这种情况，或者Firebase无法初始化，则需要关闭Flow。

image.png

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.

image.png

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.
在那里，我们检查接收到的集合的快照是否为新的。
如果没有，则调用offer将一个项发送到流。
Offer 不是挂起函数。这就是为什么我们可以在回调中调用它。

image.png

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.
到目前为止，我们初始化了Firestore并添加了订阅。
接下来是什么?
现在，我们希望保持与流的使用者的通信打开，以便它能够接收Firestore发送的所有事件。

image.png

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.

将基于回调的 API 转换为数据流的完整代码

class FirestoreUserEventsDataSource(
    private val firestore: FirebaseFirestore
) {
    // Method to get user events from the Firestore database
    fun getUserEvents(): Flow<UserEvents> = callbackFlow {

        // Reference to use in Firestore
        var eventsCollection: CollectionReference? = null
        try {
            eventsCollection = FirebaseFirestore.getInstance()
                .collection("collection")
                .document("app")
        } catch (e: Throwable) {
            // If Firebase cannot be initialized, close the stream of data
            // flow consumers will stop collecting and the coroutine will resume
            close(e)
        }

        // Registers callback to firestore, which will be called on new events
        val subscription = eventsCollection?.addSnapshotListener { snapshot, _ ->
            if (snapshot == null) { return@addSnapshotListener }
            // Sends events to the flow! Consumers will get the new events
            try {
                offer(snapshot.getEvents())
            } catch (e: Throwable) {
                // Event couldn't be sent to the flow
            }
        }

        // The callback inside awaitClose will be executed when the flow is
        // either closed or cancelled.
        // In this case, remove the callback from Firestore
        awaitClose { subscription?.remove() }
    }
}

---

结束语

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.