Coroutines con Kotlin — Flow: I — Estructura, dinámica, restricciones y operadores

Puedes ir a cualquier artículo de esta serie haciendo clic en alguno de los siguientes enlaces:

Coroutines con Kotlin

1. Introducción
2. Constructores de Coroutines
3. CoroutineScope y CoroutineContext
4. CoroutineContext: I — El elemento ‘Job’
5. CoroutineContext: II — El elemento ‘ContinuationInterceptor’
6. CoroutineContext: III — El elemento ‘CoroutineExceptionHandler’
7. CoroutineContext: IV — El elemento ‘CoroutineName’
8. SupervisorJob y supervisorScope
9. Funciones suspendibles
10. Construcción de Coroutines con GlobalScope
11. Channels: I — Comunicación entre coroutines
12. Channels: II — El constructor ‘produce’
13. Estados mutables compartidos: I — Técnicas
14. Estados mutables compartidos: II — El constructor ‘actor’
15. Flow: I — Estructura, dinámica, restricciones y operadores
16. Flow: II — Operadores personalizados y manejo de excepciones
17. Pruebas unitarias de coroutines y de funciones suspendibles
18. Anexo: Cancelación cooperativa, Coroutines perezosas y Canales de difusión

A diferencia de los canales que son estructuras de flujo de datos “en caliente” — ‘hot stream’ en Inglés — , Flow provee un flujo de datos “en frío” — ‘cold stream’ en Inglés — . ¿Qué quiere decir ésto? Se considera una estructura de flujo de datos “en caliente” a la producción y emisión de elementos sin importar si éstos son o no son requeridos. Ésto quiere decir que tan pronto se crea la fuente o productor, los elementos se empiezan a producir y emitir. En una estructura de flujo de datos “en frío”, por el contrario, los elementos son producidos y emitidos bajo demanda. Ésto quiere decir que los elementos se empiezan a producir y emitir hasta que el consumidor lo requiera.

Un flujo de datos Flow consta de una estructura de 3 partes:

1. Creación del flujo de datos
2. Operadores intermedios
3. Operadores terminales

Antes de empezar a crear Flows y entrar de lleno con los operadores, debo establecer algunas restricciones que se deben tomar en cuenta para una correcta implementación. Además debo proveerte de las herramientas más básicas para ir avanzando poco a poco con los diferentes operadores que veremos en este artículo.

Lo primero que haré será mostrarte un Flow básico y explicarte cómo funciona:

Se crea un Flow con una llamada a la función flow (línea 1). Para crear un Flow se debe hacer uso de alguna de las funciones que permiten la construcción de Flows — más adelante en este artículo te mostraré algunas de ellas — . Para emitir elementos a través del flujo de datos, contamos con la función emit, pasándole como parámetro el elemento en cuestión (líneas 2 y 4). El bloque de código de la función flow se ejecuta de manera secuencial, línea por línea, en el contexto o Scope correspondiente al contexto o Scope de la coroutine que invoca al operador terminal que activa el Flow, siempre y cuando no haya un operador intermedio que cambie dicho contexto. En el caso del ejemplo, la coroutine que invoca al operador terminal collect (línea 9) es la creada con el constructor runBlocking, por lo tanto, el código que está dentro del bloque collect e incluso el código dentro del bloque flow — aunque esté “afuera” de la coroutine — forman parte de su Scope. El operador collect es llamado operador terminal porque permite que el Flow inicie la emisión de los elementos a través del flujo de datos. Además del operador collect existen otros operadores terminales que podremos usar según nuestra conveniencia — más adelante en este artículo te mostraré algunos de ellos —. Sin un operador terminal, el flujo de datos no se iniciará, por esta razón un Flow es un flujo de datos “en frío”.

Para una correcta implementación de un Flow, debes respetar las siguientes dos restricciones:

1. Preservación del contexto.
2. Transparencia para las excepciones.

— Preservación del contexto: Un Flow debe preservar el contexto en el que la función flow es invocada y hacer todas las emisiones dentro de dicho contexto. Por ejemplo, no debes hacer algo como lo siguiente:

Violación de la preservación del contexto con withContext.

