Reactive primitive in the form of Emitters

example.ts

declare global {
  interface SymbolConstructor {
    /**
     * Similar to Symbol.iterator it is used to lookup the emitter of any Emittable
     */
    readonly emitter: unique symbol;
  }

  /**
   * Reactive primitive used for propagating messages through graph connections
   */
  interface Emitter<TEmitInput, TEmitOutput = TEmitInput> {
    /**
     * Subscribes to emitted output messages after propagation to consumers (runs after full graph propagation)
     */
    subscribe(
      handler: (message: TEmitOutput) => void,
      registerDisposer: (disposer: () => void) => void
    ): void;
    /**
     * Convenience self reference so Emitter can be used as Emittables
     */
    readonly [Symbol.emitter]: this;
  }

  interface Emittable<TEmitInput, TEmitOutput = TEmitInput> {
    readonly [Symbol.emitter]: Emitter<TEmitInput, TEmitOutput>;
  }

  /**
   * Internal operators used to control the emitter, you can think of them as being similar to the internal resolve, reject functions passed to a Promise constructor
   */
  interface EmitterOperator<TEmitInput, TEmitOutput> {
    /**
     * Emits message to all consumers and subscriptions
     */
    emit(message: TEmitOutput): void;
    /**
     * Creates a connection between the consumer emitter (this) and a source emitter
     */
    connectSource(source: Emitter<unknown, TEmitInput>): void;
    /**
     * Removes connection if one exists between the consumer emitter and a source emitter
     */
    disconnectSource(source: Emitter<unknown, TEmitInput>): void;
    /**
     * Removes all source connections from the emitter
     */
    clearSourceConnections(): void;
    /**
     * Retrieve all source emitters
     */
    getSources(): Emitter<unknown, TEmitInput>[];
  }

  interface EmitterOptions<TEmitInput, TEmitOutput> {
    /**
     * Receives all messages passed to the Emitter before being transformed or sent to consumers
     */
    receive?: (message: TEmitInput) => void;
    /**
     * Transforms a message before passing to consumers
     */
    transform?: (message: TEmitInput) => TEmitOutput;
  }

  interface EmitterConstructorOptions<TEmitInput, TEmitOutput>
    extends EmitterOptions<TEmitInput, TEmitOutput> {
    /**
     * Runs at creation
     */
    setup?: (operator: EmitterOperator<TEmitInput, TEmitOutput>) => void;
  }

  interface EmitterConstructor {
    new <TEmitInput, TEmitOutput = TEmitInput>(
      options?: EmitterConstructorOptions<TEmitInput, TEmitOutput>
    ): Emitter<TEmitInput, TEmitOutput>;
    /**
     * Similar to Promise.withResolvers
     */
    withOperators<TEmitInput, TEmitOutput = TEmitInput>(
      options?: EmitterOptions<TEmitInput, TEmitOutput>
    ): {
      emitter: Emitter<TEmitInput, TEmitOutput>;
    } & EmitterOperator<TEmitInput, TEmitOutput>;
  }

  const Emitter: EmitterConstructor;
}

/**
 * # Reactive library A
 * - Uses a globally managed async context for auto tracking dependencies.
 * - This is similar to the vast majority of reactive libs, except for the fact
 *   it's using AsyncContext to allow for async auto tracking.
 * - Combined API for reads and writes with `ref.value`, similar to Vue, and Starbeam (Cell.current).
 */

declare global {
  // Including required AsyncContext types taken from https://github.com/tc39/proposal-async-context#proposed-solution
  namespace AsyncContext {
    class Variable<T> {
      constructor(options: AsyncVariableOptions<T>);
      get name(): string;
      run<R>(value: T, fn: (...args: any[]) => R, ...args: any[]): R;
      get(): T | undefined;
    }

    interface AsyncVariableOptions<T> {
      name?: string;
      defaultValue?: T;
    }

    class Snapshot {
      constructor();
      run<R>(fn: (...args: any[]) => R, ...args: any[]): R;
    }
  }
}

interface ARef<T> extends Emittable<void> {
  value: T;
}

export const aConnectContext = new AsyncContext.Variable<
  (emitter: Emitter<any>) => void
>({
  name: 'Emitter Connection Context',
});

