Programación asíncrona: Streams

La programación asíncrona en Dart se caracteriza por las clases Future y Stream.

Un Future representa un cálculo que no se completa de inmediato. Donde una función normal devuelve el resultado, una función asíncrona devuelve un Future, que eventualmente contendrá el resultado. El future te dirá cuándo está listo el resultado.

Un stream es una secuencia de eventos asíncronos. Es como un Iterable asíncrono, donde, en lugar de obtener el siguiente evento cuando lo solicitas, el stream te dice que hay un evento cuando está listo.

Los streams se pueden crear de muchas formas, lo cual es tema de otro artículo, pero todos se pueden usar de la misma manera: el bucle for asíncrono (comúnmente llamado simplemente await for) itera sobre los eventos de un stream tal como el bucle for itera sobre un Iterable. Por ejemplo:

Future<int> sumStream(Stream<int> stream) async {
  var sum = 0;
  await for (final value in stream) {
    sum += value;
  }
  return sum;
}

Este código simplemente recibe cada evento de un stream de eventos de enteros, los suma y devuelve (un future de) la suma. Cuando el cuerpo del bucle termina, la función se pausa hasta que llega el siguiente evento o el stream termina.

La función está marcada con la palabra clave async, la cual es requerida cuando se usa el bucle await for.

El siguiente ejemplo prueba el código anterior generando un stream simple de enteros usando una función async*:

Los streams terminan cuando no hay más eventos en ellos, y el código que recibe los eventos es notificado de esto del mismo modo que se le notifica la llegada de un nuevo evento. Al leer eventos usando un bucle await for, el bucle se detiene cuando el stream termina.

En algunos casos, ocurre un error antes de que el stream termine; tal vez la red falló al intentar obtener un archivo de un servidor remoto, o tal vez el código que crea los eventos tiene un error, pero alguien necesita saberlo.

Los streams también pueden entregar eventos de error del mismo modo que entregan eventos de datos. La mayoría de los streams se detendrán después del primer error, pero es posible tener streams que entreguen más de un error, y streams que entreguen más datos después de un evento de error. En este documento solo discutimos streams que entregan como máximo un error.

Al leer un stream usando await for, la declaración del bucle lanza el error. Esto también finaliza el bucle. Puedes capturar el error usando try-catch. El siguiente ejemplo lanza un error cuando el iterador del bucle es igual a 4:

La clase Stream contiene una serie de métodos auxiliares que pueden realizar operaciones comunes en un stream por ti, similares a los métodos de un Iterable. Por ejemplo, puedes encontrar el último entero positivo en un stream usando lastWhere() de la API de Stream.

Future<int> lastPositive(Stream<int> stream) =>
    stream.lastWhere((x) => x >= 0);

Hay dos tipos de streams.

El tipo más común de stream contiene una secuencia de eventos que forman parte de un todo más grande. Los eventos deben entregarse en el orden correcto y sin omitir ninguno de ellos. Este es el tipo de stream que obtienes cuando lees un archivo o recibes una solicitud web.

Tal stream solo se puede escuchar una vez. Escucharlo de nuevo más tarde podría significar perderse los eventos iniciales, y entonces el resto del stream no tendría sentido. Cuando comienzas a escuchar, los datos se obtendrán y se proporcionarán en fragmentos.

El otro tipo de stream está destinado a mensajes individuales que se pueden manejar uno a la vez. Este tipo de stream se puede utilizar para eventos del mouse en un navegador, por ejemplo.

Puedes comenzar a escuchar dicho stream en cualquier momento, y obtendrás los eventos que se activen mientras escuchas. Más de un oyente puede escuchar al mismo tiempo, y puedes volver a escuchar más tarde después de cancelar una suscripción anterior.

Los siguientes métodos en Stream<T> procesan el stream y devuelven un resultado:

Future<T> get first;
Future<bool> get isEmpty;
Future<T> get last;
Future<int> get length;
Future<T> get single;
Future<bool> any(bool Function(T element) test);
Future<bool> contains(Object? needle);
Future<E> drain<E>([E? futureValue]);
Future<T> elementAt(int index);
Future<bool> every(bool Function(T element) test);
Future<T> firstWhere(bool Function(T element) test, {T Function()? orElse});
Future<S> fold<S>(S initialValue, S Function(S previous, T element) combine);
Future forEach(void Function(T element) action);
Future<String> join([String separator = '']);
Future<T> lastWhere(bool Function(T element) test, {T Function()? orElse});
Future pipe(StreamConsumer<T> streamConsumer);
Future<T> reduce(T Function(T previous, T element) combine);
Future<T> singleWhere(bool Function(T element) test, {T Function()? orElse});
Future<List<T>> toList();
Future<Set<T>> toSet();

Todas estas funciones, excepto drain() y pipe(), corresponden a una función similar en Iterable. Cada una se puede escribir fácilmente usando una función async con un bucle await for (o simplemente usando uno de los otros métodos). Por ejemplo, algunas implementaciones podrían ser:

Future<bool> contains(Object? needle) async {
  await for (final event in this) {
    if (event == needle) return true;
  }
  return false;
}
​
Future forEach(void Function(T element) action) async {
  await for (final event in this) {
    action(event);
  }
}
​
Future<List<T>> toList() async {
  final result = <T>[];
  await forEach(result.add);
  return result;
}
​
Future<String> join([String separator = '']) async =>
    (await toList()).join(separator);

(Las implementaciones reales son ligeramente más complejas, pero principalmente por razones históricas).

