Project Reactor – Flux<T>

A Flux is a Reactive Streams Publisher that emits 0 to N elements, and then completes (successfully or with an error). It is the multi-element reactive type in Project Reactor, analogous to Observable in RxJava or Flow in Kotlin Coroutines.

Flux is intended to be used in implementations and return types. Input parameters should keep using raw Publisher as much as possible. If you know the upstream will emit 0 or 1 element, use Mono instead.

import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.time.Duration

// Simulate a reactive microservice pipeline
data class Order(val id: String, val product: String, val quantity: Int)

fun fetchOrders(): Flux<Order> = Flux.just(
    Order("ORD-1", "Laptop", 1),
    Order("ORD-2", "Mouse", 5),
    Order("ORD-3", "Monitor", 2)
)

fun main() {
    // Build a reactive pipeline: fetch → filter → transform → subscribe
    fetchOrders()
        .filter { it.quantity > 1 }                    // keep bulk orders
        .map { "${it.id}: ${it.quantity}x ${it.product}" } // format
        .doOnNext { println("Processing: $it") }       // side-effect logging
        .subscribe(
            { value -> println("Fulfilled: $value") },
            { error -> println("Pipeline error: ${error.message}") },
            { println("All orders processed!") }
        )
}

Crear un Flux

Todo pipeline reactivo empieza con una fuente. Reactor trae una familia completa de métodos de fábrica que convierten lo que ya tienes — valores literales, colecciones, Streams de Java, rangos de números, un reloj, callbacks imperativos, incluso "nada en absoluto" — en un Flux. Este capítulo recorre cada fábrica con un ejemplo ejecutable, la salida exacta que produce y los detalles que muerden en código real.

just

Flux.just("A", "B", "C")
    .subscribe { value -> println("got " + value) }

Flux.just es la puerta de entrada de Reactor: le entregas valores que ya tienes a la mano y los envuelve en un Flux que emite cada uno, en orden, y luego completa. No hay cómputo, ni espera, ni I/O — solo un guion fijo que se reproduce para cada suscriptor.

Úsalo cuando los datos se conocen en el punto de llamada: sembrar datos en pruebas, proveer respaldos constantes o construir la cabeza de un pipeline pequeño sobre el cual encadenar operadores.

import reactor.core.publisher.Flux

fun main() {
    // A fixed, already-known set of order lifecycle states
    Flux.just("PENDING", "PAID", "SHIPPED", "DELIVERED")
        .map { status -> "Order #4711 is now $status" }
        .subscribe { println(it) }
}

Tenemos cuatro estados del ciclo de vida de una orden, conocidos antes de que el stream siquiera exista. just los emite uno por uno en el orden declarado; map decora cada estado con el número de orden; al final nos suscribimos e imprimimos cada línea. Después de DELIVERED, el Flux emite onComplete.

Order #4711 is now PENDING
Order #4711 is now PAID
Order #4711 is now SHIPPED
Order #4711 is now DELIVERED

fromIterable / fromArray / fromStream / from

val letters = listOf("A", "B", "C", "D")
Flux.fromIterable(letters)
    .subscribe(
        { value -> println("next: " + value) },
        { error -> println("error") },
        { println("done") }
    )

La mayoría de los streams no nacen de literales — nacen de datos que ya tienes: una List cargada de un repositorio, un arreglo, un Stream de Java o un Publisher producido por otra biblioteca reactiva. Esta familia de fábricas convierte cada uno de ellos en un Flux: fromIterable para colecciones, fromArray para arreglos, fromStream para Streams de Java y from para cualquier Publisher.

Las cuatro se comportan igual aguas abajo: cada elemento se vuelve un onNext, en orden de iteración, seguido de un único onComplete.

import reactor.core.publisher.Flux
import java.util.stream.Stream

fun main() {
    // 1. fromIterable — any List, Set or other Iterable
    val cart = listOf("Keyboard", "Mouse", "Monitor")
    Flux.fromIterable(cart)
        .subscribe { println("In cart: $it") }

    // 2. fromArray — a plain array of prices
    val prices = arrayOf(49.99, 19.99, 229.0)
    Flux.fromArray(prices)
        .subscribe { println("Price: $it") }

    // 3. fromStream — a Java Stream (consumable only ONCE!)
    Flux.fromStream(Stream.of("card", "paypal", "transfer"))
        .subscribe { println("Payment method: $it") }

    // 4. from — adapt any existing Publisher (here another Flux)
    val publisher = Flux.just("EUR", "USD")
    Flux.from(publisher)
        .subscribe { println("Currency: $it") }
}

