Programación asincrónica en Dart: Streams

Nota: Puede que quieras consultar una versión de este artículo donde el código está actualizado para ser null safe. (El video no ha cambiado.)

Este artículo cubre uno de los fundamentos de la programación reactiva: los streams, que son objetos de tipo Stream.

Si has leído el artículo anterior sobre futures, puede que recuerdes que cada future representa un único valor (ya sea un error o datos) que entrega de forma asíncrona. Los streams funcionan de manera similar, pero en lugar de una única cosa, un stream puede entregar cero o más valores y errores a lo largo del tiempo.

Este artículo es el tercero basado en la serie de videos Flutter in Focus Asynchronous Programming in Dart. El primer artículo, Isolates and event loops, cubrió los fundamentos del soporte de Dart para el trabajo en segundo plano. El segundo, Futures, discutió la clase Future. Si prefieres aprender viendo o escuchando, todo lo que se trata en este artículo está cubierto en el siguiente video.

Si piensas en cómo se relaciona un único valor con un iterador del mismo tipo, así es como se relaciona un future con un stream.

Al igual que con los futures, la clave es decidir de antemano: “Aquí está qué hacer cuando un dato esté listo, y cuando haya un error, y cuando el stream se complete.”

También, al igual que con los futures, el event loop de Dart sigue llevando el control.

Si estás usando el método openRead() de la clase File para leer datos de un archivo, por ejemplo, ese método devuelve un stream.

Fragmentos de datos se leen del disco y llegan al event loop. Una librería de Dart los examina y dice: “Ah, tengo a alguien esperando esto,” añade los datos al stream, y aparecen en el código de tu app.

Cuando llega otro fragmento de datos, entra y sale. Los streams basados en timers, el streaming de datos desde un socket de red — también funcionan con el event loop, usando eventos de reloj y de red.

Hablemos sobre cómo trabajar con los datos proporcionados por un stream. Supongamos que tienes una clase que te da un stream que emite un nuevo entero una vez por segundo: 1, 2, 3, 4, 5…

Puedes usar el método listen() para suscribirte al stream. El único parámetro obligatorio es una función.

final myStream = NumberCreator().stream;
​
final subscription = myStream.listen(
  (data) => print('Data: $data'),
);

Cada vez que el stream emite un nuevo valor, la función es llamada e imprime el valor:

Data: 1
Data: 2
Data: 3
Data: 4
...

Así funciona listen().

Importante: Por defecto, los streams están configurados para suscripción única. Mantienen sus valores hasta que alguien se suscribe, y solo permiten un único listener durante toda su vida útil. Si intentas escuchar un stream dos veces, obtendrás una excepción. Afortunadamente Dart también ofrece broadcast streams. Puedes usar el método asBroadcastStream() para convertir un stream de suscripción única en un broadcast stream. Los broadcast streams funcionan igual que los streams de suscripción única, pero pueden tener múltiples listeners, y si nadie está escuchando cuando un dato está listo, ese dato se descarta.

final myStream = NumberCreator().stream.asBroadcastStream();
​
final subscription = myStream.listen(
  (data) => print('Data: $data'),
);
​
final subscription2 = myStream.listen(
  (data) => print('Data again: $data'),
);

Volvamos a esa primera llamada a listen(), porque hay un par de cosas más de las que hablar.

Como mencionamos antes, los streams pueden producir errores igual que los futures. Al añadir una función onError a la llamada de listen(), puedes capturar y procesar cualquier error.

También hay una propiedad cancelOnError que es true por defecto, pero puede ponerse a false para mantener la suscripción activa incluso después de un error.

Y puedes añadir una función onDone para ejecutar algo de código cuando el stream ha terminado de enviar datos, como por ejemplo cuando un archivo se ha leído completamente.

Con los cuatro parámetros combinados — onError, onDone, cancelOnError, y el parámetro obligatorio (onData) — puedes estar preparado de antemano para lo que sea que ocurra.

final subscription = myStream.listen(
  (data) {
    print('Data: $data');
  },
  onError: (err) {
    print('Error!');
  },
  cancelOnError: false,
  onDone: () {
    print('Done!');
  },
);

Consejo: El pequeño objeto de suscripción que listen() devuelve tiene algunos métodos útiles propios. Es un StreamSubscription, y puedes usarlo para pausar, reanudar, e incluso cancelar el flujo de datos.

final subscription = myStream.listen(...);
​
subscription.pause();
subscription.resume();
subscription.cancel();

Ahora que sabes cómo usar listen() para suscribirte a un stream y recibir eventos de datos, podemos hablar de lo que hace que los streams sean realmente geniales: manipularlos. Una vez que tienes datos en un stream, muchas operaciones se vuelven fluidas y elegantes.

Volviendo a ese stream de números de antes, podemos usar un método llamado map() para tomar cada valor del stream y convertirlo al vuelo en otra cosa. Dale a map() una función para hacer la conversión, y devuelve un nuevo stream, tipado para coincidir con el valor de retorno de la función. En lugar de un stream de ints, ahora tienes un stream de strings. Puedes añadir una llamada a listen() al final, darle la función print(), y ahora estás imprimiendo strings directamente desde el stream, de forma asíncrona, a medida que llegan.

NumberCreator().stream
    .map((i) => 'String $i')
    .listen(print);
