Event Bus

Event Bus

Veloce-TS includes a built-in in-process event bus for decoupling modules without Redis or Kafka. Use it for cross-module communication, domain events, and feeding SSE streams.

EventBus

import { EventBus } from 'veloce-ts';

const bus = new EventBus();

Or use the pre-built singleton:

import { globalEvents } from 'veloce-ts';

API

on(event, handler) — subscribe

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

emit() calls all handlers in parallel (non-blocking). If you need serial execution, use emitSync().

once(event, handler) — subscribe once

The handler is automatically removed after the first call.

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

emit(event, payload) — fire async (parallel)

Returns a Promise<void> that resolves when all handlers complete.

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

emitSync(event, payload) — fire sync

Calls handlers in registration order, one at a time. Waits for each handler before calling the next.

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

off(event, handler) — unsubscribe

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

removeAllListeners(event?) — clear listeners

globalEvents.removeAllListeners('user.created');  // clear specific event
globalEvents.removeAllListeners();                 // clear all events

listenerCount(event) — count listeners

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

Patterns

Domain events

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 — registers at startup
import { globalEvents } from 'veloce-ts';

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

// audit.service.ts — multiple handlers for the same event
globalEvents.on('user.created', async ({ userId }) => {
  await db.insert(auditLog).values({
    event: 'user.created',
    resourceId: userId,
    timestamp: new Date(),
  });
});

Feed SSE streams via event bus

// Combine EventBus + SSE streaming
@Get('/events')
@SSE()
async* streamUserEvents(): AsyncGenerator<{ data: string; event?: string }> {
  yield { data: 'connected', 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);
  }
}

Multiple modules, one bus

// Module 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;
  }
}

// Module B — handles order events independently
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);
});

Using a local bus (not global)

import { EventBus } from 'veloce-ts';

// Scope the bus to a plugin or module
export class NotificationPlugin {
  readonly events = new EventBus();

  register(app: VeloceTS) {
    // internal-only events — not polluting globalEvents
    this.events.on('notification.queued', this.process.bind(this));
  }
}

Cleanup in tests

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

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

Next Steps

• Streaming — combine EventBus with SSE
• Interceptors
• Exception Filters