Observable Event Pattern Example

Thing.ts

import { Subject } from 'rxjs';

/**
 * Event declarations.
 */
export type ThingEvent = ThingEventFoo | ThingEventBar; // Type merging.

export type ThingEventFoo = {
  type: 'FOO';
  payload: {
    eid: string | number; // event-id (example)
    stage: 'START' | 'DONE';
    result?: any;
  };
};

export type ThingEventBar = {
  type: 'BAR';
  payload: {
    eid: string | number; // event-id (example)
    // etc.
  };
};

/**
 * Example class.
 */
export class Thing {
  private _dispose$ = new Subject();
  private _events$ = new Subject<ThingEvent>();

  /**
   * All events, single observable.
   *  - This is a "shared" (multi-plex) observable version of the underlying subject.
   *  - The .share() method casts the Subject to an Observable.
   *  - Runs until the dispose observable fires.
   *  - Uses the "$" suffix naming convention to call itself out as an Observable.
   */
  public readonly events$ = this._events$.takeUntil(this._dispose$).share(); 

  /**
   * Example methods.
   */
  public foo(args: { delay: number }) {
    idCounter++;
    const eid = idCounter; // NB: Would be smarter than a stupid counter increment :)
    const stop$ = new Subject();

    // Create a subset of the master events observable scoped to just this method invocation.
    const events$ = this.events$ // Because this comes of the common `events$` this is automatically dispose-safe.
      .takeUntil(stop$) // Cancellable.
      .filter(e => e.payload.type === 'FOO') // Filter out any other kind of events (NB: redendant because of ID below but making the point :) )
      .filter(e => e.payload.eid === eid) // Filter on this specific event id.
      .map(e => e.payload as ThingEventFoo['payload']); // Scope this into the actual data object for the event.

    this._events$.next({ type: 'FOO', payload: { eid, stage: 'START' } });

    // Dummy implementation.
    setTimeout(() => {
      this._events$.next({
        type: 'FOO',
        payload: { eid, stage: 'DONE', result: 'some result data' },
      });
    }, args.delay);

    return {
      events$, // Just the "foo" method invokcation events.
      cancel: () => stop$.next(),
    };
  }

  public bar() {
    // Same sorta thing...
  }

  /**
   * Causes all deriving observable to stop.
   */
  public dispose() {
    this._dispose$.next();
  }
}

let idCounter = 0;