Creando streams en Dart

Written by Lasse Nielsen
April 2013 (updated May 2021)

La librería dart:async contiene dos tipos que son importantes para muchas APIs de Dart: Stream y Future. Mientras que un Future representa el resultado de un solo cálculo, un stream es una secuencia de resultados. Escuchas un stream para recibir notificaciones de los resultados (tanto datos como errores) y del cierre del stream. También puedes pausar mientras escuchas o dejar de escuchar el stream antes de que se complete.

Pero este artículo no trata sobre usar streams. Trata sobre crear tus propios streams. Puedes crear streams de varias formas:

• Transformar streams existentes.
• Crear un stream desde cero usando una función async*.
• Crear un stream usando un StreamController.

Este artículo muestra el código para cada enfoque y ofrece consejos para ayudarte a implementar tu stream correctamente.

Para obtener ayuda sobre cómo usar streams, consulta la sección de Programación asíncrona: Streams.

El caso común para crear streams es que ya tengas un stream y quieras crear un nuevo stream basado en los eventos del stream original. Por ejemplo, podrías tener un stream de bytes que deseas convertir a un stream de cadenas mediante la decodificación UTF-8 de la entrada. El enfoque más general es crear un nuevo stream que espere eventos en el stream original y luego emita nuevos eventos. Ejemplo:

/// Splits a stream of consecutive strings into lines.
///
/// The input string is provided in smaller chunks through
/// the `source` stream.
Stream<String> lines(Stream<String> source) async* {
  // Stores any partial line from the previous chunk.
  var partial = '';
  // Wait until a new chunk is available, then process it.
  await for (final chunk in source) {
    var lines = chunk.split('\n');
    lines[0] = partial + lines[0]; // Prepend partial line.
    partial = lines.removeLast(); // Remove new partial line.
    for (final line in lines) {
      yield line; // Add lines to output stream.
    }
  }
  // Add final partial line to output stream, if any.
  if (partial.isNotEmpty) yield partial;
}

Para muchas transformaciones comunes, puedes usar métodos de transformación suministrados por la clase Stream como map(), where(), expand() y take().

Por ejemplo, supón que tienes un stream, counterStream, que emite un contador creciente cada segundo. Así es como se podría implementar:

var counterStream = Stream<int>.periodic(
  const Duration(seconds: 1),
  (x) => x,
).take(15);

Para ver los eventos rápidamente, puedes usar un código como este:

counterStream.forEach(print); // Print an integer every second, 15 times.

Para transformar los eventos del stream, puedes invocar un método de transformación como map() en el stream antes de escucharlo. El método devuelve un nuevo stream.

// Double the integer in each event.
var doubleCounterStream = counterStream.map((int x) => x * 2);
doubleCounterStream.forEach(print);

En lugar de map(), puedes usar cualquier otro método de transformación, como los siguientes:

.where((int x) => x.isEven) // Retain only even integer events.
.expand((x) => [x, x]) // Duplicate each event.
.take(5) // Stop after the first five events.

A menudo, un método de transformación es todo lo que necesitas. Sin embargo, si necesitas aún más control sobre la transformación, puedes especificar un StreamTransformer con el método transform() de Stream. Las librerías de la plataforma proporcionan transformadores de stream para muchas tareas comunes. Por ejemplo, el siguiente código utiliza los transformadores utf8.decoder y LineSplitter proporcionados por la librería dart:convert.

Stream<List<int>> content = File('someFile.txt').openRead();
List<String> lines = await content
    .transform(utf8.decoder)
    .transform(const LineSplitter())
    .toList();

Una forma de crear un nuevo stream es con una función generadora asíncrona (async*). El stream se crea cuando se llama a la función, y el cuerpo de la función comienza a ejecutarse cuando se escucha el stream. Cuando la función retorna, el stream se cierra. Hasta que la función retorne, puede emitir eventos en el stream utilizando sentencias yield o yield*.

Aquí tienes un ejemplo primitivo que emite números a intervalos regulares:

Stream<int> timedCounter(Duration interval, [int? maxCount]) async* {
  int i = 0;
  while (true) {
    await Future.delayed(interval);
    yield i++;
    if (i == maxCount) break;
  }
}

Esta función devuelve un Stream. Cuando se escucha ese stream, el cuerpo comienza a ejecutarse. Se retrasa repetidamente durante el intervalo solicitado y luego produce (yields) el siguiente número. Si se omite el parámetro maxCount, no hay condición de parada en el bucle, por lo que el stream emite números cada vez mayores para siempre, o hasta que el oyente cancele su suscripción.

