Bus de Eventos

Bus de Eventos

Veloce-TS incluye un bus de eventos en proceso integrado para desacoplar módulos sin Redis ni Kafka. Úsalo para comunicación entre módulos, eventos de dominio y para alimentar streams SSE.

EventBus

import { EventBus } from 'veloce-ts';

const bus = new EventBus();

O usa el singleton pre-construido:

import { globalEvents } from 'veloce-ts';

API

on(event, handler) — suscribirse

globalEvents.on('user.created', async (payload) => {
  await emailService.sendWelcome(payload.email);
});

emit() llama a todos los handlers de forma concurrente y se resuelve cuando todos terminaron. emitSync() es la variante “dispara y olvida” para handlers sincrónicos — no los espera.

once(event, handler) — suscribirse una vez

El handler se elimina automáticamente después de la primera llamada.

globalEvents.once('app.ready', () => {
  console.log('App lista');
});

emit(event, payload) — disparar async (paralelo)

Devuelve una Promise<void> que se resuelve cuando todos los handlers terminaron.

await globalEvents.emit('user.created', { userId: '123', email: 'usuario@ejemplo.com' });

emitSync(event, payload) — disparar sync

Llama a los handlers en orden de registro y devuelve void — no los espera. Un handler async pasado a emitSync se inicia pero no se aguarda, así que usa emit() siempre que los handlers hagan trabajo asíncrono.

globalEvents.emitSync('order.placed', { orderId: '456' });

Manejo de errores

Cambió en v2.0.0

Un handler que lanza ya no impide que los demás se ejecuten.

Antes emit() usaba Promise.all, así que el primer rechazo abortaba el emit completo y cualquier handler al que no se había llegado se saltaba en silencio — un solo listener defectuoso podía detener el resto de tus efectos secundarios.

Ahora emit() y emitSync() ejecutan todos los listeners, recolectan los fallos y los relanzan juntos como un AggregateError una vez que todos terminaron. Los listeners once se eliminan incluso si lanzan.

globalEvents.on('user.created', async () => { throw new Error('mailer down'); });
globalEvents.on('user.created', async () => { await analytics.track('signup'); });

try {
  await globalEvents.emit('user.created', { userId: '123' });
} catch (error) {
  // analytics.track() sí se ejecutó — aquí solo se reporta el listener que falló
  if (error instanceof AggregateError) {
    for (const cause of error.errors) {
      logger.error('Event listener failed', cause);
    }
  }
}

Si un listener que falla nunca debe interrumpir al emisor, captura dentro del propio handler:

globalEvents.on('user.created', async (payload) => {
  try {
    await emailService.sendWelcome(payload.email);
  } catch (error) {
    logger.error('Welcome email failed', error);
  }
});

off(event, handler) — desuscribirse

const handler = (payload: any) => console.log(payload);
globalEvents.on('user.created', handler);
// más tarde...
globalEvents.off('user.created', handler);

removeAllListeners(event?) — eliminar listeners

globalEvents.removeAllListeners('user.created');  // limpiar evento específico
globalEvents.removeAllListeners();                 // limpiar todos los eventos

listenerCount(event) — contar listeners

const count = globalEvents.listenerCount('user.created');

El decorador @On

Nuevo en v3.3.0

Listeners declarativos. Hasta ahora globalEvents.on(...) era la única forma de suscribirse.

Una clase cuyos métodos llevan @On se registra con app.include(), igual que un controlador. Su instancia sale del contenedor de DI, así que un listener puede inyectar los mismos servicios que un controlador:

import { On, Inject } from 'veloce-ts';

class NotificationListeners {
  constructor(@Inject(Mailer) private mailer: Mailer) {}

  @On('user.created')
  async welcome(payload: { email: string; name: string }) {
    await this.mailer.send({ to: payload.email, subject: `¡Bienvenido, ${payload.name}!` });
  }

  @On('app.ready', { once: true })
  warmCaches() {
    // se elimina tras la primera entrega
  }
}

app.include(NotificationListeners);

Emitir no cambia — el decorador solo reemplaza el lado de la suscripción:

await globalEvents.emit('user.created', { email: 'ada@example.com', name: 'Ada' });

Qué hace la aplicación por ti

