Node.js Streams: Procesamiento eficiente de datos

Los Streams son una característica fundamental de Node.js que te permite procesar datos de manera eficiente, especialmente cuando trabajas con grandes volúmenes.

¿Por qué usar Streams?

// ❌ Malo: Carga todo el archivo en memoria
const fs = require('fs');
const data = fs.readFileSync('archivo-grande.txt');
console.log(data);

// ✅ Bueno: Procesa el archivo en chunks
const stream = fs.createReadStream('archivo-grande.txt');
stream.on('data', (chunk) => {
  console.log(chunk);
});

Tipos de Streams

1. Readable Streams

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

const readable = new Readable({
  read(size) {
    this.push('datos ');
    this.push('en ');
    this.push('chunks\n');
    this.push(null); // No más datos
  }
});

readable.on('data', (chunk) => {
  console.log(chunk.toString());
});

2. Writable Streams

const { Writable } = require('stream');
const fs = require('fs');

const writable = fs.createWriteStream('output.txt');

writable.write('Primera línea\n');
writable.write('Segunda línea\n');
writable.end('Última línea\n');

writable.on('finish', () => {
  console.log('Escritura completa');
});

3. Duplex Streams

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

const duplex = new Duplex({
  read(size) {
    this.push('datos de lectura\n');
    this.push(null);
  },
  write(chunk, encoding, callback) {
    console.log('Escribiendo:', chunk.toString());
    callback();
  }
});

4. Transform Streams

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

const uppercase = new Transform({
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
});

process.stdin
  .pipe(uppercase)
  .pipe(process.stdout);

Pipe: Conectando Streams

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

// Leer archivo, comprimir y guardar
fs.createReadStream('input.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('input.txt.gz'));

Ejemplo práctico: Procesador de CSV

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

class CSVParser extends Transform {
  constructor() {
    super({ objectMode: true });
    this.header = null;
  }

  transform(chunk, encoding, callback) {
    const lines = chunk.toString().split('\n');
    
    lines.forEach((line, index) => {
      if (!this.header) {
        this.header = line.split(',');
      } else if (line.trim()) {
        const values = line.split(',');
        const obj = {};
        this.header.forEach((key, i) => {
          obj[key] = values[i];
        });
        this.push(obj);
      }
    });
    
    callback();
  }
}

fs.createReadStream('data.csv')
  .pipe(new CSVParser())
  .on('data', (obj) => {
    console.log(obj);
  });

Backpressure: Manejo de flujo

const fs = require('fs');

const readable = fs.createReadStream('input.txt');
const writable = fs.createWriteStream('output.txt');

readable.on('data', (chunk) => {
  const canContinue = writable.write(chunk);
  
  if (!canContinue) {
    // El buffer está lleno, pausar lectura
    readable.pause();
  }
});

writable.on('drain', () => {
  // El buffer se vació, continuar lectura
  readable.resume();
});

Stream de API HTTP

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

https.get('https://example.com/largefile.zip', (response) => {
  const file = fs.createWriteStream('downloaded.zip');
  
  let downloadedSize = 0;
  const totalSize = parseInt(response.headers['content-length'], 10);
  
  response.on('data', (chunk) => {
    downloadedSize += chunk.length;
    const progress = (downloadedSize / totalSize * 100).toFixed(2);
    console.log(`Descargado: ${progress}%`);
  });
  
  response.pipe(file);
  
  file.on('finish', () => {
    file.close();
    console.log('Descarga completa');
  });
});

Pipeline: Manejo de errores

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

pipeline(
  fs.createReadStream('input.txt'),
  zlib.createGzip(),
  fs.createWriteStream('input.txt.gz'),
  (err) => {
    if (err) {
      console.error('Pipeline falló:', err);
    } else {
      console.log('Pipeline exitoso');
    }
  }
);

Conclusión

Los Streams son esenciales para aplicaciones Node.js eficientes. Te permiten procesar grandes cantidades de datos sin saturar la memoria, haciendo tu aplicación más escalable y performante.