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 en paralelo (sin bloqueo). Si necesitas ejecución en serie, usa emitSync().

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 terminan.

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

emitSync(event, payload) — disparar sync

Llama a los handlers en orden de registro, uno por uno. Espera a cada handler antes de llamar al siguiente.

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

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');

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