Tenemos un carrito de compras como List, precios de productos como Array, métodos de pago como Stream de Java y un feed de monedas que ya es un Publisher. Cada fábrica recorre su fuente y emite cada elemento como una señal propia — la lambda del suscriptor imprime cada línea al llegar, y cada mini-stream completa justo después de su último elemento.

In cart: Keyboard
In cart: Mouse
In cart: Monitor
Price: 49.99
Price: 19.99
Price: 229.0
Payment method: card
Payment method: paypal
Payment method: transfer
Currency: EUR
Currency: USD

range

Flux.range(1, 5)
    .subscribe { println("got " + it) }

range es el for-loop reactivo: dale un entero inicial y una cantidad, y emite esa cantidad de enteros consecutivos, y luego completa. No hay ninguna colección detrás — los números se generan bajo demanda, a medida que el suscriptor los solicita.

Brilla donde necesitas N de algo: números de página por consultar, contadores de reintentos, o índices de asientos y lotes para construir etiquetas.

import reactor.core.publisher.Flux

fun main() {
    // Generate the 5 seat numbers of row A: no list needed, just start + count
    Flux.range(1, 5)
        .map { n -> "A$n" }
        .subscribe { seat -> println("Seat $seat reserved") }
}

Pedimos 5 enteros comenzando en 1. map convierte cada número pelado en una etiqueta de asiento de la fila A, y el suscriptor imprime una reserva por asiento. Después del asiento A5 el stream completa por sí solo.

Seat A1 reserved
Seat A2 reserved
Seat A3 reserved
Seat A4 reserved
Seat A5 reserved

interval

val ticks: Flux<Long> = Flux.interval(Duration.ofSeconds(1))
ticks.take(5)
    .subscribe { println(it) }

interval es un reloj convertido en stream: tras cada periodo emite el siguiente número de una secuencia infinita 0, 1, 2, 3… Es la forma estándar de hacer polling a una API, refrescar un dashboard o impulsar un heartbeat.

Dos cosas lo distinguen de todas las fábricas anteriores: es asíncrono (los ticks se producen en el Scheduler parallel, no en tu hilo) y nunca completa por sí solo — casi siempre lo combinas con take(n) o takeUntil.

import reactor.core.publisher.Flux
import java.time.Duration

fun main() {
    // Poll a service heartbeat every 200ms, but only 3 times
    Flux.interval(Duration.ofMillis(200))
        .take(3)
        .map { tick -> "heartbeat #${tick + 1}" }
        .doOnNext { println(it) }
        .blockLast() // interval runs on another thread: wait for it here
}

Arrancamos un reloj de 200 ms. take(3) deja pasar solo los tres primeros ticks (0, 1, 2) y luego cancela el temporizador; map renombra cada tick a un número de heartbeat legible. Como los ticks llegan en un hilo del temporizador, main terminaría antes del primer latido — blockLast() detiene el hilo principal hasta que el stream finaliza.

heartbeat #1
heartbeat #2
heartbeat #3

empty / error / never

Flux.empty<Int>()
    .concatWith(Flux.error(RuntimeException("boom")))
    .subscribe(
        { println("next: " + it) },
        { println("err: " + it.message) },
        { println("done") }
    )

No todo stream tiene datos. empty() completa de inmediato sin emitir nada; error(e) falla de inmediato con tu excepción; never() no hace absolutamente nada — ni valor, ni completado, ni error. Suenan triviales, pero son el vocabulario para "sin resultado", "falló" y "sigue pendiente" en código reactivo.

Los devolverás desde funciones todo el tiempo: empty() como el "not found" del mundo reactivo, error() para señalar fallos sin lanzar excepciones, y never() en pruebas para verificar el manejo de timeouts y cancelaciones.

import reactor.core.publisher.Flux
import reactor.core.publisher.Mono

fun main() {
    fun lookup(id: Int): Mono<String> = when {
        id < 0  -> Mono.error(IllegalArgumentException("bad id")) // failed lookup
        id == 0 -> Mono.empty()                                   // no result
        else    -> Mono.just("user-$id")
    }

    // error -> fallback, empty -> default; never() would hang, so we never subscribe to it
    Flux.just(-1, 0, 2)
        .flatMap { id ->
            lookup(id)
                .onErrorResume { Mono.just("<error>") }
                .defaultIfEmpty("<none>")
        }
        .subscribe { println(it) }
}

lookup devuelve un publisher degenerado distinto según el id: un error para -1, un vacío para 0 y un valor real para 2. flatMap se suscribe a cada Mono interno: onErrorResume convierte el fallo en "<error>", defaultIfEmpty llena el vacío con "<none>", y user-2 pasa intacto.

<error>
<none>
user-2

create / push