Cuando el oyente cancela (al invocar cancel() en el objeto StreamSubscription devuelto por el método listen()), la próxima vez que el cuerpo alcance una sentencia yield, el yield en su lugar actúa como una sentencia return. Se ejecuta cualquier bloque finally contenedor y la función finaliza. Si la función intenta producir un valor antes de salir, eso falla y actúa como un retorno.

Cuando la función finalmente sale, el future devuelto por el método cancel() se completa. Si la función sale con un error, el future se completa con ese error; de lo contrario, se completa con null.

Otro ejemplo más útil es una función que convierte una secuencia de futures en un stream:

Stream<T> streamFromFutures<T>(Iterable<Future<T>> futures) async* {
  for (final future in futures) {
    var result = await future;
    yield result;
  }
}

Esta función le pide al iterable futures un nuevo future, espera ese future, emite el valor resultante y luego hace un bucle. Si un future se completa con un error, el stream se completa con ese error.

Es raro tener una función async* construyendo un stream de la nada. Necesita obtener sus datos de alguna parte, y la mayoría de las veces ese lugar es otro stream. En algunos casos, como la secuencia de futures anterior, los datos provienen de otras fuentes de eventos asíncronos. En muchos casos, sin embargo, una función async* es demasiado simplista para manejar fácilmente múltiples fuentes de datos. Ahí es donde entra la clase StreamController.

Si los eventos de tu stream provienen de diferentes partes de tu programa, y no solo de un stream o futures que pueden ser recorridos por una función async, entonces usa un StreamController para crear y poblar el stream.

Un StreamController te da un nuevo stream y una forma de añadir eventos al stream en cualquier momento y desde cualquier lugar. El stream tiene toda la lógica necesaria para manejar oyentes y pausas. Devuelves el stream y te quedas con el controlador.

El siguiente ejemplo (de stream_controller_bad.dart) muestra un uso básico, aunque defectuoso, de StreamController para implementar la función timedCounter() de los ejemplos anteriores. Este código crea un stream para devolver, y luego le suministra datos basados en eventos de temporizador, que no son ni futures ni eventos de stream.

// NOTE: This implementation is FLAWED!
// It starts before it has subscribers, and it doesn't implement pause.
Stream<int> timedCounter(Duration interval, [int? maxCount]) {
  var controller = StreamController<int>();
  int counter = 0;
  void tick(Timer timer) {
    counter++;
    controller.add(counter); // Ask stream to send counter values as event.
    if (maxCount != null && counter >= maxCount) {
      timer.cancel();
      controller.close(); // Ask stream to shut down and tell listeners.
    }
  }
​
  Timer.periodic(interval, tick); // BAD: Starts before it has subscribers.
  return controller.stream;
}

Como antes, puedes usar el stream devuelto por timedCounter() así:

var counterStream = timedCounter(const Duration(seconds: 1), 15);
counterStream.listen(print); // Print an integer every second, 15 times.

Esta implementación de timedCounter() tiene un par de problemas:

• Comienza a producir eventos antes de tener suscriptores.
• Sigue produciendo eventos incluso si el suscriptor solicita una pausa.

Como muestran las siguientes secciones, puedes solucionar ambos problemas especificando callbacks como onListen y onPause al crear el StreamController.

Como regla general, los streams deben esperar a los suscriptores antes de comenzar su trabajo. Una función async* hace esto automáticamente, pero al usar un StreamController, tienes el control total y puedes agregar eventos incluso cuando no deberías. Cuando un stream no tiene suscriptor, su StreamController almacena en búfer (buffers) los eventos, lo que puede provocar una fuga de memoria si el stream nunca obtiene un suscriptor.

Intenta cambiar el código que usa el stream por el siguiente:

void listenAfterDelay() async {
  var counterStream = timedCounter(const Duration(seconds: 1), 15);
  await Future.delayed(const Duration(seconds: 5));
​
  // After 5 seconds, add a listener.
  await for (final n in counterStream) {
    print(n); // Print an integer every second, 15 times.
  }
}

Cuando este código se ejecuta, no se imprime nada durante los primeros 5 segundos, aunque el stream está trabajando. Luego se agrega el oyente, y los primeros 5 eventos más o menos se imprimen todos a la vez, ya que fueron almacenados en el búfer por el StreamController.

Para recibir notificaciones de suscripciones, especifica un argumento onListen al crear el StreamController. El callback onListen se llama cuando el stream obtiene su primer suscriptor. Si especificas un callback onCancel, se llama cuando el controlador pierde su último suscriptor. En el ejemplo anterior, Timer.periodic() debería moverse a un manejador onListen, como se muestra en la siguiente sección.

