Контроллеры Stream

Часть раздела Объектно-ориентированное программирование путешествия по Dart на Coddy — урок 78 из 110.

Хотя генераторы async* отлично подходят для создания потоков с предопределенными данными, иногда вам требуется больше контроля — например, добавление данных в поток из разных частей вашего кода. Именно здесь на помощь приходит StreamController.

StreamController выступает в роли менеджера для потока. Он предоставляет вам «sink» для добавления данных и «stream», на который могут подписываться слушатели:

import 'dart:async';

void main() {
  var controller = StreamController<String>();
  
  // Прослушивание потока
  controller.stream.listen((data) {
    print('Received: $data');
  });
  
  // Добавление данных через sink
  controller.add('Hello');
  controller.add('World');
  
  controller.close();  // Всегда закрывайте, когда закончите
}

Метод add() помещает новые значения в поток, и любые слушатели получают их немедленно. Всегда следует вызывать close() по завершении работы, чтобы освободить ресурсы.

По умолчанию поток может иметь только одного слушателя. Для нескольких слушателей используйте широковещательный контроллер (broadcast controller):

import 'dart:async';

void main() {
  var controller = StreamController<int>.broadcast();
  
  controller.stream.listen((n) => print('Listener 1: $n'));
  controller.stream.listen((n) => print('Listener 2: $n'));
  
  controller.add(42);
  controller.close();
}

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

Задание

Легко

Давайте создадим систему центра сообщений (message hub), которая позволяет нескольким подписчикам получать сообщения в режиме реального времени! Вы будете использовать StreamController для управления потоком сообщений и продемонстрируете, как широковещательные потоки (broadcast streams) позволяют работать нескольким слушателям.

Вы организуете свой код в двух файлах:

message_hub.dart: Создайте класс MessageHub, который управляет распределением сообщений с помощью StreamController. Ваш хаб должен:
- Использовать широковещательный StreamController<String> внутри, чтобы несколько слушателей могли подписаться
- Предоставить геттер с именем messages, который возвращает поток для подписки слушателей
- Иметь метод broadcast(String message), который добавляет сообщение в поток
- Иметь метод shutdown(), который закрывает контроллер
main.dart: Импортируйте ваш message hub и настройте сценарий с несколькими подписчиками, слушающими один и тот же поток:
- Создайте экземпляр MessageHub
- Добавьте первого слушателя, который выводит сообщения в формате: Subscriber A: [message]
- Добавьте второго слушателя, который выводит сообщения в формате: Subscriber B: [message]
- Отправьте (broadcast) три сообщения: Hello everyone, Welcome to the hub и Goodbye
- Завершите работу хаба (shutdown)

Не забудьте импортировать dart:async в файле вашего message hub, чтобы получить доступ к StreamController. Обратите внимание, как оба подписчика получают каждое транслируемое сообщение — в этом и заключается сила широковещательных потоков!

Ожидаемый вывод:

Subscriber A: Hello everyone
Subscriber B: Hello everyone
Subscriber A: Welcome to the hub
Subscriber B: Welcome to the hub
Subscriber A: Goodbye
Subscriber B: Goodbye

Шпаргалка

StreamController управляет потоком, обеспечивая контроль над тем, когда и как добавляются данные. Он предоставляет вам «sink» (приемник) для добавления данных и «stream» (поток), на который могут подписываться слушатели.

Базовое использование:

import 'dart:async';

var controller = StreamController<String>();

// Listen to the stream
controller.stream.listen((data) {
  print('Received: $data');
});

// Add data through the sink
controller.add('Hello');
controller.add('World');

controller.close();  // Always close when done

Метод add() отправляет новые значения в поток. Слушатели получают их немедленно. Всегда вызывайте close() по завершении работы, чтобы освободить ресурсы.

По умолчанию у потока может быть только один слушатель. Для нескольких слушателей используйте широковещательный контроллер (broadcast):

var controller = StreamController<int>.broadcast();

controller.stream.listen((n) => print('Listener 1: $n'));
controller.stream.listen((n) => print('Listener 2: $n'));

controller.add(42);
controller.close();

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

Попробуйте сами

import 'message_hub.dart';

void main() {
  // TODO: Создать экземпляр MessageHub
  
  // TODO: Добавить первого слушателя, который выводит: Subscriber A: [message]
  
  // TODO: Добавить второго слушателя, который выводит: Subscriber B: [message]
  
  // TODO: Транслировать три сообщения:
  // - "Hello everyone"
  // - "Welcome to the hub"
  // - "Goodbye"
  
  // TODO: Завершить работу хаба
}