El contrato de contexto

Los flujos son la parte de las corrutinas donde el lenguaje se pone estricto. En casi todo lo demás, escribir algo dudoso produce como mucho un aviso o un comportamiento discutible; aquí hay dos invariantes que la propia librería comprueba en tiempo de ejecución y cuyo incumplimiento aborta el programa con un mensaje explícito. La primera dice que el productor debe emitir desde la misma corrutina desde la que se le llamó. La segunda dice que ningún tramo de la cadena puede tragarse una excepción que venga de más abajo. Las dos parecen restricciones caprichosas hasta que se entiende que son la contrapartida exacta de las virtudes de las lecciones anteriores: son el precio de que la cadena entera sea una sola unidad de ejecución legible, y sin ellas nada de lo dicho hasta aquí sería cierto.

La invariante de preservación del contexto

Empecemos por el código que parece razonable y no lo es. Un productor que hace trabajo de disco quiere ejecutarlo en el despachador adecuado, y la forma habitual de conseguirlo en corrutinas es envolver ese trabajo.

// Falla en ejecucion con un mensaje sobre la invariante del flujo
val roto = flow {
    withContext(Dispatchers.IO) {
        emit(leerFichero())
    }
}

El mecanismo de detección es sencillo y conviene conocerlo porque explica la regla mejor que cualquier justificación. El colector que la librería entrega al bloque productor no es el del usuario directamente, sino un envoltorio que guarda el contexto vigente en el instante en que empezó la recolección. En cada llamada a emit, ese envoltorio consulta el contexto actual y lo compara con el guardado; si el trabajo asociado no es el mismo, lanza. No hay análisis estático ni magia del compilador: hay una comparación por emisión, con el resultado memorizado para que el coste sea despreciable cuando nada cambia.

La razón de fondo es la que sostiene las tres lecciones anteriores. emit es una llamada directa al cuerpo del recolector, sobre la misma pila. Si esa llamada ocurriera desde otra corrutina, el cuerpo del usuario se ejecutaría en un contexto que él no eligió, su cancelación colgaría de un trabajo que él no conoce y sus excepciones subirían por una cadena que no es la suya. Es decir, se perderían de golpe las cuatro propiedades que hacían que la cadena se leyera como una función ordinaria.

flowOn y lo que construye

Si el bloque no puede cambiar de contexto por dentro, alguien tiene que poder cambiarlo por fuera, y ese alguien es un operador que se coloca después del tramo al que afecta.

val bien = flow { emit(leerFichero()) }   // se ejecutara en IO
    .map { parsear(it) }                   // esto tambien
    .flowOn(Dispatchers.IO)                // afecta a todo lo de arriba
    .map { presentar(it) }                 // esto ya no: va en el contexto del recolector

La dirección desconcierta la primera vez y es perfectamente coherente con todo lo anterior. Como una cadena se construye de abajo hacia arriba, cuando alguien recolecta el flujo devuelto por el operador, ese operador es quien decide en qué contexto va a recolectar lo que tiene encima. Sobre lo que está debajo no tiene ninguna autoridad, porque cuando se ejecutó ya existía y no lo conoce.

Lo que construye por dentro es exactamente lo que la lección anterior describía. Lanza una corrutina nueva con el contexto pedido, recolecta allí todo el tramo superior, y transporta los valores hasta la corrutina del recolector a través de un canal. De ahí se siguen tres hechos prácticos: cambiar de contexto implica siempre un búfer, dos operadores adyacentes de esta familia se fusionan en un solo canal, y el coste de un cambio de contexto en una cadena no es despreciable, así que conviene tener uno y no cinco.

flowchart TD
A[Bloque productor] --> B[map de arriba]
B --> C[flowOn con el despachador pedido]
C --> D[Canal de transporte]
D --> E[map de abajo]
E --> F[Cuerpo del recolector]
C -.-> G[Todo lo de arriba corre en el contexto dado]
D -.-> H[Todo lo de abajo corre donde se llamo a collect]