Evita producir eventos cuando el oyente ha solicitado una pausa. Una función async* se pausa automáticamente en una sentencia yield mientras la suscripción del stream está pausada. Un StreamController, por otro lado, almacena en búfer los eventos durante la pausa. Si el código que proporciona los eventos no respeta la pausa, el tamaño del búfer puede crecer indefinidamente. Además, si el oyente deja de escuchar poco después de pausar, entonces el trabajo invertido en crear el búfer se desperdicia.

Para ver qué sucede sin soporte de pausa, intenta cambiar el código que usa el stream por el siguiente:

void listenWithPause() {
  var counterStream = timedCounter(const Duration(seconds: 1), 15);
  late StreamSubscription<int> subscription;
​
  subscription = counterStream.listen((int counter) {
    print(counter); // Print an integer every second.
    if (counter == 5) {
      // After 5 ticks, pause for five seconds, then resume.
      subscription.pause(Future.delayed(const Duration(seconds: 5)));
    }
  });
}

Cuando expiran los cinco segundos de pausa, los eventos generados durante ese tiempo se reciben todos a la vez. Eso sucede porque la fuente del stream no respeta las pausas y continúa agregando eventos al stream. Por lo tanto, el stream almacena los eventos en búfer y luego vacía su búfer cuando el stream se reanuda.

La siguiente versión de timedCounter() (de stream_controller.dart) implementa la pausa usando los callbacks onListen, onPause, onResume y onCancel en el StreamController.

Stream<int> timedCounter(Duration interval, [int? maxCount]) {
  late StreamController<int> controller;
  Timer? timer;
  int counter = 0;
​
  void tick(_) {
    counter++;
    controller.add(counter); // Ask stream to send counter values as event.
    if (counter == maxCount) {
      timer?.cancel();
      controller.close(); // Ask stream to shut down and tell listeners.
    }
  }
​
  void startTimer() {
    timer = Timer.periodic(interval, tick);
  }
​
  void stopTimer() {
    timer?.cancel();
    timer = null;
  }
​
  controller = StreamController<int>(
    onListen: startTimer,
    onPause: stopTimer,
    onResume: startTimer,
    onCancel: stopTimer,
  );
​
  return controller.stream;
}

Ejecuta este código con la función listenWithPause() anterior. Verás que deja de contar mientras está pausado y se reanuda correctamente después.

Debes usar todos los listeners (onListen, onCancel, onPause y onResume) para recibir notificaciones sobre cambios en el estado de pausa. La razón es que si tanto el estado de suscripción como el de pausa cambian al mismo tiempo, solo se llama al callback onListen o onCancel.

Al crear un stream sin usar una función async*, ten en cuenta estos consejos:

• Ten cuidado al usar un controlador síncrono; por ejemplo, uno creado con StreamController(sync: true). Cuando envías un evento a un controlador síncrono no pausado (por ejemplo, utilizando los métodos add(), addError() o close() definidos por EventSink), el evento se envía inmediatamente a todos los oyentes del stream. Los oyentes del Stream nunca deben ser llamados hasta que el código que agregó el oyente haya retornado por completo, y usar un controlador síncrono en el momento equivocado puede romper esta promesa y hacer que un código correcto falle. Evita el uso de controladores síncronos.

• Si utilizas StreamController, el callback onListen se llama antes de que la llamada a listen devuelva el StreamSubscription. No permitas que el callback onListen dependa de que la suscripción ya exista. Por ejemplo, en el siguiente código, se dispara un evento onListen (y se llama a handler) antes de que la variable subscription tenga un valor válido.

  dart

  subscription = stream.listen(handler);
  

  content_copy

• Los callbacks onListen, onPause, onResume y onCancel definidos por StreamController son llamados por el stream cuando cambia el estado del oyente del stream, pero nunca durante el lanzamiento de un evento o la llamada de otro manejador de cambio de estado. En esos casos, el callback de cambio de estado se retrasa hasta que se completa el callback anterior.

• No intentes implementar la interfaz Stream por ti mismo. Es fácil equivocarse sutilmente en la interacción entre eventos, callbacks y la adición y eliminación de oyentes. Utiliza siempre un stream existente, posiblemente de un StreamController, para implementar la llamada listen de un nuevo stream.

• Aunque es posible crear clases que extiendan de Stream con más funcionalidad heredando de la clase Stream e implementando el método listen y la funcionalidad adicional sobre ella, generalmente no se recomienda porque introduce un nuevo tipo que los usuarios tienen que considerar. En lugar de una clase que sea un Stream (y más), a menudo puedes crear una clase que tenga un Stream (y más).