Eventos y streams

Gran parte de Node está construida sobre dos ideas: los eventos (algo pasa y reaccionas) y los streams (procesar datos por pedacitos en lugar de todo de golpe). Gracias a ellos puedes procesar un archivo de 10 GB usando sólo unos MB de memoria.

Concepto

Eventos

Un EventEmitter es como un timbre: alguien lo toca (emit) y todos los que están escuchando (on) reaccionan. Servidores HTTP, sockets y streams son EventEmitters.

Streams

Imagina llenar una alberca:

• Sin streams: esperas a tener toda el agua en un camión gigante y luego la viertes (cargar todo el archivo en memoria).
• Con streams: conectas una manguera y el agua fluye poco a poco (procesas pedazo por pedazo, llamados chunks).

Tipos de streams:

Tipo	Ejemplo
Readable (se lee)	fs.createReadStream, petición HTTP entrante
Writable (se escribe)	fs.createWriteStream, respuesta HTTP
Duplex (ambos)	sockets TCP
Transform (modifica al pasar)	compresión gzip, cifrado

EventEmitter

// pedidos.js
import { EventEmitter } from 'node:events'

class Tienda extends EventEmitter {
  crearPedido(cliente, total) {
    const pedido = { id: Date.now(), cliente, total }
    // Emitimos un evento con datos
    this.emit('pedido:creado', pedido)
    if (total > 5000) this.emit('pedido:grande', pedido)
    return pedido
  }
}

const tienda = new Tienda()

// Varios "oyentes" independientes para el mismo evento
tienda.on('pedido:creado', (p) => console.log(`📧 Email de confirmación a ${p.cliente}`))
tienda.on('pedido:creado', (p) => console.log(`📦 Avisar a almacén del pedido ${p.id}`))
tienda.on('pedido:grande', (p) => console.log(`⭐ Pedido grande de $${p.total}: avisar a ventas`))

// once: se ejecuta sólo la primera vez
tienda.once('pedido:creado', () => console.log('🎉 ¡Primer pedido del día!'))

tienda.crearPedido('ana@correo.com', 1200)
console.log('---')
tienda.crearPedido('luis@correo.com', 8000)

Métodos clave: on, once, off (quitar oyente), emit, listenerCount.

Error común

Si emites un evento 'error' y nadie lo escucha, Node lanza la excepción y el proceso se cae. Siempre registra emisor.on('error', ...) en streams y emisores propios.

Esperar un evento con promesas

import { once } from 'node:events'

const [pedido] = await once(tienda, 'pedido:creado')
console.log('Llegó el pedido', pedido.id)

Streams de lectura y escritura

// copiar-grande.js
import { createReadStream, createWriteStream } from 'node:fs'

const lectura = createReadStream('video.mp4')        // lee en chunks de 64 KB
const escritura = createWriteStream('copia-video.mp4')

let bytes = 0
lectura.on('data', (chunk) => {
  bytes += chunk.length
})

lectura.on('end', () => console.log(`Copiados ${(bytes / 1024 / 1024).toFixed(1)} MB`))
lectura.on('error', (err) => console.error('Error leyendo:', err.message))

// pipe conecta la salida de uno con la entrada del otro
lectura.pipe(escritura)

pipeline: la forma correcta

pipe no propaga errores ni cierra los streams si algo falla. Usa pipeline de node:stream/promises:

// comprimir.js
import { createReadStream, createWriteStream } from 'node:fs'
import { createGzip } from 'node:zlib'
import { pipeline } from 'node:stream/promises'

const [, , archivo] = process.argv

try {
  await pipeline(
    createReadStream(archivo),
    createGzip(),                       // Transform: comprime al pasar
    createWriteStream(`${archivo}.gz`)
  )
  console.log('Comprimido ✔')
} catch (error) {
  console.error('Falló la compresión:', error.message)
}

node comprimir.js log-gigante.txt

Leer un archivo línea por línea

// lineas.js — cuenta errores en un log enorme sin cargarlo en memoria
import { createReadStream } from 'node:fs'
import readline from 'node:readline'

const rl = readline.createInterface({
  input: createReadStream('servidor.log'),
  crlfDelay: Infinity
})

