Channel Manager

channelManager.ts

/**
 * This is a small library I wrote when I was doing R&D work and needed a way to communicate
 * between an iFrame on the same domain and its parent tab. The existing browser API kinda sucked
 * and had a lot of issues, and it wasn't particularly enjoyable to use. So I made this small library to solve that.
 * 
 * The library allows you to communicate using *channels*, which are just streams of events with a given name.
 * You can subscribe to events of a particular type. Each event type has its own event queue, and each subscriber
 * must subscribe to a particular event type. This keeps things simple and fast.
 *
 * Events are buffered and sent asychronously. There are two ways to send events: firing and blocking.
 * Firing is a 'fire and forget' send, and blocking returns a promise that resolves when the event is received.
 * This works just like how channel messaging works in Elixir or golang.
 *
 * A *channel manager* handles creating channels and ensuring that the same channel object / scope is used
 * both in the iFrame and the parent document.
*/

/** 
 * An event is a redux-style event.
 */
interface BaseEvent<T> {
  type: T;
}

/**
 * An event can have an optional payload.
 */
interface EventWithPayload<T, P> extends BaseEvent<T> {
  payload: P;
}

export type Event<T, P = never> = BaseEvent<T> | EventWithPayload<T, P>;

/**
 * A subscriber is a function subscribed to a particular event stream. The function
 * is called whenever there is a new event on the event stream.
 */
type Subscriber = (payload?: any) => void;

/**
 * A subscriber table is a map between event types and the callbacks subscribed to them.
 * It also contains a buffer of events left to process (`toSend`). One of these is created
 * for each channel.
 */
type SubscriberTable = Map<
  string,
  {
    toSend: (undefined | unknown)[];
    subscribers: Set<Subscriber>;
  }
>;

/**
 * A channel table maps channel names to their subscriber tables.
 * Maps all the way down!
 */
type ChannelTable = Map<string, SubscriberTable>;

/**
 * Utility type to pull the payload off an event (if the event has a payload).
 */
type Payload<
  E extends BaseEvent<string>,
  N extends string = E["type"]
> = E extends EventWithPayload<N, infer P> ? P : never;

type BaseListener = () => void;
/**
 * Type that will resolve to either a function that accepts the payload `P`, or
 * if the payload is `never`, resolves to `never`.
 */
type PayloadListener<P> = P extends unknown ? (payload: P) => void : never;
/**
 * A more concrete `Subscriber` function that accepts a single argument `P` if `P` is defined,
 * or otherwise accepts no arguments.
 * 
 * In TypeScript, `T | never` is the same as `T`. `never` is ignored in unions. This allows us to make
 * a type that only has an argument if the payload is not `never`.
 */
type Listener<P> = BaseListener | PayloadListener<P>;

/**
 * Gets the event queue and subscribers from the subscriber table.
 *
 * @param subTable The subscriber table to pull from.
 * @param type The type of the event.
 * @returns The event queue and subscriber table for the event type.
 */
const getSubItem = (subTable: SubscriberTable, type: string) => {
  let item = subTable.get(type);

  if (item === undefined) {
    item = {
      toSend: [],
      subscribers: new Set(),
    };
    subTable.set(type, item);
  }

  return item;
};

/**
 * The type of the actual subscribe function. Given an event `E` and a
 * type `T`, subscribes the `listener` to be called when events are sent
 * with a type `T`.
 */
type Subscribe<
  E extends Event<string, any>,
  T extends E["type"] = E["type"]
> = (type: T, listener: Listener<Payload<E, T>>) => () => void;

/**
 * Simple event loop that processes new events of a given type.
 * Calls all of the listeners of a given `type`, passing all new events that have been added
 * to the queue since they were last called. If all return successfully, the event queue is emptied.
 * @param subTable The subcriber table.
 * @param type The event type to process listeners for.
 */
const triggerListeners = async (subTable: SubscriberTable, type: string) => {
  const item = getSubItem(subTable, type);
  if (item.subscribers.size !== 0 && item.toSend.length > 0) {
    let length = item.toSend.length;
    item.toSend.forEach((message) => {
      item.subscribers.forEach((listener) => listener(message));
    });
    if (item.toSend.length === length) item.toSend = [];
  }
};