En la línea 2 cambiamos de contexto con una llamada a la función withContext. Esta acción viola la preservación de contexto y de hecho al intentar hacerlo, el compilador lanza un error. Tampoco deberías hacer algo como ésto:

Violación de la preservación del contexto con GlobalScope.launch.

En este caso en la línea 2 estamos cambiando de contexto al crear una coroutine utilizando el objeto GlobalScope. El código anterior no solamente está mal por cambiar de contexto, sino que adentro del constructor flow no deberías crear coroutines. Si deseas emitir elementos que obtienes desde más de una fuente y de manera concurrente, podrías usar el constructor channelFlow explicado más adelante.

Por último, debes saber que para hacer un cambio de contexto adecuadamente a lo largo del flujo de datos, existe un operador intermedio llamado flowOn el cual veremos en la sección de operadores intermedios.

— Transparencia para las excepciones: Cualquier excepción que sea lanzada dentro del bloque flow no debería ser atrapada o manejada por un bloque try-catch, sino que se debe permitir que sea propagada a través del flujo de datos para que pueda ser capturada por algún operador catch que se encuentre en el camino. En el siguiente artículo profundizaré en el manejo de excepciones dentro de un Flow, y explicaré el operador catch con más detalle. Por ahora solo debes saber que no se permite hacer algo como lo siguiente:

Violación de la transparencia para las excepciones.

Aunque el código anterior es correcto porque atrapa la excepción y finaliza sin botar el programa, manejar las excepciones de esta manera podría exponer el flujo de datos a comportamientos extraños haciéndolo más propenso a errores y provocando que sean más difícil de localizar y corregir. Aunque no es obligatorio respetar la transparencia para las excepciones, se espera que en un futuro sea forzado mediante notificaciones de error en tiempo de compilación o con lanzamiento de errores en tiempo de ejecución.

Contexto del flujo de datos

Para tener más claro el funcionamiento de un Flow y el contexto en el que se ejecuta, vamos a hacer un experimento. Vamos a crear un Flow usando la función flow. Seguidamente vamos a crear 2 coroutines independientes entre sí con el constructor launch que hagan uso de la misma instancia del Flow creado. El código quedaría de la siguiente manera:

Al ejecutar el código anterior se obtiene la siguiente salida:

Observa que podemos usar el mismo Flow tantas veces como sea necesario aplicándole un operador terminal (collect en este caso), provocando que se emitan los elementos de inicio a fin todas las veces. La emisión de elementos dentro de ambas coroutines se puede dar de manera concurrente sin ningún problema dado que a pesar de tratarse del mismo Flow, ambas coroutines son independientes, por lo tanto, las emisiones serán independientes bajo el respaldo del Scope de cada coroutine que invoca al operador terminal.

Tomando en cuenta las restricciones anteriores y teniendo la visión más clara sobre la dinámica que se da entre una coroutine y un Flow, ya podemos entrar de lleno a ver las 3 partes que componen un Flow y sus operadores más comunes.

1. Creación del flujo de datos

Por lógica antes de trabajar con un flujo de datos debe haber una fuente de donde se extraigan o produzcan los elementos que serán emitidos. Existen varias maneras de crear un Flow y para ello nos valemos de funciones como las siguientes:

• flow {...}: Construye un flujo de datos arbitrario con emisión secuencial llamando a la función emit dentro del bloque de código.

Ésta es la única función que por el momento hemos usado para construir Flows, por lo que ya se te debe hacer familiar.

Al ejecutar el código anterior se obtiene la siguiente salida:

• flowOf(...): Crea un Flow a partir de un conjunto de elementos establecidos como parámetros de la función.

Al ejecutar el código anterior se obtiene la siguiente salida:

• asFlow(): Función de extensión de varios tipos de datos que les permite ser convertidos en un Flow.

Observa que a la lista que contiene Strings con los nombres de los planetas, se le aplica la función de extensión asFlow para crear un Flow con los elementos de la lista.

Al ejecutar el código anterior se obtiene la siguiente salida:

• channelFlow {...}: Construye un flujo de datos arbitrario con emisión potencialmente concurrente llamando a la función send dentro del bloque de código. Los elementos son enviados a través del ProducerScope que es en esencia un SendChannel — si observas la declaración de la interfaz ProducerScope te darás cuenta de que implementa la interfaz SendChannel — . La documentación oficial establece que la función channelFlow es thread-safe y asegura la preservación de contexto. Esto nos da la posibilidad de producir elementos desde varias fuentes de manera concurrente y desde contextos distintos.