let total = 0
let errores = 0
for await (const linea of rl) {
  total++
  if (linea.includes('ERROR')) errores++
}
console.log(`${errores} errores en ${total} líneas`)

Crear tus propios streams

Transform

// mayusculas.js
import { Transform } from 'node:stream'
import { pipeline } from 'node:stream/promises'

const aMayusculas = new Transform({
  transform(chunk, encoding, callback) {
    // callback(error, datosTransformados)
    callback(null, chunk.toString().toUpperCase())
  }
})

// Lee del teclado (stdin), transforma y escribe en pantalla (stdout)
await pipeline(process.stdin, aMayusculas, process.stdout)

echo "hola mundo" | node mayusculas.js
# HOLA MUNDO

Readable desde un generador

import { Readable } from 'node:stream'

async function* numeros() {
  for (let i = 1; i <= 5; i++) {
    yield `Número ${i}\n`
  }
}

Readable.from(numeros()).pipe(process.stdout)

Caso real: convertir CSV a JSON por streams

// csv-a-json.js
import { createReadStream, createWriteStream } from 'node:fs'
import { Transform } from 'node:stream'
import { pipeline } from 'node:stream/promises'
import readline from 'node:readline'

const entrada = process.argv[2] ?? 'clientes.csv'
const salida = entrada.replace(/\.csv$/, '.json')

const rl = readline.createInterface({ input: createReadStream(entrada), crlfDelay: Infinity })

// Generador que convierte cada línea en JSON
async function* aJson(lineas) {
  let encabezados = null
  let primera = true
  yield '[\n'
  for await (const linea of lineas) {
    if (!linea.trim()) continue
    const valores = linea.split(',').map((v) => v.trim())
    if (!encabezados) {
      encabezados = valores
      continue
    }
    const obj = Object.fromEntries(encabezados.map((h, i) => [h, valores[i]]))
    yield (primera ? '  ' : ',\n  ') + JSON.stringify(obj)
    primera = false
  }
  yield '\n]\n'
}

await pipeline(aJson(rl), createWriteStream(salida))
console.log(`Generado ${salida}`)

Con un clientes.csv así:

nombre,email,ciudad
Ana,ana@correo.com,CDMX
Luis,luis@correo.com,Monterrey

Buena práctica

Para archivos de más de unos MB, exportaciones, descargas y subidas, usa streams y pipeline. Tu servidor usará memoria constante sin importar el tamaño del archivo.

Backpressure

Si escribes más rápido de lo que el destino puede recibir, los datos se acumulan en memoria. pipe y pipeline manejan esto automáticamente (backpressure): pausan la lectura hasta que el destino se desahoga.

Ejercicios

1. Crea un Temporizador que extienda EventEmitter y emita 'tick' cada segundo con el número de segundo, y 'fin' al llegar a N.
2. Crea un Transform que numere las líneas de un archivo y escribe el resultado en numerado.txt.

Ver solución

// temporizador.js
import { EventEmitter } from 'node:events'

class Temporizador extends EventEmitter {
  iniciar(segundos) {
    let actual = 0
    const id = setInterval(() => {
      actual++
      this.emit('tick', actual)
      if (actual === segundos) {
        clearInterval(id)
        this.emit('fin')
      }
    }, 1000)
  }
}

const t = new Temporizador()
t.on('tick', (s) => console.log(`Segundo ${s}`))
t.on('fin', () => console.log('¡Tiempo!'))
t.iniciar(3)

// numerar.js
import { createReadStream, createWriteStream } from 'node:fs'
import { pipeline } from 'node:stream/promises'
import readline from 'node:readline'

const rl = readline.createInterface({ input: createReadStream(process.argv[2]), crlfDelay: Infinity })

async function* numerar(lineas) {
  let n = 1
  for await (const linea of lineas) {
    yield `${String(n++).padStart(4, ' ')} | ${linea}\n`
  }
}

await pipeline(numerar(rl), createWriteStream('numerado.txt'))
console.log('Listo')

Resumen

• EventEmitter: on para escuchar, emit para disparar; escucha siempre 'error'.
• Los streams procesan datos en chunks con memoria constante.
• Usa pipeline de node:stream/promises para conectar streams con manejo de errores.
• readline + for await lee archivos enormes línea por línea.