En la lección anterior consumiste streams que ya existían: createReadStream producía trozos y createWriteStream los recibía. Entendiste la contrapresión, viste que pipe() la gestiona por ti y descubriste su defecto: no propaga errores ni limpia lo que deja atrás.

Ahora toca la otra mitad. Vas a escribir tus propios streams —el eslabón intermedio que convierte una cosa en otra— y a montar la tubería completa de proceso de Escena Viva: leer datos/ventas.csv, convertir cada línea en un objeto venta, filtrar por sala, agregar por sesión y escribir el resultado, todo con memoria constante, sea el fichero de dos kilobytes o de ochocientos megabytes. Y sustituirás pipe() por stream.pipeline, la herramienta de verdad: propaga errores, destruye toda la cadena cuando algo falla y tiene versión con promesas. Al final verás la forma más legible de escribir transformaciones, que no es una clase sino un generador asíncrono.

1. objectMode: streams que transportan objetos
2. El stream Transform: _transform y _flush
3. Un Transform que parsea líneas CSV
4. Readable y Writable a medida
5. stream.pipeline frente a pipe()
6. La tubería de proceso de Escena Viva
7. Compresión con zlib en la tubería
8. Readable.from() y los generadores asíncronos
9. Errores y cancelación con AbortSignal

Por defecto, un stream transporta bytes: Buffer o cadenas. Eso es lo correcto para ficheros y sockets, pero inútil cuando lo que fluye por la tubería son ventas, eventos o registros ya interpretados. objectMode: true cambia esa naturaleza:

El modo se puede declarar por lado: new Transform({ readableObjectMode: true, writableObjectMode: false }) describe exactamente lo que hace un parser —recibe texto y emite objetos—, mientras que objectMode: true a secas activa los dos lados. Y el null merece un aviso: en un stream, push(null) significa "se acabó", así que no puedes enviar null como dato, ni siquiera en modo objeto; para representar la ausencia de algo, usa undefined o un objeto marcador.

Un Transform es un stream que lee por un lado, hace algo y escribe por el otro. Se implementa con dos métodos:

La mecánica es siempre la misma: _transform recibe un trozo, hace su trabajo, emite cero o más resultados con this.push(...) y avisa de que ha terminado llamando a callback(); _flush se ejecuta una única vez al final, cuando ya no va a entrar nada más, y es la última oportunidad de emitir.

Tres reglas que no admiten excepción:

• Hay que llamar a callback() exactamente una vez por cada _transform. Si no lo llamas, la tubería se cuelga para siempre sin ningún mensaje; si lo llamas dos veces, Node lanza ERR_MULTIPLE_CALLBACK.
• Los errores se señalan con callback(error), no con throw: una excepción lanzada en un contexto asíncrono no la captura nadie.
• _flush es donde vive el estado acumulado. Todo lo que no puedas emitir hasta ver el final —la última línea sin salto, el total de un agregado, el cierre de un JSON— sale ahí. Eso es lo que distingue un Transform de un simple map: puede acumular entre trozos.

Aquí está el problema real: los trozos del fichero no coinciden con las líneas. Un Transform con estado resuelve troceado y conversión en un solo paso.

Fíjate en el papel de #resto: es el problema que readline resolvía por nosotros en la lección anterior, ahora resuelto a mano porque necesitamos emitir objetos, no líneas. Y en _flush se procesa ese resto pendiente: sin esa línea, la última venta se perdería en silencio siempre que el fichero no termine en salto. Es un error clásico y difícil de detectar, porque falla en un registro de un millón.

Un Readable produce datos. Se implementa con _read(), que Node llama cuando quiere más material, y dentro se emite con this.push(valor); cuando ya no hay nada que producir, this.push(null) señala el final del stream. En la práctica casi nunca hace falta escribir uno: Readable.from() (apartado 8) cubre casi todos los casos con una línea.

Un Writable consume datos. Implementa _write(trozo, codificacion, callback), y ese callback es la pieza clave: hasta que no lo llamas, el stream considera que sigues ocupado. Ahí es donde nace la contrapresión que sentiste en la lección anterior.

Si el trabajo de _write fuera asíncrono —guardar en base de datos, llamar a una API—, bastaría con llamar al callback al terminar: la tubería entera se frena sola mientras tanto. Existe además _writev para procesar varios elementos de golpe, útil al insertar por lotes.

