Streams: iteradores asincronos

Un Future representa un valor que aún no ha llegado. Pero muchísimo trabajo asíncrono no produce uno, sino muchos valores repartidos en el tiempo: los mensajes que van entrando por un canal, las líneas que un cliente escribe en un socket, los latidos de un temporizador, las filas que una consulta devuelve por lotes. Para eso existe el Stream: es a Future lo que Iterator es a un valor suelto. Donde el iterador tiene next —“dame el siguiente, o None si se acabó”—, el stream tiene poll_next —“dame el siguiente cuando esté listo, o avísame que aún no hay”—. Y así como implementar Iterator te regalaba setenta adaptadores, implementar o usar un Stream te da, vía StreamExt, el mismo vocabulario —map, filter, then, collect— pero asíncrono. Aprender streams es reconocer que ya sabes iterar; solo hay que añadirle la dimensión del tiempo.

Stream: el Iterator asíncrono

Recuerda el contrato de Iterator: un tipo asociado Item y un método next que devuelve Option<Item> —Some mientras haya, None al agotarse—. Stream calca esa forma y le añade lo único que le faltaba para vivir en async: la posibilidad de decir “todavía no”:

pub trait Stream {
    type Item;
    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Self::Item>>;
}

Léelo comparándolo con next. Iterator::next devuelve Option<Item> de inmediato; Stream::poll_next devuelve Poll<Option<Item>>, y ahí está toda la diferencia. Son tres desenlaces, no dos:

• Poll::Ready(Some(v)) → hay un valor listo, v; el stream sigue vivo.
• Poll::Ready(None) → el stream terminó; no vendrán más valores.
• Poll::Pending → aún no hay valor; el runtime volverá a sondear cuando lo haya.

Ese Pending es la costura async: es donde el stream cede, igual que un future cede en .await. poll_next es maquinaria de bajo nivel —con Pin y Context, herencia del nivel de Pin—; casi nunca lo llamarás a mano, igual que rara vez llamas a poll de un future. Para eso está la capa ergonómica.

Consumir un stream: while let Some

La forma de recorrer un stream es el eco async del bucle for sobre un iterador. El método .next() —que aporta la extensión StreamExt— devuelve un future de Option<Item>; lo esperas con .await, y mientras salga Some, procesas:

use tokio_stream::StreamExt;

async fn sumar(mut s: impl Stream<Item = u32> + Unpin) -> u32 {
    let mut total = 0;
    while let Some(n) = s.next().await { // cede en cada espera
        total += n; // corre cuando hay valor; None termina el bucle
    }
    total
}

Lee la línea clave despacio: s.next().await es “pídeme el siguiente y cede mientras no llegue”. Cuando el stream produce Some(n), el cuerpo del bucle corre; cuando produce None, el while let termina, exactamente como un for se acaba al agotarse el iterador. La diferencia es que entre un valor y el siguiente puede pasar tiempo real, y en ese tiempo la tarea no ocupa el hilo: lo cede para que otras avancen.

Un detalle que el compilador te recordará: .next() exige que el stream sea Unpin, o que esté pineado. Si tienes un stream que no lo es, lo fijas en el sitio con tokio::pin! antes del bucle, igual que hacías con futures reutilizados en select!.

let flujo = crear_stream();
tokio::pin!(flujo); // ahora next() lo acepta
while let Some(x) = flujo.next().await {
    manejar(x);
}

StreamExt: el vocabulario heredado

Igual que Iterator gana decenas de métodos por defecto sobre next, un Stream gana todo un vocabulario a través del trait de extensión StreamExt. Los adaptadores son los mismos que ya conoces, con dos añadidos para lo asíncrono, y todos son perezosos: construyen un stream nuevo que no hace nada hasta que lo consumes.

use tokio_stream::StreamExt;

let resumen: Vec<String> = eventos()
    .filter(|e| e.nivel >= Nivel::Warn) // conserva solo los graves
    .map(|e| e.mensaje)                 // transforma cada elemento
    .take(100)                          // corta a los primeros 100
    .collect()                          // consume: recorre y acumula
    .await;                             // collect sobre un stream es async

Dos adaptadores no tienen gemelo en Iterator porque solo tienen sentido con el tiempo: then, que es como map pero su función es async —devuelve un future por cada elemento y lo espera—, y collect, que aquí es un consumidor asíncrono y por eso se .await. La regla mental es directa: map para transformar con una función síncrona, then cuando cada elemento dispara una espera.

// then: por cada id del stream, lanza una peticion async y espera su respuesta.
let cuerpos = ids()
    .then(|id| async move { descargar(id).await })
    .collect::<Vec<_>>()
    .await;

De dónde salen los streams y cómo acotarlos

Rara vez implementarás poll_next a mano; casi siempre envuelves algo que ya produce valores en el tiempo. Las fuentes habituales: un mpsc::Receiver convertido en stream con tokio_stream::wrappers::ReceiverStream, un temporizador periódico con IntervalStream, o un Vec vuelto stream con tokio_stream::iter. Y aquí aparece la conexión con la lección 1: un stream de futures puede procesarse con concurrencia acotada, algo que join_all no te daba.

use futures::stream::StreamExt;

// Procesa hasta 8 descargas A LA VEZ, no todas de golpe ni de una en una.
let resultados: Vec<_> = futures::stream::iter(urls)
    .map(|u| async move { descargar(u).await })
    .buffer_unordered(8) // techo de concurrencia: 8 en vuelo
    .collect()
    .await;

buffer_unordered(n) mantiene hasta n futures corriendo simultáneamente y emite cada resultado en cuanto está —sin respetar el orden—; buffered(n) hace lo mismo pero conserva el orden de entrada. Es el punto medio exacto entre la latencia de procesar uno a uno y el riesgo de join_all de disparar diez mil peticiones a la vez. Además, como el stream es pull —el consumidor tira con poll_next—, obtienes contrapresión natural: nadie produce más rápido de lo que consumes.

flowchart LR
N[Consumidor hace next await] --> PN[poll_next del Stream]
PN --> R{Estado}
R -->|Ready Some v| V[Entrega v corre el cuerpo del while]
R -->|Pending| Y[Cede al runtime espera el wake]
R -->|Ready None| F[El stream termino sale del bucle]
Y --> PN
V --> N
style V fill:#a6e3a1,color:#11111b
style Y fill:#fab387,color:#11111b
style F fill:#89b4fa,color:#11111b