Cancellation Token With Linked Sources

export class CancellationError extends Error {
  constructor(message = "Operation cancelled") {
    super(message);
    this.name = "CancellationError";
  }
}

export class CancellationToken {
  #cancelled = false;
  #reason;
  #listeners = new Set();

  get isCancelled() {
    return this.#cancelled;
  }

  get reason() {
    return this.#reason;
  }

  throwIfCancelled() {
    if (!this.#cancelled) return;
    if (this.#reason instanceof Error) throw this.#reason;
    throw new CancellationError(String(this.#reason));
  }

  onCancel(listener) {
    if (typeof listener !== "function") {
      throw new TypeError("listener must be a function");
    }
    if (this.#cancelled) {
      queueMicrotask(() => listener(this.#reason));
      return () => {};
    }
    this.#listeners.add(listener);
    return () => this.#listeners.delete(listener);
  }

  _cancel(reason) {
    if (this.#cancelled) return false;
    this.#cancelled = true;
    this.#reason = reason ?? new CancellationError();
    for (const listener of [...this.#listeners]) listener(this.#reason);
    this.#listeners.clear();
    return true;
  }
}

export class CancellationTokenSource {
  #token = new CancellationToken();

  get token() {
    return this.#token;
  }

  cancel(reason) {
    return this.#token._cancel(reason);
  }

  static link(...sources) {
    const controller = new CancellationTokenSource();
    const unsubscribers = sources.map((source) =>
      source.token.onCancel((reason) => controller.cancel(reason)),
    );
    controller.token.onCancel(() => {
      for (const unsubscribe of unsubscribers) unsubscribe();
    });
    return controller;
  }
}