Ya viste el problema de pipe(). stream.pipeline lo arregla, y su versión con promesas —const { pipeline } = require('node:stream/promises')— es la que usaremos siempre.

La diferencia práctica es enorme:

Regla del curso: nunca pipe() en código de producción. pipe sirve para explicar el concepto; pipeline es lo que se escribe.

Con las piezas anteriores montamos el proceso completo: del CSV crudo al informe en JSON, filtrando por sala y sin cargar nunca el fichero entero.

Las 1811 entradas cuadran con la semilla, que es la comprobación que buscábamos. Y observa la forma de la tubería: cuatro etapas, cada una con una sola responsabilidad, encadenadas en un pipeline que se lee de arriba abajo; cambiar el filtro, añadir una validación o escribir a otro destino es tocar una línea. Un detalle de diseño: el AcumuladorPorSesion va al final porque una agregación no puede emitir nada hasta ver el último elemento; si necesitaras continuar la tubería después de agregar, sería un Transform que emite su resultado en _flush.

Archivar el histórico es intercalar una etapa más: zlib.createGzip() es un Transform como los tuyos.

Aquí no se ha indicado encoding en el lector: los datos viajan como Buffer de principio a fin, que es lo correcto para comprimir —convertirlos a texto y de vuelta a bytes no aportaría nada y costaría CPU—. La descompresión es la misma tubería al revés: basta con intercalar zlib.createGunzip() entre el lector del .gz y el parser para procesar el histórico comprimido sin descomprimirlo a disco. Eso es lo que hace potente el modelo: las etapas se combinan sin que ninguna sepa nada de las demás, y el parser no tiene ni idea de que sus bytes vienen de un fichero comprimido.

Readable.from() convierte cualquier iterable —array, Map, generador— en un stream:

Es utilísimo para las pruebas del Módulo 9: alimentas una tubería con datos de mentira sin tocar el disco. Pero lo mejor viene ahora: un generador asíncrono puede usarse directamente como etapa de un pipeline, y es la forma más legible de escribir una transformación.

La comparación es reveladora:

Para transformaciones nuevas, empieza siempre por un generador asíncrono. Recurre a la clase Transform cuando necesites un objeto con estado accesible desde fuera —como nuestro AcumuladorPorSesion, cuyo resultado se lee al terminar— o cuando tengas que integrarte con una API que espera un stream concreto.

Un fallo en cualquier etapa aborta la tubería y rechaza la promesa; lo que llega al catch es el error original, con su code intacto:

Y pipeline acepta una señal de cancelación, igual que fetch:

Al abortar, pipeline destruye todos los streams de la cadena: los descriptores se liberan y la promesa se rechaza con un AbortError. Es exactamente el trabajo que en la lección anterior había que escribir a mano con on('error') cruzados y destroy(). En el Módulo 4 volverás a esto: cuando un cliente HTTP cierra la conexión a mitad de una descarga, lo correcto es abortar la tubería que generaba la respuesta en lugar de seguir trabajando para nadie.

• Olvidar callback() en _transform o _write. La tubería se cuelga en silencio, sin error ni traza. Es el fallo número uno.
• Usar throw dentro de _transform en lugar de callback(error), u olvidar _flush, con lo que todo lo acumulado —la última línea, el total, el cierre— se pierde sin avisar.
• No declarar objectMode al emitir objetos: Node protesta con ERR_INVALID_ARG_TYPE porque espera Buffer o cadena. Y push(null) no es un dato: significa fin de stream, siempre.
• Seguir usando pipe(). Si un error de la primera etapa no llega a la última, tienes descriptores filtrados y fallos invisibles.
• Consejo: una etapa, una responsabilidad. Un Transform que parsea, filtra y agrega a la vez es imposible de probar; tres etapas encadenadas se prueban por separado con Readable.from(). Y si dudas entre clase y generador, empieza por el generador: convertirlo en clase después es fácil, al revés no tanto.

Escribe un generador asíncrono validarVentas que se intercale entre el parser y el filtro, y que compruebe cada venta: codigoEntrada con el formato EV-<año>-<6 dígitos>, sesionId con la forma ses-NNN-M, precioCentimos entero positivo y canal entre web, taquilla y app. Las ventas válidas pasan; las inválidas se cuentan y se registran por stderr con su número de línea. Al final debe informar de cuántas se descartaron.

