Qué es un Flow de verdad

Hay una manera rápida de comprobar si alguien ha entendido los flujos, y consiste en pedirle que escriba de memoria la declaración de Flow. Quien contesta que es una clase con muchos métodos no ha abierto nunca el fuente; quien contesta que es un tipo especial que el compilador conoce, tampoco. La respuesta correcta resulta casi decepcionante: es una interfaz pública con un solo método, y ese método está marcado como suspendido. Nada más. Todo lo que se cuenta después sobre frialdad, contrapresión, cancelación, operadores y contextos no son capacidades añadidas encima de esa declaración, sino consecuencias que se deducen de ella. Este nivel entero consiste en hacer esas deducciones despacio, y empieza por leer las dos únicas declaraciones que hacen falta.

Las dos declaraciones que hay que saberse

El fuente de la librería cabe, para este asunto, en diez líneas. Dos interfaces, un método cada una, y ambos métodos suspendidos.

public interface Flow<out T> {
    public suspend fun collect(collector: FlowCollector<T>)
}

public fun interface FlowCollector<in T> {
    public suspend fun emit(value: T)
}

La varianza no es decorativa y se lee sola en cuanto se pregunta quién produce y quién consume. Un Flow produce valores de tipo T y jamás los recibe, de modo que es covariante: donde se espera un flujo de un supertipo sirve uno de un subtipo. Un FlowCollector consume valores de tipo T y jamás los devuelve, de modo que es contravariante por la razón simétrica. Esta pareja covariante y contravariante aparece una y otra vez en el lenguaje, y aquí sostiene que puedas pasar un flujo de perros donde se pide uno de animales sin que nada se rompa.

La segunda interfaz está declarada como interfaz funcional, y de ahí sale la sintaxis que todo el mundo escribe a diario. Cuando alguien recolecta pasando una lambda no está usando el método del ejemplo anterior, sino una extensión en línea que fabrica el colector por él.

// La forma que se escribe
flujo.collect { valor -> println(valor) }

// Lo que ocurre por debajo, aproximadamente
flujo.collect(object : FlowCollector<Int> {
    override suspend fun emit(value: Int) { println(value) }
})

Merece la pena detenerse en que los dos métodos suspenden, porque de ahí sale casi todo lo demás. Que collect suspenda significa que recolectar es una llamada corriente que termina cuando el flujo se agota, no una suscripción que devuelve un manejador. Que emit suspenda significa que entregar un valor puede hacer esperar al productor. Con esas dos propiedades ya no hace falta inventar nada más.

De la firma se deduce la frialdad

Un valor de tipo Flow no contiene datos ni tiene nada en marcha. Lo único que sabe hacer es responder a collect, y hasta que alguien llama a ese método no ocurre absolutamente nada. Esa es toda la definición de frío: la ausencia de trabajo antes de la recolección no es una decisión de diseño añadida, es que no hay ningún otro método por el que el trabajo pudiera empezar.

val f = flow {
    println("empiezo")
    emit(1)
    emit(2)
}
// Hasta aqui no se ha impreso nada y no existe ningun valor

f.collect { println(it) }   // empiezo, 1, 2
f.collect { println(it) }   // empiezo, 1, 2  otra vez, desde cero

La segunda consecuencia es igual de mecánica. Como el único método recibe el colector como parámetro, cada llamada trae su propio colector y ejecuta el cuerpo entero otra vez. No hay estado compartido entre dos recolecciones porque no hay dónde guardarlo: el objeto Flow es, literalmente, una receta. Dos recolectores no se reparten los valores ni comparten progreso; obtienen cada uno su ejecución completa e independiente.

La tercera consecuencia es la que sostiene el resto del nivel. Si recolectar es ejecutar la receta desde el principio, entonces reintentar un flujo fallido es trivial: basta con volver a llamar a collect. El operador de reintento que veremos en la última lección no necesita almacenar nada ni rebobinar nada, porque el flujo no ha avanzado a ningún sitio del que haya que volver.

flowchart TD
A[Alguien llama a collect] --> B[Se ejecuta el bloque productor]
B --> C[emit entrega el valor al colector]
C --> D[Se ejecuta el cuerpo del recolector]
D --> E[emit retorna y el productor continua]
E --> B
B --> F[El bloque termina y collect retorna]

Por qué no hay protocolo de petición