Los siguientes métodos en Stream devuelven un nuevo stream basado en el stream original. Cada uno espera hasta que algo escuche en el nuevo stream antes de escuchar en el original.

Stream<R> cast<R>();
Stream<S> expand<S>(Iterable<S> Function(T element) convert);
Stream<S> map<S>(S Function(T event) convert);
Stream<T> skip(int count);
Stream<T> skipWhile(bool Function(T element) test);
Stream<T> take(int count);
Stream<T> takeWhile(bool Function(T element) test);
Stream<T> where(bool Function(T event) test);

Los métodos anteriores corresponden a métodos similares en Iterable, los cuales transforman un iterable en otro iterable. Todos ellos se pueden escribir fácilmente usando una función async con un bucle await for.

Stream<E> asyncExpand<E>(Stream<E>? Function(T event) convert);
Stream<E> asyncMap<E>(FutureOr<E> Function(T event) convert);
Stream<T> distinct([bool Function(T previous, T next)? equals]);

Las funciones asyncExpand() y asyncMap() son similares a expand() y map(), pero permiten que su argumento de función sea una función asíncrona. La función distinct() no existe en Iterable, pero podría haber existido.

Stream<T> handleError(Function onError, {bool Function(dynamic error)? test});
Stream<T> timeout(
  Duration timeLimit, {
  void Function(EventSink<T> sink)? onTimeout,
});
Stream<S> transform<S>(StreamTransformer<T, S> streamTransformer);

Las tres últimas funciones son más especializadas. Implican un manejo de errores que un bucle await for no puede gestionar directamente; el primer error encontrado terminará el bucle y su suscripción al stream, sin un mecanismo incorporado de recuperación.

El siguiente código demuestra cómo usar handleError() para filtrar errores de un stream antes de que sea consumido por un bucle await for.

Stream<S> mapLogErrors<S, T>(
  Stream<T> stream,
  S Function(T event) convert,
) async* {
  var streamWithoutErrors = stream.handleError((e) => log(e));
​
  await for (final event in streamWithoutErrors) {
    yield convert(event);
  }
}

En el ejemplo anterior, nunca se regresa a un bucle await for si el stream no emite eventos. Para evitar esto, usa la función timeout() para crear un nuevo stream. timeout() te permite establecer un límite de tiempo y continuar emitiendo eventos en el stream devuelto.

El siguiente código modifica el ejemplo anterior. Agrega un tiempo de espera (timeout) de dos segundos y genera un error correspondiente si no ocurren eventos durante dos o más segundos.

Stream<S> mapLogErrors<S, T>(
  Stream<T> stream,
  S Function(T event) convert,
) async* {
  var streamWithoutErrors = stream.handleError((e) => log(e));
  var streamWithTimeout = streamWithoutErrors.timeout(
    const Duration(seconds: 2),
    onTimeout: (eventSink) {
      eventSink.addError('Timed out after 2 seconds');
      eventSink.close();
    },
  );
​
  await for (final event in streamWithTimeout) {
    yield convert(event);
  }
}

La función transform() no es solo para el manejo de errores; es un "mapa" más generalizado para streams. Un mapa normal requiere un valor por cada evento entrante. Sin embargo, especialmente para streams de E/S (I/O), puede tomar varios eventos entrantes para producir un evento de salida. Un StreamTransformer puede funcionar con eso. Por ejemplo, los decodificadores como Utf8Decoder son transformadores. Un transformador requiere solo una función, bind(), la cual se puede implementar fácilmente mediante una función async.

El siguiente código lee un archivo y ejecuta dos transformaciones sobre el stream. Primero convierte los datos de UTF8 y luego los pasa por un LineSplitter. Se imprimen todas las líneas, excepto las que comiencen con un signo de número, #.

import 'dart:convert';
import 'dart:io';
​
void main(List<String> args) async {
  var file = File(args[0]);
  var lines = utf8.decoder
      .bind(file.openRead())
      .transform(const LineSplitter());
  await for (final line in lines) {
    if (!line.startsWith('#')) print(line);
  }
}

El último método de Stream es listen(). Este es un método de "bajo nivel"; todas las demás funciones de stream se definen en términos de listen().

StreamSubscription<T> listen(
  void Function(T event)? onData, {
  Function? onError,
  void Function()? onDone,
  bool? cancelOnError,
});

Para crear un nuevo tipo de Stream, puedes simplemente extender la clase Stream e implementar el método listen(); todos los demás métodos en Stream llaman a listen() para funcionar.

El método listen() te permite comenzar a escuchar en un stream. Hasta que lo hagas, el stream es un objeto inerte que describe qué eventos quieres ver. Cuando escuchas, se devuelve un objeto StreamSubscription que representa el stream activo produciendo eventos. Esto es similar a cómo un Iterable es solo una colección de objetos, pero el iterador es el que realiza la iteración real.

La suscripción al stream (stream subscription) te permite pausar la suscripción, reanudarla después de una pausa y cancelarla por completo. Puedes establecer callbacks para que se llamen para cada evento de datos o evento de error, y cuando el stream se cierre.

Lee la siguiente documentación para obtener más detalles sobre el uso de streams y la programación asíncrona en Dart.

• Creación de streams en Dart, un artículo sobre cómo crear tus propios streams
• Futures y manejo de errores, un artículo que explica cómo manejar errores usando la API de Future
• Programación asíncrona, que profundiza en el soporte del lenguaje Dart para la asincronía
• Referencia de la API de Stream