Typescripty Stateful Observable

statefulObservable.ts

/*
let a = 1

let b = 2

const u = plus(a, mul(b,b))

const f = x => plus(a, x)

const k = f(b)

-------

const $a = Subject(1)

const $b = Subject(2)

const $_1 = $b.map(b => mul(b, b))

const $u = join($a,$_1).map((a,_1) => plus(a, _1))

const $f = $x => join($a, $x).map((a, _1) => plus(a, _1))

const k = $f.map(f => f($b))
*/

interface StatefulObservable<T> {
  subscribe(listener: Listener<T>): Unsubscriber;
  getState(): T;
}

type Listener<T> = (value: T) => void;
type Unsubscriber = () => void;
type Subscription<T> = { listener: Listener<T> };

export class Subject<T> implements StatefulObservable<T> {
  constructor(private state: T) {}
  private readonly subscriptions = new Set<Subscription<T>>();
  public publish(value: T) {
    if (value === this.state) return;
    for (const { listener } of this.subscriptions) {
      listener(value);
    }
    this.state = value;
  }
  public subscribe(listener: Listener<T>) {
    const subscription = { listener };
    this.subscriptions.add(subscription);
    return () => {
      this.subscriptions.delete(subscription);
    };
  }
  public getState() {
    return this.state;
  }
}

const emptyState = Symbol("empty");

export class StatefulObservableMap<A, B> implements StatefulObservable<B> {
  private constructor(
    private readonly source: StatefulObservable<A>,
    private readonly mapper: (sourceValue: A) => B
  ) {}
  private state: B | typeof emptyState = emptyState;
  private sourceUnsubscriber: Unsubscriber | null = null;
  private readonly subscriptions = new Set<Subscription<B>>();
  public subscribe(listener: Listener<B>) {
    const subscription = { listener };
    this.subscriptions.add(subscription);
    this.subscribeSource();
    return () => {
      this.subscriptions.delete(subscription);
      this.unsubscribeSource();
    };
  }
  getState() {
    if (this.state === emptyState) {
      this.state = this.mapper(this.source.getState());
    }
    return this.state;
  }
  private update = (value: A) => {
    const newState = this.mapper(value);
    if (newState === this.state) return;
    for (const { listener } of this.subscriptions) {
      listener(newState);
    }
this.state = newState;
  }
  private subscribeSource() {
    if (this.sourceUnsubscriber !== null) {
      this.sourceUnsubscriber = this.source.subscribe(this.update);
    }
  }
  private unsubscribeSource() {
    if (!this.subscriptions.size) {
      this.sourceUnsubscriber();
      this.sourceUnsubscriber = null;
    }
  }
}

export class StatefulObservableJoin<T extends any[]> implements StatefulObservable<T> {
  private constructor(private readonly sources: {[K in keyof T]: StatefulObservable<T[K]>}){}
  getState(){
    return this.sources.map(source => source.getState()) as T
  }
  private sourceUnsubscribers: Unsubscriber[] | null = null;
  private readonly subscriptions = new Set<Subscription<T>>();
  public subscribe(listener: Listener<T>) {
    const subscription = { listener };
    this.subscriptions.add(subscription);
    this.subscribeSource();
    return () => {
      this.subscriptions.delete(subscription);
      this.unsubscribeSource();
    };
  }
  private update = () => {
    const state = this.getState();
    for (const { listener } of this.subscriptions) {
      listener(state);
    }
  }
  private subscribeSource() {
    if (this.sourceUnsubscribers !== null) {
      this.sourceUnsubscribers = this.sources.map(source => source.subscribe(this.update));
    }
  }
  private unsubscribeSource() {
    if (!this.subscriptions.size) {
      for (const unsubscriber of this.sourceUnsubscribers) {
        unsubscriber()
      }
      this.sourceUnsubscribers = null;
    }
  }
}