/*
OUTPUT:
String 1
String 2
String 3
String 4
*/

Hay un montón de métodos que puedes encadenar de esta forma. Si solo quieres imprimir los números pares, por ejemplo, puedes usar where() para filtrar el stream. Dale una función de test que devuelva un booleano para cada elemento, y devuelve un nuevo stream que solo incluye los valores que pasan el test.

NumberCreator().stream
    .where((i) => i % 2 == 0)
    .map((i) => 'String $i')
    .listen(print);
/*
OUTPUT:
String 2
String 4
String 6
String 8
*/

El método distinct() es otro buen ejemplo. Si tienes una app que usa un Redux store, ese store emite nuevos objetos de estado de la app en un stream onChange. Puedes usar map() para convertir ese stream de objetos de estado en un stream de view models para una parte de la app. Luego puedes usar el método distinct() para obtener un stream que filtra valores consecutivos idénticos (en caso de que el store emita un cambio que no afecta el subconjunto de datos del view model). Luego puedes escuchar y actualizar la UI siempre que obtengas un nuevo view model.

myReduxStore.onChange
    .map((s) => MyViewModel(s))
    .distinct()
    .listen( /* update UI */ );

Hay un montón de métodos adicionales integrados en Dart que puedes usar para dar forma y modificar tus streams. Además, cuando estés listo para cosas aún más avanzadas, está el paquete async mantenido por el equipo de Dart y disponible en pub.dev. Tiene clases que pueden fusionar dos streams, cachear resultados, y realizar otros tipos de magia basada en streams.

Para aún más magia de streams, échale un vistazo al paquete stream_transform.

Un tema avanzado merece una mención aquí, y es cómo crear tus propios streams. Al igual que con los futures, la mayor parte del tiempo vas a estar trabajando con streams creados para ti por librerías de red, librerías de archivos, gestión de estado, y demás. Pero también puedes crear los tuyos propios, usando un StreamController.

Volvamos a ese NumberCreator que hemos estado usando hasta ahora. Aquí está el código real:

class NumberCreator {
  NumberCreator() {
    Timer.periodic(Duration(seconds: 1), (t) {
      _controller.sink.add(_count);
      _count++;
    });
  }
​
  var _count = 1;
  final _controller = StreamController<int>();
  Stream<int> get stream => _controller.stream;
}

Como puedes ver, mantiene un conteo continuo, y usa un timer para incrementar ese conteo cada segundo. Lo interesante, sin embargo, es el stream controller.

Un StreamController crea un stream completamente nuevo desde cero, y te da acceso a ambos extremos. Está el extremo del stream en sí, donde llegan los datos. (Hemos estado usando ese a lo largo de este artículo.)

Stream<int> get stream => _controller.stream;

Luego está el extremo sink, que es donde se añaden nuevos datos al stream:

_controller.sink.add(_count);

NumberCreator aquí usa ambos. Cuando el timer se dispara, añade el último conteo al sink del controller, y luego expone el stream del controller con una propiedad pública para que otros objetos puedan suscribirse a él.

Ahora que hemos cubierto la creación, manipulación y escucha de streams, hablemos sobre cómo ponerlos a trabajar construyendo widgets en Flutter.

Si viste el video anterior sobre futures, puede que recuerdes FutureBuilder. Le das un future y un método builder, y construye widgets basándose en el estado del future.

Para streams, hay un widget similar llamado StreamBuilder. Dale un stream y un método builder, y reconstruirá sus hijos siempre que el stream emita un nuevo valor.

StreamBuilder<String>(
  stream: NumberCreator().stream.map((i) => 'String $i'),
  builder: (context, snapshot) {
    /* Build widgets! */
  }
)

El parámetro snapshot es un AsyncSnapshot, al igual que con FutureBuilder. Puedes verificar su propiedad connectionState para ver si el stream aún no ha enviado ningún dato o si ha terminado completamente. Puedes usar la propiedad hasError para ver si el último valor es un error. Y, por supuesto, puedes manejar los valores de datos.

StreamBuilder<String>(
  stream: NumberCreator().stream.map((i) => 'String $i'),
  builder: (context, snapshot) {
    if (snapshot.connectionState == ConnectionState.waiting) {
      return Text('No data yet.');
    } else if (snapshot.connectionState == ConnectionState.done) {
      return Text('Done!');
    }
  }
)

Lo principal es simplemente asegurarte de que tu builder sepa cómo manejar todos los posibles estados del stream. Una vez que tengas eso, puede reaccionar a lo que sea que haga el stream.

Este artículo habló sobre qué representan los streams, cómo obtienes valores de un stream, formas de manipular esos valores, y cómo StreamBuilder te ayuda a usar valores de stream en una app de Flutter.

Puedes aprender más sobre los streams en la documentación de Dart y Flutter:

• En dart.dev, consulta el tutorial de streams, el tutorial de app de línea de comandos, y la sección de streams del tour de librerías.

• En flutter.dev, consulta Work with WebSockets, que incluye un ejemplo que usa StreamBuilder.

O continúa con el siguiente video de la serie Asynchronous Programming in Dart. Habla sobre async y await, que son dos keywords que Dart ofrece para ayudarte a mantener tu código asíncrono compacto y fácil de leer.

Muchas gracias a Andrew Brogdon, quien creó el video en el que se basa este artículo.