pubsub.js

import { makeStream, nullQueue } from '@endo/stream';

/**
 * @template TValue
 */
const makePubSub = () => {
  // Request pubsub async queue internals
  let { promise: tailPromise, resolve: tailResolve } = makePromiseKit();

  /**
   * @param {TValue} value
   */
  const pub = value => {
    const { resolve, promise } = makePromiseKit();
    tailResolve(Object.freeze({ value, promise }));
    tailResolve = resolve;
    // Unlike a queue, advance the read head for future subscribers.
    tailPromise = promise;
  };

  const sub = () => {
    // Capture the read head for the next published value.
    let cursor = tailPromise;
    return () => {
      const promise = cursor.then(next => next.value);
      cursor = cursor.then(next => next.promise);
      return promise;
    };
  };

  return harden({ pub, sub });
};

export const makeTopic = () => {
  const { pub, sub } = makePubSub();
  const publisher = makeStream(nullQueue, pub);
  return harden({
    publisher,
    subscribe: () => makeStream(sub(), nullQueue),
  });
};