Los Streams son colecciones de datos procesados por fragmentos, lo que permite un manejo eficiente de archivos grandes y red sin saturar la memoria.
Descripción de Tipos de Streams
| Tipo | Dirección | Descripción |
|---|---|---|
Readable |
Solo lectura | Origen desde el cual se pueden leer y consumir datos |
Writable |
Solo escritura | Destino en el que se pueden escribir datos |
Transform |
Lectura y escritura | Transforma o modifica los datos mientras pasan a traves de el |
Duplex |
Lectura y escritura | Tanto legible como escribible con canales independientes (ej. socket TCP) |
Pipeline y el moderno stream/promises
Node.js moderno proporciona pipeline para encadenar flujos de forma segura con gestion automatica de errores y cierre de recursos:
const { pipeline } = require('stream/promises');
const fs = require('fs');
const zlib = require('zlib');
async function compressFile() {
await pipeline(
fs.createReadStream('source.tar'),
zlib.createGzip(),
fs.createWriteStream('source.tar.gz')
);
console.log('Compresion completada');
}
Creación de Streams Personalizados
1. Stream Readable Personalizado
const { Readable } = require('stream');
function createTimestampStream() {
let count = 0;
return new Readable({
objectMode: true,
read() {
if (count >= 5) {
this.push(null); // Fin del flujo de datos
} else {
this.push({ id: ++count, time: new Date() });
}
}
});
}
2. Stream Transform Personalizado
const { Transform } = require('stream');
const uppercaseTransformer = new Transform({
transform(chunk, encoding, callback) {
const transformed = chunk.toString().toUpperCase();
callback(null, transformed);
}
});
3. Stream Writable Personalizado
const { Writable } = require('stream');
const consoleWriter = new Writable({
objectMode: true,
write(chunk, encoding, callback) {
console.log('Escribiendo fragmento:', chunk);
callback();
}
});
Modo Flowing frente a Modo Paused
Los flujos de lectura operan en modo pausado (paused) o en modo fluido (flowing):
const stream = fs.createReadStream('data.csv');
// Anadir un escuchador 'data' activa automaticamente el modo flowing
stream.on('data', (chunk) => {
console.log(`Recibidos ${chunk.length} bytes`);
});
// Control explicito de flujo
stream.pause(); // Suspender temporalmente eventos
stream.resume(); // Reanudar la emision de eventos
Eventos de los Streams
| Evento | Aplicable a | Condición de Disparo |
|---|---|---|
data |
Readable |
Se emite cuando hay un fragmento de datos listo para consumirse |
end |
Readable |
Se emite cuando ya no hay mas datos disponibles |
finish |
Writable |
Se emite cuando todos los datos se han vaciado en el destino |
error |
All | Se emite ante cualquier error no controlado en el flujo |
close |
All | Se emite cuando se cierra el descriptor de recurso subyacente |