val flux: Flux<Int> = Flux.create { sink ->
    sink.next(10)
    sink.next(20)
    sink.next(30)
    sink.complete()
}
flux.subscribe(::println)

Tarde o temprano tendrás que conectar una API no reactiva — un listener de WebSocket, un consumidor JMS, un manejador de eventos de GUI — a un Flux. create() te entrega un FluxSink: un control remoto imperativo donde sink.next(v) emite un valor y sink.complete() / sink.error(e) terminan el stream.

push() es la misma API con una restricción: solo un hilo puede llamar al sink. create() serializa emisiones provenientes de cualquier cantidad de hilos; push() se salta esa protección a cambio de un poco menos de sobrecarga.

import reactor.core.publisher.Flux

// A tiny callback-based API — think of a message-broker or WebSocket client
class PriceFeed {
    private var listener: ((String) -> Unit)? = null
    fun onPrice(callback: (String) -> Unit) { listener = callback }
    fun start() {
        listOf("BTC = 67,100", "BTC = 67,240", "BTC = 67,180")
            .forEach { listener?.invoke(it) }
    }
}

fun main() {
    val feed = PriceFeed()

    // create() hands you a FluxSink: every callback becomes an emission
    val prices = Flux.create<String> { sink ->
        feed.onPrice { price -> sink.next(price) } // callback -> onNext
        feed.start()
        sink.complete()                            // don't forget the terminal!
    }

    prices.subscribe { println("Tick: $it") }
}

PriceFeed es una API de callbacks clásica: registras un listener y ella te llama de vuelta. Dentro de create registramos un listener que reenvía cada callback a sink.next, arrancamos el feed y señalamos complete cuando termina. El suscriptor ve los tres callbacks como un Flux de ticks perfectamente normal.

Tick: BTC = 67,100
Tick: BTC = 67,240
Tick: BTC = 67,180

Sinks.many / asFlux

val sink = Sinks.many().multicast().onBackpressureBuffer<String>()
sink.tryEmitNext("e1")                            // before anyone subscribes
sink.asFlux().subscribe { println("A: " + it) }   // subscriber A
sink.tryEmitNext("e2")
sink.asFlux().subscribe { println("B: " + it) }   // subscriber B (late)
sink.tryEmitNext("e3")

create() envuelve el registro de un callback, pero muchas veces solo quieres un handle independiente al que tu servicio pueda emitir desde cualquier parte: un handler REST empuja una notificación, un listener de Kafka empuja un evento, y quien esté suscrito lo ve. Eso es Sinks — un sink thread-safe que construyes directamente, alimentas con tryEmitNext y expones como stream con asFlux(). Es el reemplazo moderno de las clases Processor deprecadas (EmitterProcessor, ReplayProcessor y compañía).

El builder define el contrato de entrega de forma explícita. Sinks.many().unicast() permite exactamente un suscriptor y almacena todo hasta que llega. .multicast() permite muchos suscriptores en vivo: los valores emitidos antes del primero se almacenan, pero los suscriptores tardíos solo ven lo emitido después de unirse. .replay().all() / .limit(n) / .latest() re-entregan el historial a cada rezagado.

import reactor.core.publisher.Sinks

fun main() {
    // A multicast sink: one imperative producer, many live subscribers
    val sink = Sinks.many().multicast().onBackpressureBuffer<String>()
    val notifications = sink.asFlux()

    sink.tryEmitNext("order-1 placed")   // no subscriber yet: buffered (warm-up)

    notifications.subscribe { println("A got: $it") }
    sink.tryEmitNext("order-2 placed")   // A is live -> delivered immediately

    notifications.subscribe { println("B got: $it") }
    sink.tryEmitNext("order-3 placed")   // both A and B are live
    sink.tryEmitComplete()

    // replay().limit(1) keeps history for late subscribers instead
    val lastPrice = Sinks.many().replay().limit<Int>(1)
    lastPrice.tryEmitNext(101)
    lastPrice.tryEmitNext(105)
    lastPrice.asFlux().subscribe { println("late subscriber sees: $it") }
}

order-1 se emite cuando nadie escucha: multicast().onBackpressureBuffer() lo aparca en un buffer de calentamiento, así que el suscriptor A lo recibe en cuanto se suscribe. order-2 encuentra a A en vivo y se entrega de inmediato. B llega tarde — se perdió las dos primeras señales — así que cuando se emite order-3 tanto A como B lo imprimen, pero B nunca ve order-1 ni order-2. El sink replay().limit(1) guarda el último valor como historial: el suscriptor tardío igual recibe 105.

A got: order-1 placed
A got: order-2 placed
A got: order-3 placed
B got: order-3 placed
late subscriber sees: 105

generate

