Transformaciones entre Flow y Channel

En Kotlin, Flow y Channel representan dos formas distintas de manejar secuencias de datos. Mientras que Flow es un flujo perezoso (cold stream) que comienza a emitir únicamente cuando un colector se suscribe, Channel actúa como un conducto de comunicación activo (hot stream). Kotlin proporciona varias funciones para convertir entre ambos modelos según las necesidades de cada caso.

De Flow a Channel

ChannelFlow como base de conversión

La clase abstracta ChannelFlow sirve como puente interno entre un Flow y un Channel. Aunque está marcada como API interna y no debe usarse directamente, es importante conocer su funcionamiento: recolecta los valores emitidos por el Flow original y los envía a un ReceiveChannel. A partir de este mecanismo se derivan las funciones públicas de conversión.

produceIn: Flow a Channel unicast

produceIn crea una corrutina productora que consume el Flow original y emite los elementos a un ReceiveChannel. El canal resultante mantiene la capacidad de manejar backpressure. Es responsabilidad del desarrollador cancelar o cerrar la corrutina productora cuando ya no se necesite.

fun main() = runBlocking {
    val source = flow {
        (1..5).forEach {
            delay(300)
            emit(it * 10)
        }
    }

    val channel = source.produceIn(this)
    for (value in channel) {
        println(value)
    }
}

Salida esperada:

10
20
30
40
50

La implementación interna se reduce a llamar a asChannelFlow().produceImpl(scope), lo que evidencia que el flujo resultante es caliente.

broadcastIn: Flow a BroadcastChannel

broadcastIn funciona de manera similar a produceIn, pero genera un BroadcastChannel que permite que varios consumidores reciban los mismos elementos. Sin embargo, tanto BroadcastChannel como broadcastIn están marcados como APIs obsoletas. Se recomienda reemplazarlos por SharedFlow o StateFlow.

De Channel a Flow

consumeAsFlow y receiveAsFlow

Para convertir un ReceiveChannel en Flow se utilizan dos extensiones: consumeAsFlow y receiveAsFlow. Ambas construyen internamente un ChannelAsFlow, por lo que el resultado es un flujo caliente.

fun main() = runBlocking {
    val channel = Channel<String>()
    val flow = channel.receiveAsFlow()

    launch {
        flow.collect { println(it) }
    }

    listOf("hola", "mundo", "coroutines").forEach {
        delay(120)
        channel.send(it)
    }

    channel.close()
}

La diferencia entre ambas radica en la propiedad consume que se pasa a ChannelAsFlow:

• consumeAsFlow(): el flujo solo puede ser recolectado una vez. Si se intenta usar varios colectores, se lanza IllegalStateException.
• receiveAsFlow(): permite múltiples colectores, pero cada elemento es entregado a un único consumidor (semántica unicast).

asFlow para BroadcastChannel

Un BroadcastChannel se puede convertir a Flow mediante asFlow(). Al igual que broadcastIn, esta API está obsoleta y su equivalente moderno es SharedFlow.

Compartir flujos con shareIn

shareIn convierte un Flow frío en un SharedFlow caliente, compartiendo una única suscripción subyacente entre todos los colectores. Su firma es la siguiente:

fun <T> Flow<T>.shareIn(
    scope: CoroutineScope,
    started: SharingStarted,
    replay: Int = 0
): SharedFlow<T>

El parámetro scope determina el ciclo de vida del flujo compartido; debe sobrevivir a todos los consumidores. replay indica cuántos elementos previos se reenvían a nuevos suscriptores. started controla cuándo comienza y finaliza la emisión:

• SharingStarted.Eagerly: inicia inmediatamente y nunca se detiene.
• SharingStarted.Lazily: inicia con el primer suscriptor y nunca se detiene.
• SharingStarted.WhileSubscribed(...): inicia con el primer suscriptor y se detiene cuando no queda ninguno.

callbackFlow: envolviendo APIs basadas en callbacks

callbackFlow permite transformar APIs que usan callbacks en flujos de Kotlin. Es especialmente útil cuando se trabaja con librerías que aún no exponen corrutinas.

Ejemplo básico

Supongamos la siguiente interfaz callback:

interface Callback<T> {
    fun onComplete(value: T)
    fun onError(message: String)
}

fun fetchData(callback: Callback<String>) {
    thread {
        Thread.sleep(1000)
        callback.onComplete("datos")
    }.start()
}

Se puede exponer como Flow de la siguiente manera:

fun fetchDataFlow(): Flow<String> = callbackFlow {
    val callback = object : Callback<String> {
        override fun onComplete(value: String) {
            trySend(value)
            close()
        }

        override fun onError(message: String) {
            close(RuntimeException(message))
        }
    }

    fetchData(callback)

    awaitClose {
        println("cerrando flujo")
    }
}

Caso práctico: búsqueda reactiva en Android

Un escenario común es escuchar los cambios de texto de un EditText, aplicar operadores de Flow y realizar peticiones. Primero se define la conversión del callback a flujo:

fun textChangeFlow(editText: EditText): Flow<String> = callbackFlow {
    val watcher = object : TextWatcher {
        override fun beforeTextChanged(s: CharSequence?, start: Int, count: Int, after: Int) {}
        override fun onTextChanged(s: CharSequence?, start: Int, before: Int, count: Int) {}
        override fun afterTextChanged(s: Editable?) {
            s?.toString()?.let { trySend(it) }
        }
    }

    editText.addTextChangedListener(watcher)

    awaitClose {
        editText.removeTextChangedListener(watcher)
    }
}

Luego se combina con el resto de la cadena de operadores:

scope.launch {
    textChangeFlow(editText)
        .debounce(300)
        .flatMapLatest { keyword ->
            search(keyword)
        }
        .collect { result ->
            // Mostrar resultados
        }
}

De esta forma, operadores como debounce, flatMapLatest o catch se integran de manera declarativa en el flujo de trabajo.