Typed Observables for EventBus Events

Hello again! Welcome back to our project.

In the last lesson, we successfully implemented the core on and emit methods for our EventBus, creating a functional publisher-subscriber system. We explored two potential architectures: one managing multiple sets of listeners, and another, more reactive-friendly pattern using a single stream of events shaped by a discriminated union.

Today, we will pivot to that second architecture to achieve this lesson's goal: to implement a method on the EventBus that returns a correctly typed observable for a specific event. This will transform our EventBus from a simple pub/sub utility into a powerful, reactive data source for our framework, seamlessly integrating it with the Observable and Subject classes we've already built.

1. Refactoring the EventBus for Reactivity

Our previous implementation used a listeners object to store a Set of callbacks for each event. While functional, this design isn't ideal for creating observables. A much cleaner approach is to have the EventBus manage a single, internal Subject. All events are pushed onto this Subject, and consumer methods can then filter this single stream to create observables for specific events.

Let's refactor our EventBus to use this pattern. We'll be using the Subject, Observable, filter, and map components we developed in previous modules.

First, let's define a helper type. This discriminated union is the cornerstone of the pattern. It transforms our EventMap into a union of all possible event objects (e.g., { type: 'user:login', payload: ... } | { type: 'notification:send', payload: ... }). This structure is what allows TypeScript's type narrowing to work within our observable pipeline.

// src/EventBus.ts

import { Subject } from './Subject'; // Assuming from Module 3
import { Observable } from './Observable'; // Assuming from Module 2
import { filter, map } from './operators'; // Assuming from Module 4

// A helper type to create a discriminated union of all possible event objects.
// We'll make it generic to work with our EventBus's EventMap.
type BusEvent<EventMap extends Record<string, any>> = {
  [K in keyof EventMap]: {
    type: K;
    payload: EventMap[K];
  };
}[keyof EventMap];

Now, we can update the EventBus class itself. We'll replace the listeners property and on method with a private Subject. The emit method will be updated to push event objects onto this subject.

// src/EventBus.ts (continued)

export class EventBus<
  // We'll simplify the constraint for this example
  EventMap extends Record<string, any>
> {
  // The private subject that will broadcast all events.
  // It is typed with our new discriminated union.
  private subject = new Subject<BusEvent<EventMap>>();

  // The 'emit' method now creates an event object and pushes it to the subject.
  emit<K extends keyof EventMap>(event: K, payload: EventMap[K]): void {
    // We assert the type here because TypeScript can't quite follow the
    // correlation between K and the payload on its own in this context.
    this.subject.next({ type: event, payload } as BusEvent<EventMap>);
  }

  // We will implement the 'observe' method next.
}

Note: We've removed the on method entirely, as its functionality will now be handled by subscribing to the observable returned by our new method.

2. Implementing the observe Method

This is the core of today's lesson. The observe method will take an event name and return an Observable that emits only the payloads for that specific event. We will achieve this by piping the Subject's stream through the filter and map operators.

The following resource provides an excellent walkthrough of this exact pattern.

As detailed in the article, the implementation involves two key steps within an Rx-style .pipe():

1. filter: We filter the stream of all events to only allow those whose type matches the requested event name. We use a type predicate ((event): event is ...) to inform TypeScript that any event that passes this filter is of a more specific type. This is what enables type safety in the next step.
2. map: Once the stream is filtered, we know every event object is of the correct type. We then use map to simply extract the payload property from each event object.

Let's add this observe method to our EventBus class.

// src/EventBus.ts (with the new method)

export class EventBus<EventMap extends Record<string, any>> {
  private subject = new Subject<BusEvent<EventMap>>();

  emit<K extends keyof EventMap>(event: K, payload: EventMap[K]): void {
    this.subject.next({ type: event, payload } as BusEvent<EventMap>);
  }

  /**
   * Returns a type-safe observable for a specific event.
   * @param event The name of the event to listen for.
   * @returns An Observable that emits the payload of the specified event.
   */
  observe<K extends keyof EventMap>(event: K): Observable<EventMap[K]> {
    // We start with the subject, but expose it as a plain Observable
    // to prevent consumers from calling .next() on it.
    return this.subject.asObservable().pipe(
      // 1. Filter the stream for the specific event type.
      // The type predicate is essential for type safety downstream.
      filter((e): e is Extract<BusEvent<EventMap>, { type: K }> => e.type === event),
      
      // 2. Map the filtered stream to just the payload.
      map(e => e.payload)
    );
  }
}

With this, our EventBus is now a fully-fledged, type-safe, reactive event source.

3. Usage Example

Let's see our refactored EventBus in action.

// example.ts
import { EventBus } from './EventBus';

// 1. Define the event map (using 'void' for events with no payload)
interface AppEvents {
  'user:login': { id: string; name: string };
  'user:logout': void; 
  'notification:send': { message: string; severity: 'info' | 'error' };
}

// 2. Instantiate the event bus
const bus = new EventBus<AppEvents>();

// 3. Create an observable for a specific event
const login$ = bus.observe('user:login');

// 4. Subscribe to the observable
const subscription = login$.subscribe({
  next: (user) => {
    // 'user' is correctly and automatically typed as { id: string; name: string }
    console.log(`User logged in: ${user.name} (ID: ${user.id})`);
  }
});

const notification$ = bus.observe('notification:send');
notification$.subscribe({
  next: (notification) => {
    // 'notification' is typed as { message: string; severity: 'info' | 'error' }
    console.log(`[${notification.severity.toUpperCase()}]: ${notification.message}`);
  }
});

// 5. Emit events, which will be received by the subscribers
bus.emit('user:login', { id: 'usr_123', name: 'Alice' });
bus.emit('notification:send', { message: 'Login successful', severity: 'info' });
bus.emit('user:logout', undefined); // The payload for 'user:logout' is void

// Unsubscribe when done
subscription.unsubscribe();

As you can see, the developer experience is excellent. The observe method returns an Observable where the emitted value is already correctly typed to the event's payload, with no manual type assertions needed by the consumer.

Conclusion

In this lesson, you have successfully integrated your EventBus with your reactive primitives. By refactoring the architecture to use an internal Subject and a discriminated union, you created a clean foundation for the observe method.

Key Takeaways:

• Single-Stream Architecture: Using a single Subject to broadcast all events (wrapped in a { type, payload } object) is a powerful pattern for reactive event buses.
• Discriminated Unions: This TypeScript feature is the key to making the single-stream pattern type-safe, allowing operators like filter to intelligently narrow the type of the stream.
• Observable Factory: The observe method acts as a factory, taking an event name and returning a new, correctly typed Observable derived from the main stream using filter and map.

Our EventBus is now highly capable. However, there's a subtle but important piece of resource management missing. What happens to the internal listeners inside the Subject when an observable returned by observe is unsubscribed? In the next lesson, we will address this by ensuring the EventBus automatically cleans up its internal listeners when a returned observable is unsubscribed, preventing potential memory leaks.