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 concurrently and resolves once they have all settled. emitSync() is the fire-and-forget variant for synchronous handlers — it does not await them.

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 have settled.

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

emitSync(event, payload) — fire sync

Calls handlers in registration order and returns void — it does not await them. An async handler passed to emitSync is started but not waited for, so use emit() whenever handlers do asynchronous work.

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

Error handling

Changed in v2.0.0

A handler that throws no longer prevents the other handlers from running.

Previously emit() used Promise.all, so the first rejection aborted the whole emit and any handler that had not been reached was silently skipped — one bad listener could stop the rest of your side effects.

Both emit() and emitSync() now run every listener, collect the failures, and rethrow them together as an AggregateError once all listeners have settled. once listeners are removed even when they throw.

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() still ran — only the failing listener is reported here
  if (error instanceof AggregateError) {
    for (const cause of error.errors) {
      logger.error('Event listener failed', cause);
    }
  }
}

If a failing listener should never interrupt the emitter, catch inside the handler itself:

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

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

The @On decorator

New in v3.3.0

Declarative listeners. Before this, globalEvents.on(...) was the only way to subscribe.

A class whose methods carry @On is registered with app.include(), exactly like a controller. Its instance comes from the DI container, so a listener injects the same services a controller can:

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: `Welcome, ${payload.name}!` });
  }

  @On('app.ready', { once: true })
  warmCaches() {
    // removed after the first delivery
  }
}

app.include(NotificationListeners);

Emitting is unchanged — the decorator only replaces the subscription side:

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

What the application does for you

Subscribes at compile()	Not at import time, so a listener can depend on services registered later in the bootstrap.
Unsubscribes at shutdown()	Without this, listeners bound to the process-wide globalEvents outlive the app that created them, and every app built in a test suite answers the same event.
Resolves through DI	Constructor injection with @Inject, same as a controller. One singleton instance serves every @On method on the class.
Fails loudly on a bad name	@On pointing at something that is not a method throws at compile(), naming the class and property — not silently at emit time.

One class, both roles

A controller can also listen; include() no longer stops at the first match:

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

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

Error handling is the bus’s

@On changes nothing about delivery: a throwing listener does not stop the others, and the failures come back from emit() together as an AggregateError. See Error handling above. If a listener must never interrupt the emitter, catch inside the method.

Using your own bus

@On subscribes to the bus the application is wired to — globalEvents unless you pass one. Supplying your own keeps a test suite from sharing subscriptions across 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() returns whichever bus is in use.

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