Класс Stream<T>

Источник асинхронных событий данных.

Stream предоставляет способ получать последовательность событий. Каждое событие является либо событием данных, также называемым элементом потока, либо событием ошибки, которое уведомляет о сбое. Когда поток отправил все свои события, единственное событие "done" уведомляет слушателя о достижении конца.

Поток создаётся вызовом функции async*, которая возвращает поток. При использовании этого потока функция отправляет события до своего завершения, после чего поток закрывается. Поток можно использовать либо с помощью цикла await for, доступного внутри функции async или async*, либо пересылать его события напрямую с помощью yield* внутри функции async*. Пример:

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 со значением и отправляет результат в возвращаемый поток. Если поток source не отправляет события ошибок, цикл завершается при завершении потока source, затем завершается тело функции optionalMap, и возвращаемый поток закрывается. При получении события ошибки из потока source функция await for повторно выбрасывает эту ошибку, что прерывает цикл. Затем ошибка достигает конца тела функции optionalMap, поскольку её не перехватывают. В результате ошибка отправляется в возвращаемый поток, который затем закрывается.

Класс Stream также предоставляет возможности для ручного прослушивания событий потока, преобразования потока в другой поток или в future.

Функция forEach соответствует циклу await for, так же как Iterable.forEach соответствует обычному циклу for/in. Как и цикл, она вызывает функцию для каждого события данных и прерывается при ошибке.

Метод более низкого уровня listen лежит в основе всех остальных методов. Вызов listen для потока сообщает ему, что вы хотите получать события, и регистрирует обратные вызовы, которые будут получать эти события. При вызове listen возвращается объект StreamSubscription, который является активным объектом, предоставляющим события, и позволяет прекратить прослушивание или временно приостановить поступление событий от подписки.

Существует два вида потоков: потоки «с одной подпиской» и «широковещательные» потоки.

Поток с одной подпиской допускает только одного слушателя на протяжении всего времени существования потока. Он не начинает генерировать события, пока у него нет слушателя, и прекращает отправлять события после отмены подписки слушателя, даже если источник событий может предоставить новые. Поток, созданный функцией async*, является потоком с одной подпиской, однако каждый вызов этой функции создаёт новый такой поток.

Повторная подписка на поток с одной подпиской запрещена, даже после отмены первой подписки.

Потоки с одной подпиской обычно используются для передачи фрагментов больших непрерывных данных, например при файловом вводе-выводе.

Широковещательный поток допускает любое количество слушателей и отправляет события по мере их готовности независимо от наличия слушателей.

Широковещательные потоки используются для независимых событий и наблюдателей.

Если несколько слушателей хотят подписаться на поток с одной подпиской, используйте asBroadcastStream, чтобы создать широковещательный поток на основе обычного потока.

Для обоих видов потоков преобразования потока, такие как where и skip, возвращают поток того же типа, что и поток, к которому был применён метод, если не указано иное.

При отправке события его получают слушатели, подписанные в этот момент. Если слушатель добавлен к широковещательному потоку во время отправки события, он не получит отправляемое событие. После отмены подписки слушатель сразу перестаёт получать события. Прослушивание широковещательного потока можно рассматривать как прослушивание нового потока, содержащего только события, ещё не отправленные на момент вызова listen. Например, геттер first подписывается на поток, а затем возвращает первое событие, полученное этим слушателем. Это не обязательно первое событие, отправленное потоком, а первое из оставшихся событий широковещательного потока.

При отправке события "done" подписки отменяются до получения этого события. После его отправки у потока не остаётся подписчиков. После этого к широковещательному потоку можно добавлять новых подписчиков, но они просто получат новое событие "done" при первой возможности.

Подписки на потоки всегда выполняют запросы на приостановку. При необходимости им приходится буферизировать входящие данные, но часто и предпочтительно они могут просто запросить приостановку входящего потока.

Реализация isBroadcast по умолчанию возвращает false. Широковещательный поток, наследующий от Stream, должен переопределить isBroadcast, чтобы возвращать true, если он должен обозначить, что ведёт себя как широковещательный поток.

Реализуют

Конструкторы

Stream()
Stream.empty() const factory: Создаёт пустой широковещательный поток.
Stream.error(Object error, [StackTrace? stackTrace]) factory: Создаёт поток, который отправляет одно событие ошибки перед завершением.
Stream.eventTransformed(Stream source, EventSink mapSink(EventSink<T> sink)) factory: Создаёт поток, в котором все события существующего потока проходят через преобразование приёмника.
Stream.fromFuture(Future<T> future) factory: Создаёт новый поток с одной подпиской на основе future.
Stream.fromFutures(Iterable<Future<T>> futures) factory: Создаёт поток с одной подпиской на основе группы future.
Stream.fromIterable(Iterable<T> elements) factory: Создаёт поток, получающий данные из elements.
Stream.multi(void onListen(MultiStreamController<T>), {bool isBroadcast = false}) factory: Создаёт поток с несколькими подписками.
Stream.periodic(Duration period, [T computation(int computationCount)?]) factory: Создаёт поток, который многократно отправляет события с интервалами period.
Stream.value(T value) factory: Создаёт поток, который отправляет одно событие данных перед закрытием.