Observa que a diferencia del constructor flow, dentro del bloque de código del constructor channelFlow sí se nos permite crear coroutines posibilitando la emisión de elementos de manera concurrente desde distintas coroutines, y aunque se utiliza un contexto diferente en cada una, este constructor asegura la preservación de contexto, como se indica en su documentación oficial. Otra diferencia que se puede observar es la emisión de elementos con llamadas a la función send en lugar de la función emit. La función send permite que el subproceso se suspenda en caso de que el buffer del SendChannel (ProducerScope) esté lleno.

Al ejecutar el código anterior se obtiene la siguiente salida:

2. Operadores intermedios

Los operadores intermedios son funciones que permiten manipular y/o trabajar con los elementos emitidos a través del flujo de datos antes de que lleguen al operador terminal. Puede haber uno o varios operadores intermedios encadenados unos con otros a lo largo de todo el recorrido. Algunos de los operadores intermedios más comunes son funciones como las siguientes:

• map {...}: Permite transformar los elementos que recibe desde el flujo de datos para luego encauzar hacia el flujo de datos los elementos ya modificados.

A partir del Flow creado se aplica una transformación con la función map. El bloque de la función map calcula el valor del elemento multiplicado por sí mismo y el resultado lo concatena con el String “Squared Value:” para luego encauzar el nuevo String hacia el flujo de datos.

Al ejecutar el código anterior se obtiene la siguiente salida:

• transform {...}: Al igual que el operador map, este operador permite transformar los elementos que recibe desde el flujo de datos para luego encauzar hacia el flujo de datos los elementos ya modificados. A diferencia del operador map que por cada elemento que recibe puede encauzar solamente un elemento, el operador transform puede encauzar más de un elemento por cada elemento recibido.

A partir de dos Strings se crea un Flow con la función flowOf. Se toma el Flow y se le aplica el operador transform. Dentro del bloque código del operador transform, por cada elemento recibido se divide el String en palabras con la función de extensión split, obteniendo así una lista de Strings con cada palabra que conforma el elemento que se recibió. Con un ciclo for se recorre la lista de palabras y con la función emit se encauzan al flujo de datos.

Al ejecutar el código anterior se obtiene la siguiente salida:

• flattenMerge(): A partir de un Flow de Flows Flow<Flow<T>> se procesa cada Flow y se produce un Flow de un solo nivel que contiene todos los elementos de cada uno de los Flow, encauzando cada elemento nuevamente hacia el flujo de datos. La función flattenMerge acepta un parámetro opcional llamado concurrency que indica la cantidad máxima de Flows que podrán ser procesados concurrentemente. Si no se le indica ningún valor, tomará el valor por defecto correspondiente a la constante DEFAULT_CONCURRENCY (16).

Se crean 3 Flows de números enteros, cada uno con 3 elementos con múltiplos de 2 para el primer Flow, múltiplos de 3 para el segundo Flow y múltiplos de 5 para el tercer Flow. Se crea también un Flow de Flows que contiene los 3 Flows de números enteros. Se toma el Flow de Flows y se le aplica el operador flattenMerge para producir un Flow de un solo nivel que contiene todos los elementos de cada uno de los Flows.

Al ejecutar el código anterior se obtiene la siguiente salida:

• flatMapMerge {...}: A partir de un Flow de Flows Flow<Flow<T>> se procesa cada Flow y se produce un Flow de un solo nivel que contiene todos los elementos de cada uno de los Flow, encauzando cada elemento ya modificado nuevamente hacia el flujo de datos. La función flatMapMerge acepta un parámetro opcional llamado concurrency que indica la cantidad máxima de Flows que podrán ser procesados concurrentemente. Si no se le indica ningún valor, tomará el valor por defecto correspondiente a la constante DEFAULT_CONCURRENCY (16). El operador flatMapMerge es un atajo de aplicar los operadores map y flattenMerge (map(transform).flattenMerge(concurrency)).

