Node.js Streams: Procesamiento Eficiente de Datos

Streams en Node.js

Los streams permiten procesar datos de forma eficiente sin cargar todo en memoria. Esenciales para aplicaciones de alto rendimiento.

Tipos de Streams

// Readable: Leer datos
const fs = require('fs');
const readStream = fs.createReadStream('file.txt');

// Writable: Escribir datos
const writeStream = fs.createWriteStream('output.txt');

// Duplex: Leer y escribir
const { Duplex } = require('stream');

// Transform: Modificar datos mientras fluyen
const { Transform } = require('stream');

Leer Archivos con Streams

const fs = require('fs');

// ❌ Sin streams - carga todo en memoria
const data = fs.readFileSync('large-file.txt');
console.log(data);

// ✅ Con streams - procesa por chunks
const readStream = fs.createReadStream('large-file.txt', {
  encoding: 'utf8',
  highWaterMark: 64 * 1024 // 64KB chunks
});

readStream.on('data', (chunk) => {
  console.log('Chunk:', chunk.length);
});

readStream.on('end', () => {
  console.log('Finished reading');
});

readStream.on('error', (error) => {
  console.error('Error:', error);
});

Piping Streams

// Copiar archivo
const readStream = fs.createReadStream('input.txt');
const writeStream = fs.createWriteStream('output.txt');

readStream.pipe(writeStream);

// Comprimir archivo
const zlib = require('zlib');
const gzip = zlib.createGzip();

fs.createReadStream('input.txt')
  .pipe(gzip)
  .pipe(fs.createWriteStream('input.txt.gz'));

// Descomprimir
fs.createReadStream('input.txt.gz')
  .pipe(zlib.createGunzip())
  .pipe(fs.createWriteStream('output.txt'));

Transform Stream

const { Transform } = require('stream');

// Convertir a mayúsculas
const upperCaseTransform = new Transform({
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
});

fs.createReadStream('input.txt')
  .pipe(upperCaseTransform)
  .pipe(fs.createWriteStream('output.txt'));

// CSV Parser
const csvTransform = new Transform({
  transform(chunk, encoding, callback) {
    const lines = chunk.toString().split('\n');
    lines.forEach(line => {
      const [name, age] = line.split(',');
      this.push(JSON.stringify({ name, age: parseInt(age) }) + '\n');
    });
    callback();
  }
});

fs.createReadStream('data.csv')
  .pipe(csvTransform)
  .pipe(fs.createWriteStream('data.json'));

HTTP Streams

const http = require('http');
const fs = require('fs');

// Servir archivo grande
http.createServer((req, res) => {
  const readStream = fs.createReadStream('large-video.mp4');
  
  res.writeHead(200, {
    'Content-Type': 'video/mp4'
  });
  
  readStream.pipe(res);
}).listen(3000);

// Upload con streams
app.post('/upload', (req, res) => {
  const writeStream = fs.createWriteStream('uploaded-file.txt');
  
  req.pipe(writeStream);
  
  writeStream.on('finish', () => {
    res.send('Upload complete');
  });
});

Backpressure

// Manejar backpressure correctamente
const readStream = fs.createReadStream('input.txt');
const writeStream = fs.createWriteStream('output.txt');

readStream.on('data', (chunk) => {
  const canContinue = writeStream.write(chunk);
  
  if (!canContinue) {
    // Buffer lleno, pausar lectura
    readStream.pause();
  }
});

writeStream.on('drain', () => {
  // Buffer vaciado, continuar lectura
  readStream.resume();
});

// O simplemente usa pipe (maneja backpressure automáticamente)
readStream.pipe(writeStream);

Stream Pipeline

const { pipeline } = require('stream');
const zlib = require('zlib');

// Pipeline maneja errores y cleanup automáticamente
pipeline(
  fs.createReadStream('input.txt'),
  zlib.createGzip(),
  fs.createWriteStream('input.txt.gz'),
  (err) => {
    if (err) {
      console.error('Pipeline failed:', err);
    } else {
      console.log('Pipeline succeeded');
    }
  }
);

Async Iterators

// Leer stream con for await
const fs = require('fs');

async function processFile() {
  const readStream = fs.createReadStream('large-file.txt');
  
  for await (const chunk of readStream) {
    console.log('Processing chunk:', chunk.length);
    // Procesar chunk
  }
  
  console.log('Done');
}

processFile();

Mejores Prácticas

1. Usa streams para archivos grandes: Evita cargar todo en memoria
2. Pipe cuando sea posible: Maneja backpressure automáticamente
3. Pipeline para múltiples streams: Mejor manejo de errores
4. Cleanup: Siempre maneja eventos de error
5. Chunks apropiados: Ajusta highWaterMark según necesidad

Conclusión

Los streams son fundamentales para aplicaciones Node.js eficientes. Permiten procesar grandes volúmenes de datos con mínimo uso de memoria, esencial para aplicaciones escalables.