export function createARef<T>(initialValue: T): ARef<T> {
  let storedValue = initialValue;
  const { emitter, emit } = Emitter.withOperators<void>();

  // Alternatively:
  // let emit;
  // const emitter = new Emitter<void>({
  //   setup(operator) {
  //     emit = operator.emit;
  //   }
  // });

  return {
    set value(freshValue) {
      if (freshValue === storedValue) {
        // Avoid update if value not changed
        return;
      }
      storedValue = freshValue;
      emit();
    },
    get value() {
      aConnectContext.get()?.(emitter);
      return storedValue;
    },
    [Symbol.emitter]: emitter,
  };
}

interface AComputed<T> extends Emittable<void> {
  readonly value: T;
}

const emptyCacheToken = Symbol();

export function createAComputed<T>(computer: () => T): AComputed<T> {
  let cachedValue: T | typeof emptyCacheToken = emptyCacheToken;

  const { emitter, connectSource } = Emitter.withOperators<void>({
    receive() {
      cachedValue = emptyCacheToken;
    },
  });

  return {
    get value() {
      aConnectContext.get()?.(emitter);
      if (cachedValue === emptyCacheToken) {
        cachedValue = aConnectContext.run(connectSource, computer);
      }
      return cachedValue;
    },
    [Symbol.emitter]: emitter,
  };
}

function aConnect(emittable: Emittable<any>) {
  aConnectContext.get()?.(emittable[Symbol.emitter]);
}

/**
 * # Reactive library B
 * - Uses locally scoped connection contexts which also allow for creating connections asynchronously,
 *   without need of AsyncContext. This is the approach Metron uses internally.
 * - Split API for reads and writes similar to React, Solid, and Metron.
 */

const valueKey = Symbol();

interface BAtom<T> extends Emittable<void> {
  readonly [valueKey]: T;
}

export function createBAtom<T>(
  initialValue: T
): [atom: BAtom<T>, setAtom: (value: T) => void] {
  let storedValue = initialValue;
  const { emitter, emit } = Emitter.withOperators<void>();

  function set(freshValue: T) {
    if (freshValue === storedValue) {
      // Avoid update if value not changed
      return;
    }
    storedValue = freshValue;
    emit();
  }

  return [
    {
      get [valueKey]() {
        return storedValue;
      },
      [Symbol.emitter]: emitter,
    },
    set,
  ];
}

interface BConnectContext {
  read<T>(atom: BAtom<T>): T;
  connect(emittable: Emittable<any>): void;
}

export function createBComputed<T>(
  computer: (context: BConnectContext) => T
): BAtom<T> {
  let cachedValue: T | typeof emptyCacheToken = emptyCacheToken;

  const { emitter, connectSource } = Emitter.withOperators<void>({
    receive() {
      cachedValue = emptyCacheToken;
    },
  });

  const localConnectContext: BConnectContext = {
    read(atom) {
      connectSource(atom[Symbol.emitter]);
      return atom[valueKey];
    },
    connect(emittable) {
      connectSource(emittable[Symbol.emitter]);
    },
  };

  return {
    get [valueKey]() {
      aConnectContext.get()?.(emitter);
      if (cachedValue === emptyCacheToken) {
        cachedValue = computer(localConnectContext);
      }
      return cachedValue;
    },
    [Symbol.emitter]: emitter,
  };
}

export function bUntracked<T>(atom: BAtom<T>) {
  return atom[valueKey];
}

/**
 * Libraries have almost seamless interop, thanks to the shared primitive.
 * Despite the libraries using completely different data access patterns, and connection mechanisms.
 */

const foo = createARef('Foo');
const [bar, setBar] = createBAtom('Bar');

const foobar = createAComputed(() => {
  aConnect(bar);
  return foo.value + bUntracked(bar);
});

const barfoo = createBComputed(({ read, connect }) => {
  connect(foo);
  return read(bar) + foo.value;
});

bUntracked(barfoo); // "BarFoo"
foobar.value; // "FooBar"

setBar('World!');

bUntracked(barfoo); // "World!Foo"
foobar.value; // "FooWorld!"

foo.value = 'Hello ';

bUntracked(barfoo); // "World!Hello "
foobar.value; // "Hello World!"