val flux: Flux<Int> = Flux.generate({ 0 }) { state, sink ->
    sink.next(state)
    if (state == 4) sink.complete()
    state + 1
}
flux.subscribe { print("$it ") }

generate es el hermano disciplinado de create: en lugar de un sink abierto al que puedes martillar a gusto, llama a tu función una vez por cada solicitud del consumidor, y puedes emitir como máximo un valor por llamada. Eso lo hace perfectamente seguro ante backpressure — físicamente no puede rebasar al suscriptor.

Su superpoder es el parámetro de estado: cada invocación recibe el estado devuelto por la anterior, que es exactamente lo que necesitas para contadores, cursores de paginación y secuencias matemáticas.

import reactor.core.publisher.Flux

fun main() {
    // Stateful generate: a lazy Fibonacci sequence, one value per invocation
    Flux.generate<Long, Pair<Long, Long>>({ 0L to 1L }) { state, sink ->
        val (a, b) = state
        sink.next(a)        // at most ONE next() per call
        b to (a + b)        // return the state for the next round
    }
        .take(8)            // generate is infinite: demand decides when to stop
        .subscribe { println(it) }
}

El estado arranca en (0, 1). En cada solicitud la lambda desestructura el par, emite el primer componente y devuelve (b, a + b) como estado para la siguiente ronda. La secuencia es infinita a propósito: take(8) solicita exactamente ocho valores y luego cancela, así que solo se calculan los primeros ocho números de Fibonacci.

0
1
1
2
3
5
8
13

defer

var counter = 0
val flux = Flux.defer { Flux.just("v" + (++counter)) }
flux.subscribe { println(it) }   // subscriber A
flux.subscribe { println(it) }   // subscriber B

defer vuelve perezosas las fuentes: guarda un proveedor y lo ejecuta una vez por suscriptor, en el momento de la suscripción. El Publisher que devuelva el proveedor es el que consume ese suscriptor en particular — así cada suscripción puede ver valores recién calculados.

Es la solución estándar a la trampa de captura ansiosa de just: todo lo que se calcula dentro de la lambda de defer ocurre por suscripción, no una sola vez al ensamblar.

import reactor.core.publisher.Flux
import java.time.LocalTime

fun main() {
    // Eager: the timestamp is captured ONCE, when the Flux is assembled
    val eager = Flux.just(LocalTime.now().toString())

    // Lazy: the supplier runs again for EVERY subscriber
    val lazy = Flux.defer { Flux.just(LocalTime.now().toString()) }

    eager.subscribe { println("Eager 1: $it") }
    Thread.sleep(1000)
    eager.subscribe { println("Eager 2: $it") } // same value!

    lazy.subscribe { println("Lazy 1: $it") }
    Thread.sleep(1000)
    lazy.subscribe { println("Lazy 2: $it") }   // one second later!
}

Las dos suscripciones ansiosas imprimen la misma marca de tiempo: LocalTime.now() corrió exactamente una vez, cuando se ensambló Flux.just. El Flux perezoso ejecuta su proveedor en cada subscribe, así que Lazy 1 y Lazy 2 — con un segundo de diferencia — imprimen horas distintas. La misma forma de código, un modelo de evaluación completamente diferente.

Eager 1: 11:51:16.169791
Eager 2: 11:51:16.169791
Lazy 1: 11:51:17.243813
Lazy 2: 11:51:18.248844

Transforming

Los operadores de transformación remodelan los elementos que fluyen por un Flux: convierten strings crudos en objetos de dominio, despliegan un usuario en sus pedidos o pliegan un flujo de transacciones en un saldo acumulado. Son el pan de cada día de la programación reactiva: casi todo pipeline real empieza con alguno de ellos.

La familia de map se presta a confusiones, así que este es el modelo mental desde el inicio:

• map — síncrono y 1 a 1: entra un valor, sale exactamente uno.
• flatMap — despliegue asíncrono: fusiona los publishers internos a medida que emiten; el más rápido, pero sin garantía de orden.
• concatMap — asíncrono, un interno a la vez: orden estricto de la fuente, a costa de latencia.
• flatMapSequential — despliegue asíncrono con salida ordenada: se suscribe con anticipación y usa buffers para restaurar el orden de la fuente.
• switchMap — gana el último: cada valor nuevo cancela el publisher interno anterior.

map / mapNotNull

map es el operador que vas a usar con más frecuencia: toma cada valor que fluye por el Flux y le aplica una función síncrona común y corriente — entra un valor, sale exactamente uno, en el mismo orden. Piensa en una banda transportadora donde a cada caja se le pega una etiqueta al pasar. mapNotNull es el hermano pragmático: aplica la función y descarta en silencio los resultados null, fusionando un map y un filter en un solo paso.