Escribe src/informes/ventas-por-canal.js con una tubería que lea datos/ventas.csv, agregue por canal (entradas, recaudación y ticket medio) y escriba informes/ventas-por-canal.json.gz comprimido, sin generar el JSON sin comprimir en disco. Pista: Readable.from([JSON.stringify(informe, null, 2)]) convierte el resultado en el origen de una segunda tubería.

Escribe una función generarConReintento(opciones) que ejecute la tubería de generarVentasPorSesion con un AbortSignal de 10 segundos y, si falla por AbortError o por un error de E/S transitorio, la reintente hasta tres veces usando reintentar y dormir del Módulo 2. No debe reintentar ante ENOENT ni ante datos corruptos: esos fallos no se arreglan repitiendo.

Solución 1. El validador como generador queda casi declarativo:

El bloque final, tras el for await, es el equivalente exacto del _flush de una clase Transform: se ejecuta al agotarse la entrada. Si necesitaras parametrizar el validador, envuélvelo en una función que devuelva el generador, como el filtro del apartado 8.

Solución 2. Dos tuberías encadenadas, la segunda con Readable.from:

El JSON nunca toca el disco sin comprimir: sale de memoria como stream, pasa por el compresor y llega al fichero. Con un informe pequeño da igual; con uno de cientos de megabytes es la diferencia entre necesitar el doble de espacio libre o no.

Solución 3. La clave es clasificar los errores antes de decidir si reintentar:

Reintentar un ENOENT es inútil: el fichero no va a aparecer solo y solo consigues retrasar el mensaje de error tres veces. Un AbortError por límite de tiempo o un EBUSY transitorio, en cambio, sí pueden resolverse en el segundo intento. Reintentar sin clasificar el error es de las formas más caras de esconder un problema.

Ya no solo consumes streams: los escribes. Sabes que objectMode convierte una tubería de bytes en una tubería de objetos, que el highWaterMark pasa a contar elementos y que readableObjectMode y writableObjectMode se declaran por separado porque un parser recibe texto y emite objetos. Dominas el Transform con _transform —llamando a callback() exactamente una vez y señalando los errores con callback(error)— y _flush, donde se emite todo lo acumulado, incluida esa última línea sin salto que se pierde en silencio cuando falta. Has construido un parser de CSV con estado, un Writable agregador y un filtro, y los has encadenado en la tubería completa de Escena Viva: CSV → objetos → filtro por sala → agregación → JSON, con las 1811 ventas cuadrando con la semilla y memoria constante en todo el recorrido. Y has sustituido pipe() por stream.pipeline de node:stream/promises, que propaga errores por toda la cadena, destruye todos los streams al fallar, se integra con try/catch y acepta AbortSignal. La regla es firme: pipe para explicar, pipeline para trabajar.

Has visto además cómo zlib.createGzip se intercala como una etapa más —y cómo una tubería puede leer directamente de un .gz sin descomprimir a disco—, cómo Readable.from() convierte cualquier iterable en un stream, y por qué un generador asíncrono es casi siempre mejor que una clase Transform: menos código, objectMode implícito, await natural y el fin de la entrada tratado sin ceremonias. Y sabes cancelar una tubería con AbortSignal sin dejar descriptores vivos.

Queda una pieza en el fondo de todo esto. Cada vez que hemos escrito { encoding: 'utf8' } estábamos pidiendo una traducción de bytes a texto, y cada vez que la hemos omitido —al comprimir, al copiar— los datos viajaban como Buffer. En Buffers y Datos Binarios miraremos por fin dentro de esa caja: qué es un Buffer y por qué vive fuera del montículo de V8, por qué allocUnsafe puede mostrarte memoria de otro, las codificaciones y sus usos, el orden de bytes, el error clásico de slice compartiendo memoria, por qué un emoji rompe un troceado ingenuo y cómo lo evita StringDecoder. Y lo aplicaremos a Escena Viva: detectar por sus números mágicos si el cartel que sube un organizador es de verdad un PNG, y codificar en base64url el código QR de una entrada.