/**
 * Given a subTable, returns the subscribe function for the given subTable. Must be passed
 * a valid Event type through the generic, or no payload listeners will be usable.
 *
 * @param subTable The subscriber table.
 */
const createListen = <E extends Event<string, any>>(
  subTable: SubscriberTable
) => {
  /**
   * Subscribes to events of a given `type` on the channel. When an event of `type` is sent on the channel,
   * the channel will execute the `listener`. Listeners are executed asychronously as events are sent asynchronously.
   *
   * If the given event `type` has a payload type attached to it, the listener will receive the payload as its first argument.
   * If there is no payload type on the `type`, the listener will be passed no arguments.
   *
   * Note that if the event type of the channel is not set, Typescript will not allow you to have a listener that
   * accepts in a payload as the payload type will be `unknown`.
   *
   * ## Buffering
   * If there are events buffered on the channel of the specific type, the listener will be immediately executed with
   * the buffered events. The events will then be removed from the buffer.
   *
   * Buffering exists to allow a grace period for listeners to join sightly after events have been fired into the channel.
   * This prevents several common race conditions (eg. a sender waiting on a return event might wait indefinitely if buffering
   * did not exist.)
   *
   * @param type A specific event type to listen to (must be a single string).
   * @param listener The listener that will be executed asynchronously when an event of `type` is sent on the channel.
   * @returns An `unsubscribe` function. When run, it will stop the listener from receiving any further events.
   */
  const listen = <T extends E["type"]>(
    type: T,
    listener: Listener<Payload<E, T>>
  ) => {
    const item = getSubItem(subTable, type);

    if (item.subscribers.size === 0 && item.toSend.length > 0) {
      triggerListeners(subTable, type);
    }

    item.subscribers.add(listener);

    return () => {
      getSubItem(subTable, type).subscribers.delete(listener);
    };
  };

  return listen;
};

type Fire<E extends Event<string, any>> = (event: E) => void;

/**
 * Creates the fire function for the channel.
 * @param subTable The subscriber table.
 * @returns The fire function.
 */
const createFire = <E extends Event<string, any>>(
  subTable: SubscriberTable
) => {
  /**
   * Fires an event into the channel and immediately resumes execution. This is the 'fire and forget' way of sending.
   *
   * If there are no listeners on a given event type, the event will be **buffered**. A buffered event sticks around indefinitely
   * in a processing queue. When a new listener is created for the event type, an asynchronous execution of each event in the buffering
   * queue will be triggered, just as if the event had been fired immediately.
   *
   * Events are fired as an object. The object must have a 'type' string that tells the channel which event queue to send
   * the event to. If the specific event type has a payload, the payload will be required.
   *
   * Events are processed asynchronously. This ensures that the browser can still do work if it needs to.
   * @param event
   */
  const fire: Fire<E> = (event) => {
    const item = getSubItem(subTable, event.type);
    item.toSend.push((event as any).payload || undefined);
    if (item.subscribers.size !== 0) triggerListeners(subTable, event.type);
  };

  return fire;
};

type VoidNever<P> = P extends unknown ? P : void;

type Block<E extends Event<string, any>, T extends E["type"]> = (
  type: T
) => Promise<VoidNever<Payload<E, T>>>;

const createBlock = <E extends Event<string, any>>({
  listen,
}: {
  listen: Subscribe<E>;
}) => {
  /**
   * Returns a promise that resolves when an event of the `type` is received.
   * Subsequent events of `type` will not be received as the promise is resolved.
   *
   * @param type The type of event to listen for.
   * @returns A promise that resolves when a single event of `type` is received.
   */
  const block = <N extends E["type"]>(
    type: N
  ): Promise<VoidNever<Payload<E, N>>> => {
    return new Promise((resolve, _) => {
      let unsubscribe = () => {};
      unsubscribe = listen(type, ((payload: VoidNever<Payload<E, N>>) => {
        if (payload !== undefined) resolve(payload);
        else (resolve as any)();
        unsubscribe();
      }) as any);
    });
  };

  return block;
};

