FITXER_VENDES fa des de la lliçó anterior que està declarat apuntant a dades/vendes.csv, un fitxer que encara no existeix. Ha arribat el seu moment, i amb ell el problema que cap de les eines vistes fins ara no pot resoldre.

El catàleg d'Escena Viva ocupa dos kilobytes: readFile el llegeix sencer sense despentinar-se. L'històric de vendes d'una temporada és una altra cosa. Tres sales, dues-centes sessions l'any, milers d'entrades per sessió, cadascuna amb el seu codi, el seu canal de compra i la seva marca de temps: centenars de megabytes. I readFile no té manera de llegir això sense ficar-ho sencer a la memòria.

Els streams són la resposta de Node a aquest problema, i són molt més que una optimització: són el model amb què Node pensa l'entrada i la sortida. Les peticions i respostes HTTP del Mòdul 4 són streams; la compressió, la criptografia i les connexions de xarxa són streams. I —això et sonarà— els streams són EventEmitter, així que tot el de la lliçó 02-05 s'aplica aquí directament.

1. El problema: mesurar el que costa readFile
2. Què és un stream i els seus quatre tipus
3. Els streams són EventEmitter
4. Mode fluid i mode pausat
5. createReadStream, createWriteStream i highWaterMark
6. Contrapressió: el que torna write()
7. pipe() i el seu punt feble
8. El fitxer dades/vendes.csv d'Escena Viva
9. Llegir línia a línia amb readline
10. for await...of sobre un stream

No cal creure's això de paraula. Anem a mesurar-ho.

Amb un vendes.csv de 800 MB, la sortida és demolidora:

I si el fitxer fos una mica més gran, readFile ni tan sols arribaria a imprimir: fallaria amb ERR_STRING_TOO_LONG —una cadena de V8 té un límit d'uns 512 MB— o amb JavaScript heap out of memory, que mata el procés sencer. La diferència és de dos ordres de magnitud, i l'explicació és simple: readFile necessita tenir el fitxer complet a la memòria alhora, mentre que l'stream el processa en trossos de 64 KB i en descarta cadascun tan bon punt l'ha fet servir. El consum de l'stream no depèn de la mida del fitxer: aquesta és la propietat clau, memòria constant.

Un stream és una seqüència de dades que es processa per parts, a mesura que estan disponibles, en comptes d'esperar a tenir-les totes. L'analogia de l'aigua és bona: readFile és omplir la banyera sencera abans de tocar l'aigua; un stream és obrir l'aixeta i treballar amb el que va sortint.

Els Transform són el tema de la lliçó següent. Aquí ens concentrem en els dos primers, que són la base de tota la resta.

Això no és cap analogia: Readable i Writable hereten d'EventEmitter. on, once, emit, off i tot el de la lliçó 02-05 funciona tal qual, inclosa la regla d'or que ara pren el seu veritable sentit.

