Stateful observer for guaranteed delivery of continuations.

observer.ts

import type { Observer, Resolve } from "../types.ts";

import { shift } from "../deps.ts";

export function createObservable<T>() {
  let observers = new Map<Observer<T>, Resolve<T>>();

  return {
    notify(value: T) {
      for (let dispatch of observers.values()) {
        dispatch(value);
      }
    },
    observe(): Observer<T> {
      let events: T[] = [];
      let consumers: Resolve<T>[] = [];
      let observer = {
        *[Symbol.iterator]() {
          let event = events.pop();
          if (event) {
            return event;
          } else {
            return yield* shift<T>(function*(k) {
              consumers.unshift(k);
            })
          }
        },
        drop() {
          observers.delete(observer);
        },
      };
      observers.set(observer, (event: T) => {
        events.unshift(event);
        while (events.length > 0 && consumers.length > 0) {
          let consume = consumers.pop() as Resolve<T>;
          let event = events.pop() as T;
          consume(event);
        }
      });
      return observer;
    },
  };
}