Свойства

first → Future<T> только для чтения: Первый элемент этого потока.
hashCode → int только для чтения, наследуется: Хеш-код этого объекта.
isBroadcast → bool только для чтения: Является ли этот поток широковещательным.
isEmpty → Future<bool> только для чтения: Содержит ли этот поток какие-либо элементы.
last → Future<T> только для чтения: Последний элемент этого потока.
length → Future<int> только для чтения: Количество элементов в этом потоке.
runtimeType → Type только для чтения, наследуется: Представление типа объекта во время выполнения.
single → Future<T> только для чтения: Единственный элемент этого потока.

Методы

any(bool test(T element)) → Future<bool>: Проверяет, принимает ли test хотя бы один элемент этого потока.
asBroadcastStream({void onListen(StreamSubscription<T> subscription)?, void onCancel(StreamSubscription<T> subscription)?}) → Stream<T>: Возвращает поток с несколькими подписками, который генерирует те же события, что и этот поток.
asyncExpand<E>(Stream<E>? convert(T event)) → Stream<E>: Преобразует каждый элемент в последовательность асинхронных событий.
asyncMap<E>(FutureOr<E> convert(T event)) → Stream<E>: Создаёт новый поток, асинхронно преобразуя каждое событие данных этого потока в новое событие.
cast<R>() → Stream<R>: Адаптирует этот поток, чтобы он был Stream<R>.
contains(Object? needle) → Future<bool>: Возвращает, встречается ли needle среди элементов этого потока.
distinct([bool equals(T previous, T next)?]) → Stream<T>: Пропускает события данных, если они равны предыдущему событию данных.
drain<E>([E? futureValue]) → Future<E>: Отбрасывает все данные этого потока, но сообщает о его завершении или возникновении ошибки.
elementAt(int index) → Future<T>: Возвращает значение события данных этого потока с номером index.
every(bool test(T element)) → Future<bool>: Проверяет, принимает ли test все элементы этого потока.
expand<S>(Iterable<S> convert(T element)) → Stream<S>: Преобразует каждый элемент этого потока в последовательность элементов.
firstWhere(bool test(T element), {T orElse()?}) → Future<T>: Находит первый элемент этого потока, соответствующий условию test.
fold<S>(S initialValue, S combine(S previous, T element)) → Future<S>: Объединяет последовательность значений, многократно применяя combine.
forEach(void action(T element)) → Future: Выполняет action для каждого элемента этого потока.
handleError(Function onError, {bool test(dynamic error)?}) → Stream<T>: Создаёт поток-обёртку, перехватывающий некоторые ошибки этого потока.
join([String separator = ""]) → Future<String>: Объединяет строковые представления элементов в одну строку.
lastWhere(bool test(T element), {T orElse()?}) → Future<T>: Находит последний элемент этого потока, соответствующий условию test.
listen(void onData(T event)?, {Function? onError, void onDone()?, bool? cancelOnError}) → StreamSubscription<T>: Добавляет подписку на этот поток.
map<S>(S convert(T event)) → Stream<S>: Преобразует каждый элемент этого потока в новое событие потока.
noSuchMethod(Invocation invocation) → dynamic наследуется: Вызывается при обращении к несуществующему методу или свойству.
pipe(StreamConsumer<T> streamConsumer) → Future: Передаёт события этого потока в streamConsumer.
reduce(T combine(T previous, T element)) → Future<T>: Объединяет последовательность значений, многократно применяя combine.
singleWhere(bool test(T element), {T orElse()?}) → Future<T>: Находит единственный элемент этого потока, соответствующий условию test.
skip(int count) → Stream<T>: Пропускает первые count события данных этого потока.
skipWhile(bool test(T element)) → Stream<T>: Пропускает события данных этого потока, пока они соответствуют условию test.
take(int count) → Stream<T>: Предоставляет не более первых count событий данных этого потока.
takeWhile(bool test(T element)) → Stream<T>: Пересылает события данных, пока test возвращает true.
timeout(Duration timeLimit, {void onTimeout(EventSink<T> sink)?}) → Stream<T>: Создаёт новый поток с теми же событиями, что и у этого потока.
toList() → Future<List<T>>: Собирает все элементы этого потока в List.
toSet() → Future<Set<T>>: Собирает данные этого потока в Set.
toString() → String наследуется: Строковое представление этого объекта.
transform<S>(StreamTransformer<T, S> streamTransformer) → Stream<S>: Применяет streamTransformer к этому потоку.
where(bool test(T event)) → Stream<T>: Создаёт из этого потока новый поток, отбрасывающий некоторые элементы.

Операторы

operator ==(Object other) → bool наследуется: Оператор равенства.

Статические методы

castFrom<S, T>(Stream<S> source) → Stream<T>: Адаптирует source, чтобы он стал Stream<T>.

© 2012 the Dart project authors
Licensed under the BSD 3-Clause "New" or "Revised" License.
https://api.dart.dev/stable/2.19.6/dart-async/Stream-class.html