Dos advertiments que valen per mitja lliçó. El primer: end i finish no són el mateix. end és de la banda que llegeix («s'ha acabat l'entrada») i finish és de la banda que escriu («he acabat d'abocar»); confondre'ls porta a tancar un fitxer abans que les seves dades siguin al disc. El segon: l'esdeveniment error d'un stream segueix la regla de l'EventEmitter que ja coneixes —un error sense oients llança una excepció no capturada i mata el procés—. Un stream sense on('error') és una caiguda de producció esperant a passar, i el cas més habitual no és un disc trencat: és un ENOENT perquè el fitxer no hi era.

Un Readable té dues maneres de lliurar dades, i aquesta dualitat és la font de la meitat dels problemes de qui comença amb streams.

No els barregis. Subscriure's a data i cridar read() al mateix stream produeix comportaments difícils de raonar: trossos que es perden, end que no arriba, ordre alterat. Tria un mode per stream i mantén-lo —encara que a la pràctica la recomanació és no fer-ne servir cap directament: pipe, pipeline o for await...of resolen el problema millor—. I un tercer detall del mode fluid: si et subscrius a data després d'un await, pots perdre trossos, perquè l'stream va començar a fluir abans que arribés el teu oient. Subscriu-t'hi sempre de manera síncrona, just després de crear l'stream.

El highWaterMark és la mida de la memòria intermèdia interna: quants bytes acumula l'stream abans de considerar que ja en té prou. Abaixar-lo (16 KB) gasta menys memòria per stream a canvi de més crides al sistema i més esdeveniments; apujar-lo (1 MB) fa el contrari. Els 64 KB per defecte són raonables gairebé sempre: abaixar-lo té sentit quan manegues milers d'streams simultanis —un servidor amb moltes connexions— i apujar-lo quan processes pocs fitxers molt grans. En mode objecte (que veuràs a la lliçó següent) el highWaterMark compta objectes, no bytes, i val 16 per defecte.

Un advertiment sobre encoding: si no l'indiques, els trossos arriben com a Buffer; i si l'indiques com a 'utf8', l'stream s'encarrega que cap caràcter multibyte no quedi partit entre dos trossos —un problema real que veurem a fons a la lliçó 03-06 amb StringDecoder—.

Aquí hi ha el concepte que separa qui fa servir streams de qui els entén. Imagina que llegeixes d'un SSD ràpid i escrius en un disc lent o en una connexió de xarxa: les dades entren a 500 MB/s i surten a 50 MB/s. On va la diferència? A la memòria intermèdia interna de l'stream d'escriptura, que creix sense parar fins a exhaurir la memòria. És a dir: has canviat un readFile que consumia 800 MB per un stream que consumeix... 800 MB.

La contrapressió (backpressure) és el mecanisme que ho impedeix, i es recolza en un detall que gairebé tothom ignora:

write() torna un booleà. I aquest booleà no és informatiu: és una ordre.

Aquest codi funciona amb fitxers petits i peta amb els grans: la memòria intermèdia de l'escriptor creix indefinidament perquè ningú no deixa d'alimentar-la. És la fallada més comuna en escriure streams a mà, i la més difícil de diagnosticar, perquè en desenvolupament —fitxer petit, disc ràpid— no es manifesta mai.

El cicle complet és: data → write() torna false → pause() → la memòria intermèdia es buida i l'escriptor emet drain → resume() → tornada a data. El resultat és que el consumidor lent marca el ritme del productor ràpid, i la memòria es manté acotada pel highWaterMark. Aquesta idea no és exclusiva dels fitxers: és la mateixa que governa TCP i la que farà que un client lent no tombi el teu servidor HTTP al Mòdul 4.

Escriure a mà el ball de pause/resume/drain a cada canonada seria insuportable. pipe() ho fa per tu:

Aquesta línia equival a tot el bloc anterior: connecta els dos streams, gestiona la contrapressió i crida escriptor.end() quan el lector acaba. Es poden encadenar —a.pipe(b).pipe(c)— perquè pipe torna l'stream de destinació.

Però pipe té un defecte greu i poc conegut: no propaga errors ni neteja el que deixa enrere. Si lector falla —un ENOENT, un disc amb errors—, l'error no arriba a escriptor, que queda obert amb el seu descriptor sense alliberar; i si ningú no escolta error a lector, el procés mor.

Amb una canonada de tres o quatre etapes el problema es multiplica: cal subscriure on('error') a cada stream i tancar a mà els que quedin penjant. És tediós, és fàcil oblidar-se'n d'un, i aquest oblit es paga en descriptors perduts fins a l'EMFILE.

Per això la recomanació moderna és rotunda: fes servir pipe() per entendre el concepte i stream.pipeline per treballar. pipeline propaga errors, destrueix tots els streams de la cadena quan alguna cosa falla i té versió amb promeses. És el tema central de la lliçó següent.

Necessitem dades amb què treballar. Aquest és el format de l'històric de vendes:

Les columnes segueixen les convencions del projecte: codiEntrada amb el format EV-<any>-<6 dígits>, sessioId amb la forma ses-NNN-M, dataVenda com a cadena ISO sense zona, preuCentims enter i canal amb un de tres valors (web, taquilla, app). Generem un fitxer coherent amb la llavor: exactament 1811 vendes, repartides segons les venudes de cada sessió.

El comentari sobre la contrapressió és deliberat: amb 1811 línies no hi ha problema, però el codi correcte per a volums reals hauria de respectar el retorn de write() o, millor, construir-se amb pipeline i un generador, com farem a la lliçó següent.

Un CSV es processa per línies, però els trossos d'un stream no coincideixen amb les línies: un tros de 64 KB talla per la meitat la línia que li toqui. Al laboratori de l'apartat 1 ho vam resoldre a mà amb una variable resta; el mòdul node:readline ho fa bé i sense codi propi.

Tres decisions dignes d'esment. crlfDelay: Infinity evita que un fitxer generat a Windows produeixi línies acabades en \r invisible que espatlla l'últim camp. Una línia corrupta es registra i se salta, no avorta l'informe: en un històric de milions de registres sempre hi ha brossa, i un informe que mor a la línia 4.000.000 no serveix de res. I el consum de memòria és constant: perSessio guarda una entrada per sessió (set), no per venda.

Aquest for await (const linia of lector) mereix explicació, perquè és la manera més llegible de consumir un stream i ja la coneixes de la lliçó 02-05, quan vas veure events.on().

Tot Readable és un iterable asíncron. Es pot recórrer directament:

Els seus avantatges sobre on('data') són concrets: la contrapressió és automàtica (el bucle no demana el tros següent fins que acaba el cos) en comptes de manual; els errors es capturen amb un try/catch normal al voltant del bucle en comptes d'amb un on('error') a part; es pot fer servir await a dins; i un break destrueix l'stream tot sol, sense destroy() explícit.

Aquest await dins del bucle és la diferència decisiva. Si per cada línia del CSV cal consultar una base de dades, amb on('data') les consultes es llançarien totes alhora i enfonsarien el servidor; amb for await, la lectura s'atura sola fins que la consulta acaba, així que la contrapressió surt de franc. El preu és rendiment brut: for await és una mica més lent perquè cada iteració crea una promesa. Per a un procés per lots és irrellevant; en una ruta crítica amb molt trànsit, mesura-ho.

• No subscriure on('error'). Un error sense oients en un EventEmitter mata el procés, i un stream és un EventEmitter. La solució definitiva és pipeline.
• Ignorar el retorn de write(). Funciona en desenvolupament i exhaureix la memòria en producció. Si escrius a mà, respecta el drain.
• Confondre end amb finish. end és el lector exhaurit; finish és l'escriptor buidat. Tancar per l'esdeveniment equivocat talla dades.
• Suposar que un tros és una línia. Mai no ho és. Fes servir readline o un Transform que trossegi.
• Barrejar mode fluid i pausat, o subscriure's a data després d'un await. En el primer cas el comportament és erràtic; en el segon perds trossos.
• Acumular tots els trossos en un array per ajuntar-los al final. Això és readFile amb passos extra: has perdut l'únic avantatge de l'stream.
• Consell: si el fitxer cap folgadament a la memòria i només el llegeixes un cop, readFile és més simple i més ràpid; els streams són per al que és gran, infinit o el que arriba a poc a poc. I quan dubtis de si hi ha contrapressió, imprimeix escriptor.writableLength de tant en tant: si creix sense parar, l'estàs ignorant.

Amplia src/informes/vendes-per-sessio.js amb una funció resumirPerCanal() que recorri dades/vendes.csv una sola vegada i torni, per canal, el nombre d'entrades, la recaptació total i l'hora punta (l'hora del dia amb més vendes). Ha de fer servir readline, mantenir memòria constant i validar que la suma d'entrades de tots els canals és 1811.

Escriu src/laboratori/copiar-amb-contrapressio.js que copiï un fitxer gran sense fer servir pipe, respectant el drain, i imprimeixi cada segon: megabytes copiats, writableLength de l'escriptor i nombre de pauses per contrapressió. Executa després una versió que ignori el retorn de write() i compara el consum de memòria amb process.memoryUsage().rss.

Escriu src/laboratori/filtrar-vendes.js que llegeixi dades/vendes.csv, es quedi només amb les vendes les sessions de les quals pertanyin a una sala donada (--sala="Sala Boveda", creuant amb el catàleg) i escrigui el resultat a dades/vendes-<sala>.csv conservant la capçalera. Fes servir readline per llegir i un createWriteStream per escriure, respectant la contrapressió.

Solució 1. La clau és acumular en dos nivells durant el mateix recorregut:

L'array de 24 posicions per canal és el truc que manté la memòria constant: no es guarden les vendes, només el comptador de cada hora. Un Map de dates completes creixeria amb el fitxer.

Solució 2. El nucli de la còpia manual:

Amb la contrapressió respectada, writableLength oscil·la al voltant del highWaterMark i el rss es manté pla. Sense ella, writableLength creix sense límit i el rss puja fins a acostar-se a la mida del fitxer. Fixa't també en els dos on('error') creuats que calen per no perdre descriptors: aquesta feina manual és exactament el que pipeline automatitza.

Solució 3. El que importa és creuar el catàleg abans de començar a llegir el CSV, per no fer-ho per línia:

Aquest await new Promise(... 'drain' ...) és la traducció de la contrapressió al món d'async/await, i és un patró que val la pena memoritzar: dins d'un for await, esperar el drain atura també la lectura, perquè el bucle no demana l'element següent fins que acaba el cos. Un Set amb els ids de sessió evita recórrer el catàleg 1811 vegades.

Ja saps per què existeixen els streams i quin problema resolen exactament: has mesurat la diferència entre 1621 MB de readFile i 6 MB d'un stream sobre el mateix fitxer, i entens que el consum d'un stream no depèn de la mida de l'entrada. Coneixes els quatre tipus —Readable, Writable, Duplex i Transform— i el fet fonamental que són EventEmitter, amb data, end, error, close, finish i drain, sense confondre end amb finish i sabent que un error sense oients tomba el procés.

Distingeixes el mode fluid del pausat i per què no s'han de barrejar; saps què és el highWaterMark i què es guanya i es perd movent-lo. I, sobretot, entens la contrapressió: que write() torna un booleà que és una ordre, que ignorar-lo converteix el teu stream en un readFile disfressat, i que el cicle pause → drain → resume és el que fa que el consumidor lent marqui el ritme. Saps que pipe() automatitza aquest ball però no propaga errors ni neteja descriptors, i per això no és l'eina definitiva. Escena Viva, mentrestant, ja té el seu dades/vendes.csv amb les 1811 vendes de la llavor i un src/informes/vendes-per-sessio.js que el recorre línia a línia amb readline, acumulant entrades, recaptació i canals per sessió amb memòria constant. I has vist la manera més llegible de consumir qualsevol stream: for await...of, que porta la contrapressió de regal i permet await dins del bucle.

A Streams de Transformació i pipeline tanquem el cercle. Aprendràs a escriure els teus propis streams amb Transform (_transform i _flush), a muntar canonades de procés completes —CSV cru a objectes, filtratge per sala, agregació per sessió, escriptura de l'informe—, i a substituir pipe() per stream.pipeline, que propaga errors, destrueix tota la cadena quan alguna cosa falla i té versió amb promeses. Veuràs a més Readable.from(), els generadors asíncrons com a transformacions, la compressió amb zlib intercalada a la canonada i la cancel·lació amb AbortSignal.