Flux.just(1, 2, 3, 4, 5)
    .map { it * 10 }
    .subscribe(::println)

import reactor.core.publisher.Flux

data class User(val name: String, val age: Int)

fun main() {
    // 1-to-1 transform: square each number
    Flux.just(1, 2, 3, 4, 5)
        .map { it * it }
        .subscribe { println("Squared: $it") }

    // Map raw lines into domain objects
    Flux.just("Alice:30", "Bob:25", "Charlie:35")
        .map { line ->
            val (name, age) = line.split(":")
            User(name, age.toInt())
        }
        .subscribe { println("User: ${it.name}, Age: ${it.age}") }

    // mapNotNull: transform and drop nulls in one step
    Flux.just("1", "two", "3", "four", "5")
        .mapNotNull { it.toIntOrNull() }
        .subscribe { println("Parsed: $it") }
}

Corremos tres pipelines pequeños. El primero eleva cada número al cuadrado: entran del 1 al 5 y salen sus cuadrados, uno por uno y en orden. El segundo toma strings crudos "nombre:edad" y mapea cada línea a un objeto User tipado — la clásica conversión de DTO a dominio. El tercero intenta parsear cada string con toIntOrNull(): "two" y "four" producen null, así que mapNotNull los descarta y solo sobreviven 1, 3 y 5. Al final nos suscribimos e imprimimos cada resultado:

Squared: 1
Squared: 4
Squared: 9
Squared: 16
Squared: 25
User: Alice, Age: 30
User: Bob, Age: 25
User: Charlie, Age: 35
Parsed: 1
Parsed: 3
Parsed: 5

flatMap

flatMap es la forma de hacer trabajo asíncrono por elemento: mapea cada valor a un Publisher (normalmente una llamada de red o de base de datos), se suscribe a todos esos publishers internos con anticipación y fusiona sus resultados en un solo Flux a medida que llegan. Es el más rápido de los operadores de mapeo asíncrono precisamente porque nunca espera: el que responde primero se emite primero.

Dos extras útiles: un segundo argumento opcional limita cuántos internos corren a la vez — flatMap(mapper, 4) mantiene como máximo cuatro llamadas en vuelo — y la variante flatMapDelayError sigue fusionando los internos restantes cuando uno falla, entregando el error recién al final.

fun inner(x: String): Flux<String> = Flux.just(x + "1", x + "2")

Flux.just("A", "B")
    .flatMap { inner(it) }
    .subscribe { println(it) }

import reactor.core.publisher.Flux
import java.time.Duration

// Each user's orders arrive from a (simulated) remote service
fun fetchOrders(user: String, latencyMs: Long): Flux<String> =
    Flux.just("$user-order-1", "$user-order-2")
        .delayElements(Duration.ofMillis(latencyMs))

fun main() {
    Flux.just("ana", "luis", "sofia")
        .flatMap { user ->
            val latency = when (user) {
                "ana" -> 120L   // slowest service
                "luis" -> 80L
                else -> 50L     // fastest service
            }
            fetchOrders(user, latency)
        }
        .doOnNext { println(it) }
        .blockLast()
}

Desplegamos tres usuarios hacia un servicio de pedidos simulado donde cada uno tiene una latencia distinta: sofia responde cada 50 ms, luis cada 80 ms y ana cada 120 ms. flatMap se suscribe a los tres Flux internos de inmediato, así que la salida fusionada queda ordenada por tiempo de llegada, no por el orden de la fuente — el primer pedido de sofia le gana a todos aunque ella fue la última usuaria emitida. Bloqueamos en el último elemento para que main no termine antes de tiempo. Una ejecución real imprimió:

sofia-order-1
luis-order-1
sofia-order-2
ana-order-1
luis-order-2
ana-order-2

flatMapSequential

flatMapSequential responde a la pregunta "¿puedo tener la velocidad de flatMap con el orden de concatMap?". Se suscribe a cada publisher interno con anticipación, así que el trabajo asíncrono corre en paralelo — pero guarda en buffer los resultados de los internos que terminan antes y lo reproduce todo estrictamente en el orden de la fuente. Lo mejor de ambos mundos, pagado con memoria.

fun inner(n: Int): Flux<String> =
    Flux.just(n.toString() + "a", n.toString() + "b")

Flux.just(1, 2, 3)
    .flatMapSequential { inner(it) }
    .subscribe { println(it) }

import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.time.Duration

fun main() {
    val start = System.currentTimeMillis()
    Flux.just("slow", "fast", "medium")
        .flatMapSequential { item ->
            val delay = when (item) {
                "slow" -> 300L
                "fast" -> 50L
                else -> 150L
            }
            Mono.just(item.uppercase()).delayElement(Duration.ofMillis(delay))
        }
        .doOnNext { println("$it after ${System.currentTimeMillis() - start} ms") }
        .blockLast()
    println("total: ${System.currentTimeMillis() - start} ms")
}