type BaseFilter = () => true;
type PayloadFilter<P> = P extends unknown ? (payload: P) => void : never;
type Filter<P> = BaseFilter | PayloadFilter<P>;

const createBlockUntil = <E extends Event<string, any>>({
  block,
}: {
  block: Block<E, string>;
}) => {
  const baseBlock = async <T extends E["type"]>(
    type: T,
    filter: Filter<Payload<E, T>>
  ): ReturnType<Block<E, T>> => {
    while (true) {
      const event = await block(type);
      if (filter(event as any)) {
        return event as any;
      }
    }
  };

  const timeoutPromise = (timeout: number) =>
    new Promise((_, reject) => setTimeout(() => reject(), timeout));

  /**
   * Creates a promise that returns when an event is received that passes the given `filter`.
   *
   * For every event that is received of a given type, the `filter` is called with the payload passed in.
   * If the filter returns false, the promise continues to block. If the filter returns true, the event is resolved
   * from the promise.
   *
   * An optional `timeout` can be set. If an event is not received that passes the timeout, an error will be thrown.
   *
   * @param type The type of event to watch.
   * @param filter Function that is called for every event. Should return `true` if blocking should stop, else `false` if blocking should continue.
   * @param timeout A timeout in ms that an event should be received by.
   * @returns
   */
  const blockUntil = async <T extends E["type"]>(
    type: T,
    filter: Filter<Payload<E, T>>,
    timeout?: number
  ): ReturnType<Block<E, T>> => {
    const promise = baseBlock<T>(type, filter);
    if (timeout) {
      return Promise.race([promise, timeoutPromise(timeout)]) as any;
    } else return await promise;
  };

  return blockUntil;
};

const initChannel = <E extends Event<string, any>>(
  subTable: SubscriberTable
) => {
  const listen = createListen<E>(subTable);
  const fire = createFire<E>(subTable);
  const block = createBlock<E>({ listen: listen });
  const blockUntil = createBlockUntil<E>({ block });

  return { listen, fire, block, blockUntil };
};

/**
 * Creates a channel manager. A channel manager makes sure that all Javascript threads
 * in the browser (eg. an iFrame and the parent frame) reference the same scope automatically.
 * When the library is imported twice in these frames, it'll handle merging their scopes together
 * using `window.top` automatically so that they never become out-of-sync.
 *
 * @param domain A domain is the namespace of the global scope that the channel manager operates in.
 * Set this to something unique so that multiple channel managers created on the page don't conflict with each other.
 * @returns An object with the channel manager's API.
 */
export const createChannelManager = (domain: string) => {
  // Create the root scope if it doesn't exist.
  if (!(window as any).top.__channelManager) {
    (window as any).top.__channelManager = {};
  }
  if (
    (window as any).top.__channelManager &&
    !(window as any).top.__channelManager[domain]
  ) {
    (window as any).top.__channelManager[domain] = new Map();
  }

  // Get the initial channel table from the root scope.
  const channelTable = (window as any).top.__channelManager[
    domain
  ] as ChannelTable;

  /**
   * Creates a channel of `name` in the scope of the channel manager.
   *
   * A *channel* has events, senders, and listeners. Senders send messages on the channel using the `fire` function.
   * A listener can subscribe to events of a specific type using a listener function (using the `listen` function).
   * When a sender sends an event on a channel, every listener subscribed to that event's type will be called.
   * **Listeners are called asynchronously**. Execution of the listeners does not happen immediately to keep processing
   * broken up.
   *
   * Channels must be passed an `Event` generic type. This ensures that listening and firing functions are kept
   * type safe. If an `Event` type is not passed, Typescript might not let you do certain things.
   *
   * @param name The name of the channel
   * @returns A channel object with functions for interacting on the channel.
   */
  const createChannel = <E extends Event<string, any>>(name: string) => {
    let subTable = channelTable.get(name);
    if (subTable === undefined) {
      subTable = new Map();
      channelTable.set(name, subTable);
    }
    return initChannel<E>(subTable);
  };

  return { createChannel };
};