Se crean 3 Flows de números enteros. El primer Flow contiene múltiplos de 2, el segundo Flow contiene múltiplos de 3 y el tercer Flow contiene múltiplos de 5. Se crea también un Flow de Flows que contiene los 3 Flows de números enteros. Se toma el Flow de Flows y se le aplica el operador flatMapMerge para producir un Flow de un solo nivel que contiene todos los elementos de cada uno de los Flows modificados y cuya modificación consiste en su valor inicial al cuadrado, vale decir, multiplicado por sí mismo.

Al ejecutar el código anterior se obtiene la siguiente salida:

• filter {...}: Permite aplicar un filtro a los elementos que recibe desde el flujo de datos para luego volver a encauzar los elementos que superaron la condición de filtrado, hacia el flujo de datos.

Utilizando un Set que contiene el nombre de algunos colores, se crea un Flow con la función de extensión asFlow. Seguidamente se toma el flujo de datos y se le aplica un filtro por medio de la función filter, dejando pasar solamente los nombres de los colores que terminan con la letra ‘e’.

Al ejecutar el código anterior se obtiene la siguiente salida:

• distinctUntilChanged(): Desecha todos los elementos recibidos que son iguales al elemento más reciente. Esto quiere decir que si se emite el mismo elemento de manera consecutiva, solamente dejará pasar el elemento la primera vez desechando los demás elementos que se reciban hasta que el elemento recibido sea distinto.

Se crea un Flow a partir de un conjunto de número enteros. Se toma el Flow y se le aplica el operador distinctUntilChanged para dejar pasar los elementos que son distintos al elemento más reciente recibido. Es decir, de la secuencia 7, 7, 7 podrá pasar solamente el primer 7, y lo mismo deberá suceder con la secuencia 4, 4, 4 y la secuencia 8, 8.

Al ejecutar el código anterior se obtiene la siguiente salida:

• take(n): Deja pasar los n primeros elementos que recibe desde el flujo de datos. Cancela la emisión de datos cuando se ha alcanzado la cantidad de elementos especificados como parámetro.

Se crea un Flow con una llamada a la función channelFlow. Dentro del bloque del channelFlow se crean dos coroutines con el constructor launch para emitir elementos desde cada una de manera concurrente. En la primera coroutine se crea un rango de 1 a 100 que será recorrido con un ciclo forEach. Por cada iteración dentro del ciclo, se detiene el subproceso por 1 segundo con una llamada a la función delay, y se envía el número correspondiente con una llamada a la función send. En la segunda coroutine se crea una lista con los números 11, 22, 33, 44, 55, 66, 77, 88 y 99 y se recorre con un ciclo forEach. Por cada iteración dentro del ciclo, se detiene el subproceso durante medio segundo con una llamada a la función delay y se envía el número correspondiente con una llamada a la función send. Finalmente se toma el flujo de datos y se le aplica el operador intermedio take para obtener solamente los primeros 5 elementos emitidos.

Al ejecutar el código anterior se obtiene la siguiente salida:

• zip(Flow) {...}: Combina pares de elementos obtenidos desde dos flujos de datos diferentes, de manera intercalada, es decir que por cada elemento recibido desde uno de los Flows, se debe esperar la emisión de otro elemento desde el otro Flow para ejecutar el bloque de código. Dentro del bloque de código se procesan los elementos para obtener un nuevo elemento que será encauzando hacia el flujo de datos. El operador se mantendrá en ejecución hasta que uno de los dos Flows finalice la emisión de elementos, cancelando inmediatamente el otro Flow.

Se crean dos Flows independientes. El primer Flow emite los nombres de los 8 planetas del Sistema Solar. En el bloque del segundo Flow se crea un rango que va desde el número 11 al 1000 y se recorre con un ciclo forEach dando saltos de 11 en 11 emitiendo el número correspondiente en cada repetición con una pausa de 250 milisegundos entre cada emisión. Finalmente se toma el primer Flow y se le aplica el operador intermedio zip con el segundo Flow como parámetro combinando así las emisiones de ambos Flows. Dentro del bloque de código de la función zip se combinan ambos elementos emitidos produciendo así un String que continuará hacia abajo en el flujo de datos.

Al ejecutar el código anterior se obtiene la siguiente salida:

• combine(Flow) {...}: Combina pares de elementos obtenidos desde dos flujos de datos diferentes tan pronto son recibidos, es decir que cada elemento recibido desde uno de los Flows se combina con el elemento más reciente recibido desde el otro Flow e inmediatamente se ejecuta el bloque de código. Dentro del bloque de código se procesan los elementos para obtener un nuevo elemento que será encauzado hacia el flujo de datos. El operador se mantendrá en ejecución hasta que ambos Flows finalicen la emisión de elementos.

Se crean dos Flows independientes. El primer Flow emite los elementos de una lista que contiene el nombre de los 8 planetas del Sistema Solar con una pausa de 400 milisegundos entre cada emisión. En el bloque del segundo Flow se crea un rango que va desde el número 1 al 10 y se recorre con un ciclo forEach emitiendo el número correspondiente en cada repetición con una pausa de 250 milisegundos entre cada emisión. Finalmente se toma el primer Flow y se le aplica el operador intermedio combine con el segundo Flow como parámetro combinando así las emisiones de ambos Flows. Dentro del bloque de la función combine se combinan ambos elementos emitidos produciendo así un String que continuará hacia abajo en el en flujo de datos.

Al ejecutar el código anterior se obtiene la siguiente salida:

• onEach {...}: Es un paso intermedio que permite realizar operaciones con los elementos que se están emitiendo para luego encauzarlos íntegros nuevamente hacia el flujo de datos.

Se crea un Flow a partir de una secuencia de Strings con los nombres de los planetas. Se toma el Flow y se le aplica el operador intermedio onEach. Dentro del bloque de código del operador onEach se interrumpe el subproceso durante 200 milisegundos con una llamada a la función delay y seguidamente imprime en pantalla el nombre del planeta que está siendo emitido en ese momento.

Al ejecutar el código anterior se obtiene la siguiente salida:

No confundas el operador intermedio ‘onEach’ con el operador terminal ‘collect’. Aunque son similares, el operador ‘collect’ activa el flujo de datos (por ser un operador terminal) mientras que el operador ‘onEach’ no lo hace. Además el operador ‘onEach’ vuelve a encauzar todos los elementos íntegros hacia el flujo de datos, mientras que el operador ‘collect’ no lo hace.

• flowOn(CoroutineContext): Cambia el contexto de la coroutine en el que se ejecutan todos los operadores que le preceden. Si el CoroutineContext especificado contiene un Job, se lanza una excepción de tipo IllegalArgumentException. Si al especificar el contexto se cambia de Dispatcher, se rompe la naturaleza secuencial en la ejecución dentro del flujo de datos.

Este código es exactamente igual al ejemplo anterior. La llamada al operador flowOn cambia el contexto en el que se ejecutan los operadores que le preceden. Para efectos del ejemplo, el bloque de código del operador onEach sería ejecutado por un hilo perteneciente al pool de hilos que corresponde a Dispatchers.Default y dado que hay un cambio de Dispatcher, podrás observar que la ejecución secuencial de los operadores se pierde.

Al ejecutar el código anterior se obtiene la siguiente salida:

• buffer(n): Permite la ejecución concurrente dentro del flujo de datos dividiéndolo en dos partes. La coroutine que llama al operador terminal se adjudica la ejecución del flujo de datos desde el operador buffer hasta el operador terminal. La ejecución del flujo de datos por encima del operador buffer es ejecutada por una nueva coroutine. El argumento n indica el tamaño específico de elementos que se desea que se procesen de manera concurrente. Si no se especifica un tamaño, el buffer tendrá el tamaño por defecto que equivale a Channel.BUFFERED (64). Otros valores que se pueden asignar por medio de constantes son los siguientes: Channel.CONFLATED, Channel.RENDEZVOUS y Channel.UNLIMITED. Este es el operador necesario para controlar el comportamiento en casos en el que la frecuencia de emisión de elementos sobrepasa la frecuencia en la que el consumidor los procesa. A esta dinámica se le conoce como ‘Backpressure’ y al aplicar la técnica adecuada se mejora el rendimiento y la eficacia del Flow.

