Stream<T> 类 抽象 混合

异步数据事件的来源。

Stream提供了一个接收一系列事件的途径。每个事件要么是数据事件，也称为流的元素，要么是错误事件，它表示发生了一些错误。当一个流已发出所有事件时，一个单独的“完成”事件通知监听器已达到末尾。

您通过调用一个async*函数来产生一个流，然后该函数返回一个流。消耗该流将导致该函数发出事件，直到它结束，并且流关闭。您可以使用一个await for循环消耗一个流，该循环在async或async*函数内部可用，或者通过在async*函数中使用yield*直接将事件转发。示例

Stream<T> optionalMap<T>(
    Stream<T> source , [T Function(T)? convert]) async* {
  if (convert == null) {
    yield* source;
  } else {
    await for (var event in source) {
      yield convert(event);
    }
  }
}

当调用此函数时，它立即返回一个Stream<T>对象。然后不再发生任何事情，直到有人尝试消耗该流。那时，async*函数的主体开始运行。如果省略了convert函数，则yield*将监听source流并将所有事件（数据、错误等）都转发到返回的流中。当source流关闭时，yield*完成，并且optionalMap函数的主体也结束。这关闭了返回的流。如果提供了convert，则该函数将监听源流并进入一个await for循环，循环反复地等待下一个数据事件。在数据事件上，它调用convert并发出结果。如果源流没有发出错误事件，循环将在源流结束时结束，然后optionalMap函数主体完成，这关闭了返回的流。从源流的一个错误事件，await for重新抛出该错误，打破了循环。由于没有捕获，该错误随后到达optionalMap函数主体的末尾。这使得错误被发射到返回的流中，这导致流关闭。

('.')

forEach 函数对应于 await for 循环，正如 Iterable.forEach 对应于正常的 for/in 循环一样。像循环一样，它将为每个数据事件调用一个函数并在错误时中断。

更底层的 listen 方法是所有其他方法的基础。您在流上调用 listen 以告诉它您希望接收事件，并注册将接收这些事件的回调。当您调用 listen 时，您接收一个 StreamSubscription 对象，它是提供事件的活动对象，并且可以用来停止再次监听，或者临时暂停订阅的事件。

有两种类型的流：“单订阅”流和“广播”流。

单订阅流 在整个流的生命周期内只允许一个监听器。它在没有监听器的情况下不会开始生成事件，并且在监听器取消订阅时停止发送事件，即使事件源还能提供更多。由 async* 函数创建的流是一个单订阅流，但是函数的每个调用都将创建一个新的这样的流。

不允许在单订阅流上进行两次监听，即使在第一次订阅被取消之后也不允许。

单订阅流通常用于流式传输大块连续数据，例如文件I/O。

广播流 允许任何数量的听众，并且当它们准备好了，不论有没有听众，都会触发事件。

广播流用于独立事件/观察者。

如果多个听众想要监听一个单订阅流，可以使用 asBroadcastStream 在非广播流上创建一个广播流。

在任何一种流上，流转换，如 where 和 skip， 返回和被调用的方法相同的流类型，除非有其他说明。

当事件被触发时，当时的那个听众将接收到事件。如果在事件正在触发时向广播流添加了一个听众，那个听众将不会接收到当前正在触发的事件。如果取消了一个听众，它将立即停止接收事件。在广播流上监听可以视为在 listen 调用发生时还没有发射的事件的新流上监听。例如，first 属性监听流，然后返回听众接收到的第一个事件。这不一定是最先由流发射的事件，但是最先的 剩余 事件。

当 "done" 事件被触发时，在接收到事件之前，订阅者会被注销。事件发送后，流上没有订阅者。在此之后向广播流添加新的订阅者是被允许的，但它们将立即收到一个新的 "done" 事件。

流订阅总是尊重 "pause" 请求。如果必要，它们需要缓存它们的输入，但通常最好是它们简单地请求暂停输入。

isBroadcast 方法的默认实现返回 false。一个继承自 Stream 的广播流必须覆盖 isBroadcast 方法，如果它想要指示它像一个广播流那样运行。