Tres tareas con latencias muy distintas: slow (300 ms), fast (50 ms) y medium (150 ms). Los tres Monos con retardo arrancan al mismo tiempo. FAST está listo a los ~50 ms, pero flatMapSequential lo retiene hasta que SLOW — el primer elemento de la fuente — haya emitido. Por eso las tres líneas se imprimen juntas alrededor de los 300 ms, y el pipeline completo tarda unos 300 ms en vez de 500 ms (la suma), demostrando que el trabajo sí corrió en paralelo:

SLOW after 369 ms
FAST after 370 ms
MEDIUM after 370 ms
total: 370 ms

concatMap

concatMap es el secuenciador estricto: mapea cada elemento a un publisher interno pero se suscribe a ellos de a uno, vaciando por completo cada interno antes de tocar el siguiente. Sin entrelazado y con orden de la fuente garantizado — ideal cuando cada paso debe completarse antes de que empiece el siguiente, como escrituras ordenadas en base de datos o llamadas a APIs que dependen una de otra.

fun inner(name: String): Flux<String> =
    Flux.just(name + "1", name + "2")

Flux.just("a", "b")
    .concatMap { inner(it) }
    .subscribe { println(it) }

import reactor.core.publisher.Flux

fun main() {
    // Each order goes through two warehouse steps, strictly one order at a time
    Flux.just("burger", "fries")
        .concatMap { item ->
            Flux.just("$item: picked", "$item: packed")
        }
        .subscribe { println(it) }
}

Tenemos un flujo con dos pedidos de cocina. concatMap mapea cada uno a un Flux interno con sus dos pasos de bodega: picked y luego packed. Como concatMap vacía cada interno por completo antes de suscribirse al siguiente, la hamburguesa se recoge y se empaca antes de siquiera empezar con las papas — las emisiones nunca pueden entrelazarse:

burger: picked
burger: packed
fries: picked
fries: packed

La garantía de orden tiene un precio: la latencia se suma. Aquí cada llamada de enriquecimiento tarda 100 ms y concatMap las ejecuta estrictamente una tras otra:

import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.time.Duration

fun enrich(sku: String): Mono<String> =
    Mono.just("$sku [enriched]").delayElement(Duration.ofMillis(100))

fun main() {
    val start = System.currentTimeMillis()
    Flux.just("A", "B", "C")
        .concatMap { enrich(it) }
        .doOnNext { println(it) }
        .blockLast()
    println("total: ${System.currentTimeMillis() - start} ms")
}

A [enriched]
B [enriched]
C [enriched]
total: 470 ms

flatMapIterable

A veces cada elemento ya trae a sus hijos en memoria — un pedido con sus líneas, un post con sus etiquetas. flatMapIterable toma cada valor, le pide un Iterable y aplana todos sus elementos dentro del flujo de forma síncrona y en orden. Sin publishers internos, sin concurrencia, sin sorpresas. concatMapIterable es un alias con la nomenclatura de concat — para iterables en memoria el comportamiento es el mismo.

data class Order(val items: List<String>)

Flux.just(Order(listOf("a1", "a2")), Order(listOf("b1", "b2")))
    .flatMapIterable { it.items }
    .subscribe { println(it) }

import reactor.core.publisher.Flux

data class Order(val id: Int, val items: List<String>)

fun main() {
    Flux.just(
        Order(1, listOf("book", "pen")),
        Order(2, listOf("lamp"))
    )
        .flatMapIterable { it.items }
        .subscribe { println("item: $it") }

    // Works with any Iterable: split sentences into words
    Flux.just("hello world", "reactive kotlin")
        .flatMapIterable { it.split(" ") }
        .map { it.uppercase() }
        .subscribe { println(it) }
}

El primer pipeline es un flujo de dos pedidos, cada uno con sus líneas en un List común. flatMapIterable extrae los items de cada pedido y los emite uno por uno: los dos items del pedido 1 salen antes que los del pedido 2, siempre. El segundo pipeline muestra que funciona con cualquier Iterable — cada oración se divide en palabras, que luego fluyen por el resto de la cadena como elementos individuales:

item: book
item: pen
item: lamp
HELLO
WORLD
REACTIVE
KOTLIN

switchMap

switchMap modela "solo importa lo último": cada nuevo valor de la fuente cancela el publisher interno lanzado por el anterior y cambia a un interno nuevo. Es EL operador para búsqueda mientras escribes, autocompletado y filtrado en vivo — en cualquier lugar donde los resultados de una consulta vieja sean peores que no tener resultados.