Este código es exactamente igual al ejemplo del operador onEach con la diferencia de que se le encadena al final del operador onEach una llamada al operador buffer. El operador buffer crea una nueva coroutine para ejecutar todo lo que está sobre él de manera concurrente separado de la ejecución de todo lo que está debajo de él. Para efectos de éste ejemplo, todo lo que se encuentra dentro del bloque de código del operador onEach será ejecutado por una nueva coroutine, mientras que el código del bloque de código del operador collect será ejecutado por la coroutine que activó el Flow al invocar al operador terminal. La ejecución concurrente de ambas coroutines tendrá un impacto positivo en el tiempo de ejecución total del Flow.

Al ejecutar el código anterior se obtiene la siguiente salida:

• conflate(): El operador conflate es un atajo del operador buffer con capacidad Channel.CONFLATED (buffer(Channel.CONFLATED)). Si la coroutine que ejecuta el código que está por debajo del operador no puede seguirle el paso a la coroutine que ejecuta el código que está por encima del operador, se va desechando el exceso de elementos manteniendo siempre el más reciente en el buffer.

Como mencioné anteriormente, el operador conflate equivale al operador buffer con una capacidad de Channel.CONFLATED. Puedes hacer la prueba cambiándolo, el resultado que obtengas debería ser casi idéntico al siguiente:

3. Aplicación de operadores terminales

Los operadores terminales son funciones que activan o inician la producción y emisión de elementos. Al conectar un Flow con un operador terminal, el operador inicial o constructor de Flow comienza a emitir los elementos producidos. Dichos elementos entonces se desplazan a través del flujo de datos pasando por cada operador intermedio que se encuentre a su paso, posibilitando así la aplicación de filtros, transformaciones, modificaciones de contextos y/o comportamientos hasta desembocar en el propio operador terminal. Algunas de las funciones terminales más comunes son las siguientes:

• collect {...}: Activa el flujo de datos y procesa los elementos recibidos según se vayan recibiendo.

Como ya hemos visto este operador en todos los ejemplos, te voy a mostrar otra manera de utilizar este mismo operador.

Esta versión del operador collect cumple exclusivamente con la tarea de activar el flujo de datos. Al aplicarlo de esta manera se inicia la producción y emisión de los elementos pasando por el operador map que transforma el número recibido en su valor al cuadrado. Seguidamente se pasa el nuevo valor al operador onEach que lo imprime en pantalla. Después del operador onEach se aplica el operador flowOn para cambiar el contexto por otro que utilice el pool de hilos correspondiente a Dispatchers.Default y que afectará a todos los operadores que están sobre él e incluso al constructor flow.

Al ejecutar el código anterior se obtiene la siguiente salida:

• collectLatest {...}: Activa el flujo de datos y procesa los elementos según se vayan recibiendo, cancelando el procesamiento del último elemento recibido si éste aún está siendo procesado.

Se crea un Flow que emite los números del 1 al 5, con pausas de 200 milisegundos entre cada emisión excepto por la tercera pausa que es de 300 milisegundos. Se toma el Flow y se le aplica el operador collectLatest imprimiendo en pantalla el elemento recibido 2 veces con una pausa de 250 milisegundos entre cada impresión. La pausa es necesaria para poder observar que la ejecución del bloque de código es interrumpida para darle paso a una nueva ejecución del bloque de código tan pronto se reciba un nuevo elemento.

Al ejecutar el código anterior se obtiene la siguiente salida:

• single(): Espera recibir solamente un elemento. Si no se emite ningún elemento, lanza una excepción de tipo NoSuchElementException. Si se emite más de un elemento, lanza una excepción de tipo IllegalStateException.

A partir de un Flow que emite el String “Hello Flow!” se le aplica el operador single y se almacena el elemento en una variable. Seguidamente se muestra en pantalla el valor de la variable.

Al ejecutar el código anterior se obtiene la siguiente salida:

• first(): Activa el Flow y procesa el primer elemento que es emitido, cancelando el Flow después de haberlo recibido. Si el Flow no contiene elementos se lanza una excepción de tipo NoSuchElementException.

A partir de un Flow que contiene 5 números enteros se le aplica el operador first y se almacena el elemento en una variable. Seguidamente se muestra en pantalla el valor de la variable.

Al ejecutar el código anterior se obtiene la siguiente salida:

• reduce { accumulator, value -> ...}: Procesa los elementos uno por uno según se vayan recibiendo. La variable accumulator se inicializa con el primer elemento recibido y a partir de ahí por cada elemento recibido se le aplica la operación definida dentro de su bloque de código manteniendo siempre el nuevo resultado en la variable accumulator.

El operador terminal reduce es un poco complicado de entender inicialmente. Cuenta con dos variables, la variable accumulator y la variable value, además del bloque de código. La variable accumulator se actualizará cada vez que se ejecute el bloque de código. La variable value contiene el elemento recibido. Cuando se ejecuta el bloque de código, el resultado de la operación se almacena en la variable accumulator. El primer elemento recibido se almacena en la variable accumulator directamente.

El código de este ejemplo muestra cómo calcular el factorial. La operación del bloque de código consiste en la multiplicación del valor que contiene la variable accumulator por el valor que contiene la variable value. El resultado de esa operación se almacena en la variable accumulator, dejándola así lista para el siguiente elemento que se reciba.

Al ejecutar el código anterior se obtiene la siguiente salida:

• toList(): Retorna una lista con los elementos que fueron recibidos.

El operador terminal toList activa el Flow provocando que se empiece a producir y emitir los elementos. Al finalizar la emisión de todos los elementos, solamente los que superen el filtro serán almacenados dentro de la lista resultante.

Al ejecutar el código anterior se obtiene la siguiente salida:

El código de este ejemplo muestra la creación de una lista inmutable a partir de la llamada a la función toList. Si se desea almacenar los elementos en una lista mutable se debe crear primero la lista y luego hacer la llamada a la función toList pasándole la lista como parámetro, como muestro a continuación:

Así como existe la función toList para crear listas a partir de los elementos de un flujo de datos, también existen las funciones toSet para crear un Set y toCollection para crear algún otro tipo de estructura que necesites, como por ejemplo un Queue o un Stack.

• launchIn(CoroutineScope): Inicia una coroutine en el Scope especificado como parámetro e inmediatamente activa el Flow. Retorna un Job tal y como lo hace el constructor de coroutines launch.

Se crean dos coroutines con el operador launchIn y diferentes Scopes a partir del mismo Flow. La primera coroutine activa el Flow con el Scope de la coroutine creada con el constructor runBlocking (this). La segunda coroutine activa el Flow con el Scope de una clase anónima referenciada con la variable myScope.

Al ejecutar el código anterior se obtiene la siguiente salida:

• produceIn(CoroutineScope): Inicia una coroutine en el Scope especificado como parámetro e inmediatamente activa el Flow. Retorna un ReceiveChannel tal y como lo hace el constructor de coroutines produce.

Se crean dos coroutines con el operador produceIn y diferentes Scopes a partir del mismo Flow. La primera coroutine activa el Flow con el Scope de la coroutine creada con el constructor runBlocking (this). La segunda coroutine activa el Flow con el Scope de una clase anónima referenciada con la variable myScope. Además se crean dos coroutines con el constructor launch para consumir de manera concurrente los elementos de cada ReceiveChannel obtenido.

Al ejecutar el código anterior se obtiene la siguiente salida:

Además de todos los operadores expuestos en este artículo, existen otros que resuelven otras situaciones que podrías encontrarte cuando trabajes con flujos de datos. La lista de operadores la puedes encontrar en la documentación oficial de Flow. También tienes la posibilidad de crear tus propios operadores personalizados según lo que necesites en caso de que no exista un operador que ya lo haga.

Por ahora con lo que hemos visto en este artículo es suficiente para que lo pongas en práctica y domines su dinámica, sin embargo, es sumamente necesario que estés preparado para cualquier contingencia, es decir, debes también aprender a manejar excepciones correctamente, sin violar la “transparencia para las excepciones”. En el siguiente artículo te mostraré cómo crear tus propios operadores personalizados y además profundizaré en el manejo de excepciones.

Continúa en el siguiente enlace con Flow: II— Operadores personalizados y manejo de excepciones.

• Puedes acceder a la documentación oficial de Flow en el siguiente enlace: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-flow/
• Puedes acceder a la documentación oficial del uso de Flow en el siguiente enlace: https://kotlinlang.org/docs/reference/coroutines/flow.html