Se suscribe en compile()	No al importar, así que un listener puede depender de servicios registrados más adelante en el arranque.
Se desuscribe en shutdown()	Sin esto, los listeners atados al globalEvents del proceso sobreviven a la app que los creó, y cada app construida en una suite de tests responde al mismo evento.
Resuelve por DI	Inyección por constructor con @Inject, igual que un controlador. Una sola instancia singleton atiende todos los métodos @On de la clase.
Falla ruidosamente ante un nombre inválido	Un @On que apunta a algo que no es un método lanza en compile(), nombrando la clase y la propiedad — no en silencio al emitir.

Una clase, ambos roles

Un controlador también puede escuchar; include() ya no se detiene en la primera coincidencia:

@Controller('/orders')
class OrderController {
  @Get('/')
  list() { return this.service.findAll(); }

  @On('order.placed')
  async invalidateCache() { await cache.delete('orders:list'); }
}

El manejo de errores es el del bus

@On no cambia nada de la entrega: un listener que lanza no detiene a los demás, y los fallos vuelven de emit() juntos como un AggregateError. Ver Manejo de errores más arriba. Si un listener nunca debe interrumpir al emisor, captura dentro del método.

Usar tu propio bus

@On se suscribe al bus al que está conectada la aplicación — globalEvents salvo que pases otro. Pasar el tuyo evita que una suite de tests comparta suscripciones entre apps:

import { EventBus } from 'veloce-ts';

const bus = new EventBus();
const app = new VeloceTS({ eventBus: bus });

app.include(NotificationListeners);
await app.compile();

await bus.emit('user.created', { email: 'ada@example.com', name: 'Ada' });

app.getEventBus() devuelve el bus en uso.

Patrones

Eventos de dominio

user.service.ts

import { globalEvents } from 'veloce-ts';

class UserService {
  async createUser(data: CreateUserDto) {
    const user = await db.insert(users).values(data).returning();
    await globalEvents.emit('user.created', {
      userId: user.id,
      email: user.email,
      name: user.name,
    });
    return user;
  }
}

// email.service.ts — se registra al arrancar
import { globalEvents } from 'veloce-ts';

globalEvents.on('user.created', async ({ email, name }) => {
  await mailer.send({
    to: email,
    subject: `¡Bienvenido, ${name}!`,
    template: 'welcome',
  });
});

// audit.service.ts — múltiples handlers para el mismo evento
globalEvents.on('user.created', async ({ userId }) => {
  await db.insert(auditLog).values({
    event: 'user.created',
    resourceId: userId,
    timestamp: new Date(),
  });
});

Alimentar streams SSE mediante el bus de eventos

// Combinar EventBus + streaming SSE
@Get('/events')
@SSE()
async* streamUserEvents(): AsyncGenerator<{ data: string; event?: string }> {
  yield { data: 'conectado', event: 'open' };

  const queue: unknown[] = [];
  const handler = (payload: unknown) => queue.push(payload);

  globalEvents.on('user.created', handler);
  globalEvents.on('user.updated', handler);

  try {
    while (true) {
      if (queue.length > 0) {
        const event = queue.shift()!;
        yield { data: JSON.stringify(event), event: 'update' };
      } else {
        await new Promise(r => setTimeout(r, 100));
      }
    }
  } finally {
    globalEvents.off('user.created', handler);
    globalEvents.off('user.updated', handler);
  }
}

Múltiples módulos, un bus

// Módulo A
import { globalEvents } from 'veloce-ts';

export class OrderService {
  async placeOrder(order: Order) {
    const placed = await db.createOrder(order);
    await globalEvents.emit('order.placed', { orderId: placed.id, amount: order.total });
    return placed;
  }
}

// Módulo B — maneja eventos de orden de forma independiente
import { globalEvents } from 'veloce-ts';

globalEvents.on('order.placed', async ({ orderId, amount }) => {
  await analytics.track('purchase', { orderId, amount });
});

globalEvents.on('order.placed', async ({ orderId }) => {
  await inventory.decrementStock(orderId);
});

Usar un bus local (no global)

import { EventBus } from 'veloce-ts';

// Ámbito del bus a un plugin o módulo
export class NotificationPlugin {
  readonly events = new EventBus();

  register(app: VeloceTS) {
    // eventos internos — no contaminan globalEvents
    this.events.on('notification.queued', this.process.bind(this));
  }
}

Limpieza en tests

import { globalEvents } from 'veloce-ts';
import { afterEach } from 'bun:test';

afterEach(() => {
  globalEvents.removeAllListeners();
});

Próximos Pasos

• Streaming — combina EventBus con SSE
• Interceptores
• Filtros de Excepción