Su primo estático switchOnNext hace el mismo cambio sobre un Flux de publishers que ya tienes; y si lo que necesitas es bifurcar todo el pipeline según el primer elemento, eso es switchOnFirst, que viene a continuación.

fun inner(n: Int): Flux<String> =
    Flux.just(n.toString() + "a", n.toString() + "b").delayElements(Duration.ofMillis(50))

Flux.just(1, 2, 3)
    .switchMap { inner(it) }
    .subscribe { println(it) }

import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.time.Duration

fun search(query: String): Mono<String> =
    Mono.just("results for '$query'").delayElement(Duration.ofMillis(150))

fun main() {
    // Three keystrokes: two in quick succession, then a pause before the last
    val keystrokes = Flux.concat(
        Mono.just("re"),
        Mono.just("rea").delayElement(Duration.ofMillis(80)),
        Mono.just("react").delayElement(Duration.ofMillis(300))
    )

    keystrokes
        .switchMap { query -> search(query) }
        .doOnNext { println(it) }
        .blockLast()
}

Simulamos tres teclas: "re" de inmediato, "rea" 80 ms después, y una pausa de 300 ms antes de "react". Cada tecla dispara una búsqueda de 150 ms. La búsqueda de "re" sigue en vuelo cuando "rea" llega a los 80 ms, así que switchMap la cancela — "re" nunca produce resultados. La búsqueda de "rea" termina a los ~230 ms, antes de que "react" llegue a los ~380 ms, así que sus resultados SÍ pasan. Al final, "react" ejecuta su búsqueda sin interrupciones:

results for 'rea'
results for 'react'

switchOnFirst

Flux.just("id,name", "1,Ana", "2,Luis")
    .switchOnFirst { first, rows ->
        if (first.get()?.startsWith("id") == true)
            rows.skip(1).map { "row: $it" }
        else
            rows.map { "raw: $it" }
    }
    .subscribe { println(it) }

switchOnFirst te deja mirar la primera señal de una secuencia y decidir, ahí mismo, cómo debe ser el resto del pipeline. Tu transformador recibe dos cosas: la primera Signal (un valor, una finalización o un error) y la secuencia completa — primer elemento incluido — y devuelve el Publisher con el que continuar. El caso de uso clásico: un flujo CSV donde la fila de encabezado define cómo parsear cada fila siguiente.

import reactor.core.publisher.Flux

fun main() {
    // A CSV file as a stream: first the header row, then the data rows
    val csv = Flux.just(
        "id,name,city",
        "1,Ana,Bogota",
        "2,Luis,Lima"
    )

    csv.switchOnFirst { first, rows ->
        // `first` is the header signal; `rows` is the WHOLE sequence (header included)
        val columns = first.get()?.split(",") ?: emptyList()
        rows.skip(1).map { line ->
            columns.zip(line.split(","))
                .joinToString(", ") { (col, value) -> "$col=$value" }
        }
    }.subscribe { println(it) }
}

El primer elemento del flujo es el encabezado "id,name,city". switchOnFirst le entrega a nuestra lambda esa primera señal más la secuencia completa de filas. Dividimos el encabezado en nombres de columna y construimos el pipeline bifurcado: skip(1) descarta el propio encabezado, y map combina cada fila de datos con los nombres de columna para producir registros etiquetados. La decisión ocurre una sola vez, cuando llega el primer elemento — después de eso las filas de datos fluyen directo por el nuevo pipeline:

id=1, name=Ana, city=Bogota
id=2, name=Luis, city=Lima

handle

handle es la transformación navaja suiza: por cada elemento recibes un SynchronousSink y decides qué pasa — llamas a sink.next(v) para emitir un valor (posiblemente transformado), no haces nada para descartarlo en silencio, o llamas a sink.complete() / sink.error() para terminar el flujo ahí mismo. Es map, filter y terminación anticipada fusionados en un solo operador.

Flux.just(1, 2, 3, 4, 5)
    .handle<Int> { v, sink ->
        if (v % 2 != 0) sink.next(v * 10)
    }
    .subscribe { println("got " + it) }

import reactor.core.publisher.Flux

fun main() {
    // handle: map + filter + early completion in a single operator
    Flux.range(1, 10)
        .handle<String> { value, sink ->
            when {
                value > 7 -> sink.complete()                 // stop the stream early
                value % 2 == 0 -> sink.next("Even: $value")  // emit transformed
                // odd values below 7: no sink.next -> silently dropped
            }
        }
        .subscribe { println(it) }
}