Conviene comparar con lo que existía antes para medir cuánto se ha eliminado. La especificación clásica de flujos reactivos define cuatro interfaces, unos diez métodos entre todas ellas y un documento de reglas normativas que cualquier implementación debe cumplir para ser conforme. Hay señales de suscripción, de valor, de error y de terminación, y hay un método con el que el consumidor declara cuántos elementos está dispuesto a recibir. Toda esa maquinaria existe para resolver un solo problema: impedir que el productor vaya más rápido que el consumidor.

Aquí ese problema no se resuelve, se disuelve. Mientras el cuerpo del recolector se está ejecutando, la llamada a emit todavía no ha retornado, y por tanto el productor está detenido en esa línea. La demanda no se pide porque es implícita y siempre vale uno: el productor puede continuar exactamente cuando el consumidor ha terminado con el valor anterior. La palabra suspend en emit hace, sin protocolo alguno, lo que aquellas librerías hacen con un método dedicado y un contrato escrito.

Las otras señales desaparecen por motivos igual de simples. No hay una señal de terminación porque una función suspendida ya termina; no hay una señal de error porque una función suspendida ya lanza; no hay señal de suscripción porque no hay suscripción. Cada uno de los conceptos que la especificación reactiva tuvo que inventar existía ya en el lenguaje, y las corrutinas se limitaron a no duplicarlo.

La comparación se puede cerrar con una cuenta que resulta elocuente. Aquella especificación necesita reglas escritas para fijar que las señales no se solapen, que la terminación sea definitiva, que la cancelación sea idempotente y que la demanda no se pueda pedir en cantidad negativa. Ninguna de esas reglas hace falta aquí, porque todas ellas son consecuencias de que la producción ocurra dentro de una llamada suspendida ordinaria: dos emisiones no se solapan porque son dos líneas de la misma función, la terminación es definitiva porque una función que retornó no retorna dos veces, y no hay cantidad de demanda que validar porque no hay cantidad.

Todo lo demás son extensiones

Si la interfaz solo tiene un método, cualquier operador tiene que ser una función externa que fabrique un flujo nuevo. Y así es: el catálogo entero está escrito como funciones de extensión sobre Flow, sin acceso privilegiado a nada. Cualquiera podría haberlas escrito, y la prueba es que se escriben en tres líneas.

// Version conceptual, fiel a la idea aunque no al fuente literal
fun <T, R> Flow<T>.map(transform: suspend (T) -> R): Flow<R> = flow {
    collect { valor -> emit(transform(valor)) }
}

Léase despacio, porque ahí está el mecanismo entero del nivel. La función devuelve un flujo nuevo cuyo cuerpo, cuando alguien lo recolecte, recolectará el flujo original con un colector que transforma cada valor y se lo pasa al colector de abajo. Hay dos emit en juego que pertenecen a dos colectores distintos y el compilador los distingue por el receptor implícito de cada bloque. No se crea ninguna colección intermedia, no hay ningún búfer, y un valor recorre la cadena entera antes de que se produzca el siguiente.

Que la interfaz sea pública invita a una pregunta legítima: si solo tiene un método, por qué no implementarla directamente en una clase propia. La librería responde con una recomendación explícita y una anotación de advertencia. Implementar la interfaz a mano deja fuera las comprobaciones de contexto y de excepciones que estudiaremos en la última lección, y por eso existe una clase abstracta pensada para heredar de ella cuando alguien necesita de verdad un tipo propio en lugar de un valor construido.

class Sensor : AbstractFlow<Int>() {
    override suspend fun collectSafely(collector: FlowCollector<Int>) {
        collector.emit(leer())
    }
}

El nombre del método que hay que redefinir es la documentación de todo el asunto: recolectar con seguridad, es decir, con las invariantes comprobadas. En la práctica casi nunca hace falta, porque el constructor general ya hereda esas comprobaciones y una función que devuelve un flujo es más flexible que una clase que lo es.

De aquí sale una regla de lectura que sirve para todo lo que viene después. Cuando un operador te desconcierte, escribe su versión conceptual con flow y collect y verás que casi siempre cabe en cuatro líneas. Los que no caben son exactamente los interesantes, y son pocos: los que introducen una corrutina nueva, los que cancelan el productor y los que cambian el contexto. Esos tres grupos ocupan las lecciones tres, cuatro y cinco de este nivel.