Hay dos restricciones más que conviene recordar. La primera es que el contexto que se le pasa no puede contener un trabajo: si se intenta, el operador rechaza el argumento, porque la jerarquía de trabajos la decide la concurrencia estructurada y no un operador de flujo. La segunda es que este operador no puede en ningún caso cambiar dónde se ejecuta la operación terminal; eso lo decide quien recolecta, envolviendo su llamada o lanzándola en el ámbito que corresponda.

// Quien recolecta decide su propio contexto, no el flujo
withContext(Dispatchers.Main) {
    datos.collect { pintar(it) }
}

Transparencia de excepciones

La segunda invariante es simétrica de la primera y suele explicarse mal. Dice que un flujo no puede capturar excepciones que provengan de tramos situados por debajo de él, es decir, del cuerpo del recolector o de los operadores posteriores. La razón es que un fallo del consumidor no es un fallo del flujo, y tragárselo convertiría un error del usuario en una emisión aparentemente normal.

// Viola la transparencia: captura tambien lo que falle abajo
val opaco = flow {
    try {
        emit(1)                 // si el recolector lanza, cae aqui
    } catch (e: Exception) {
        emit(-1)                // emitir desde el catch es lo prohibido
    }
}

La herramienta que respeta la invariante es un operador dedicado. Captura únicamente lo que ocurre por encima de él en la cadena, distingue el origen del fallo y, como su bloque tiene el colector por receptor, permite emitir valores de reemplazo.

peticiones
    .map { descargar(it) }
    .catch { causa ->
        registrar(causa)
        emit(Respuesta.Vacia)     // valor de repuesto, legitimo aqui
    }
    .collect { mostrar(it) }

La posición en la cadena es la mitad del significado. Colocado ahí, cubre el bloque productor y el mapeo, pero no cubre lo que ocurra dentro del recolector final. Si se colocara justo después del constructor, no cubriría el mapeo. Y hay un tipo de excepción que nunca captura, deliberadamente: la de cancelación, porque cancelar no es fallar y tragarse esa señal rompería la concurrencia estructurada exactamente igual que en cualquier otra corrutina.

Ese último operador merece una precisión porque se usa a menudo como si fuera un finally y no lo es del todo. Se ejecuta en los tres desenlaces y recibe la causa como parámetro nulo cuando todo fue bien, pero no la consume: la excepción sigue su camino hacia abajo después de que su bloque termine. Es el sitio correcto para registrar, medir o apagar un indicador de carga, y el sitio incorrecto para intentar recuperarse.

Reintentar es recolectar otra vez

Aquí se cierra el círculo con la primera lección. Un flujo es una receta que no ha avanzado a ningún sitio, y por eso reintentarlo no requiere rebobinar nada ni recordar por dónde iba: basta con volver a llamar al único método que tiene.

descarga
    .retry(3) { causa -> causa is IOException }
    .catch { emit(Respuesta.Vacia) }
    .collect { procesar(it) }

La variante general recibe la causa y el número de intento, permite decidir con ambos y, como su bloque es suspendido, permite esperar antes de volver a intentarlo. Esa combinación es todo lo que hace falta para escribir una espera creciente sin ninguna librería adicional.

descarga.retryWhen { causa, intento ->
    if (causa is IOException && intento < 5) {
        delay(200L * (intento + 1))   // espera creciente
        true                          // reintentar
    } else false
}

Conviene tener claro qué se reintenta exactamente: todo el tramo situado por encima del operador, desde su primera línea. Si ese tramo tiene efectos secundarios, se repetirán; si abre recursos, los abrirá de nuevo. Por eso la posición del operador de reintento en la cadena es una decisión semántica y no estética, y por eso los flujos que envuelven una sola llamada suspendida, que en la segunda lección parecían gratuitos, resultan aquí perfectamente justificados: convierten una operación en algo que se puede repetir entera con una línea.