Tenemos los números del 1 al 10. Para cada uno el sink decide: los valores mayores a 7 disparan sink.complete(), terminando el flujo antes de que 8, 9 y 10 lleguen a emitirse; los pares se transforman y se emiten; los impares menores a 7 simplemente atraviesan el when sin un sink.next, así que se descartan. El suscriptor solo ve los pares transformados:

Even: 2
Even: 4
Even: 6

handle brilla en pipelines de "parsear o descartar", donde una transformación fallida simplemente significa "salta este":

import reactor.core.publisher.Flux

fun main() {
    Flux.just("12.50", "abc", "8.00", "xyz")
        .handle<Double> { raw, sink ->
            val price = raw.toDoubleOrNull()
            if (price != null) sink.next(price)   // parse or silently drop
        }
        .subscribe { println("price: $$it") }
}

price: $12.5
price: $8.0

cast / ofType

Flux.just(1, "a", 2, "b", 3)
    .ofType(Integer::class.java)
    .subscribe { value -> println(value) }

Cuando un Flux transporta tipos mezclados — un bus de eventos polimórfico, un Flux<Any> heredado — ofType(clazz) conserva solo los elementos que son instancias de la clase dada y los castea por ti, descartando en silencio todo lo demás. cast(clazz) es el hermano sin verificación: deja pasar todos los elementos y solo reinterpreta el tipo, lanzando un ClassCastException ante la primera discrepancia.

import reactor.core.publisher.Flux

open class Animal(val name: String)
class Dog(name: String, val breed: String) : Animal(name)
class Cat(name: String, val indoor: Boolean) : Animal(name)

fun main() {
    val animals: Flux<Animal> = Flux.just(
        Dog("Rex", "Shepherd"),
        Cat("Whiskers", true),
        Dog("Buddy", "Labrador"),
        Cat("Luna", false)
    )

    // ofType: filter and cast in one safe step
    animals.ofType(Dog::class.java)
        .subscribe { println("Dog: ${it.name}, Breed: ${it.breed}") }

    animals.ofType(Cat::class.java)
        .subscribe { println("Cat: ${it.name}, Indoor: ${it.indoor}") }
}

Un flujo de refugio mezcla Dogs y Cats detrás del tipo común Animal. La primera suscripción lo reduce a Dogs — las dos Cats se descartan, y el código aguas abajo puede leer .breed sin ningún cast manual. Como animals es un Flux frío, la segunda suscripción reproduce los cuatro animales de nuevo y lo reduce a Cats. Cada suscriptor ve solo su propia especie, en el orden de la fuente:

Dog: Rex, Breed: Shepherd
Dog: Buddy, Breed: Labrador
Cat: Whiskers, Indoor: true
Cat: Luna, Indoor: false

El mismo truco ordena flujos de eventos sin tipar — extrae solo los comandos e ignora el ruido numérico:

import reactor.core.publisher.Flux

fun main() {
    val events: Flux<Any> = Flux.just(42, "deploy", 3.14, "rollback", 7)

    // Keep only the String commands — everything else is silently dropped
    events.ofType(String::class.java)
        .map { it.uppercase() }
        .subscribe { println("command: $it") }
}

command: DEPLOY
command: ROLLBACK

scan

scan es un reduce con comentarista en vivo: pliega el flujo con una función acumuladora pero emite cada resultado intermedio, no solo el final. Eso lo hace perfecto para totales acumulados, estadísticas en vivo o flujos de estado estilo reducer. Con semilla, la propia semilla se emite primero; la variante scanWith recibe un supplier, así cada suscriptor obtiene una semilla nueva creada de forma perezosa.

Flux.just(1, 2, 3)
    .scan(0) { acc, x -> acc + x }
    .subscribe { println(it) }

import reactor.core.publisher.Flux

fun main() {
    // A stream of account transactions: deposits and withdrawals
    Flux.just(100, -30, -20, 50)
        .scan(0) { balance, tx -> balance + tx }
        .subscribe { println("balance: $it") }
}

Plegamos un flujo de cuatro transacciones bancarias sobre la semilla 0. scan emite la semilla de inmediato — ese es el primer "balance: 0" — y luego un saldo actualizado por transacción: +100 da 100, −30 da 70, −20 da 50, y +50 lo devuelve a 100. El suscriptor ve el saldo evolucionar en vez de enterarse solo de la cifra final:

balance: 0
balance: 100
balance: 70
balance: 50
balance: 100

Sin semilla, el propio primer elemento se convierte en el acumulador inicial y pasa tal cual:

import reactor.core.publisher.Flux

fun main() {
    // Without a seed, the first element passes through as-is
    Flux.just(3, 1, 4, 1, 5, 9, 2, 6)
        .scan { a, b -> maxOf(a, b) }
        .subscribe { println("max so far: $it") }
}