Utilities for handling async operations

async.ts

import { LinkedList } from "./base";
import { OrderedSet } from "./OrderedCollections";

export function delay(ms: number): Promise<void> {
  return new Promise(resolve => setTimeout(resolve, ms));
}

export function nextTick(): Promise<void> {
  return new Promise(resolve => {
    queueMicrotask(() => {
      resolve();
    });
  });
}

export function animationFrameLimited<P extends unknown[], Context = any>(
  fn: (this: Context, ...args: P) => void,
  context?: Context
): (...args: P) => void {
  let lastInvocation: P | null;
  function deferred(this: Context, ...args: P) {
    if (lastInvocation == null) {
      requestAnimationFrame(() => {
        fn.call(this, ...lastInvocation!);
        lastInvocation = null;
      });
    }
    lastInvocation = args;
  }
  return context ? deferred.bind(context as Context) : deferred;
}

export function withRetries<P extends unknown[], R>(
  fn: (...args: P) => R | Promise<R>,
  intervals: number[]
): (...args: P) => Promise<R> {
  return function wrappedWithRetries(this: any, ...args: P) {
    let tries = 1;
    const retry = (): Promise<R> => {
      return Promise.resolve(fn.apply(this, args)).catch(err => {
        if (tries >= intervals.length) throw err;
        tries += 1;
        return delay(intervals[tries]).then(retry);
      });
    };
    return retry();
  };
}

export function deferAndBatch<T, Context = any>(
  fn: (this: Context, targets: Set<T>) => void,
  context?: Context
): (target?: T) => void {
  const queued = new Set<T>();
  let lastInvokeCount = 0;
  function deferred(this: Context, target?: T) {
    queued.add(target as T);
    lastInvokeCount += 1;
    const currentInvokeCount = lastInvokeCount;
    queueMicrotask(() => {
      if (currentInvokeCount !== lastInvokeCount) return;
      fn.call(this, queued);
      queued.clear();
    });
  }
  return context ? deferred.bind(context as Context) : deferred;
}

export class AsyncQueue<T> {
  pushQueueHead = new LinkedList<T>();
  pullQueueHead = new LinkedList<(item?: T) => void>();
  pushQueueTail = this.pushQueueHead;
  pullQueueTail = this.pullQueueHead;

  push(v: T): void {
    if (this.pullQueueHead !== this.pullQueueTail) {
      this.pullQueueHead = this.pullQueueHead.cdr!;
      this.pullQueueHead.car(v);
    } else {
      this.pushQueueTail.cdr = new LinkedList(v);
      this.pushQueueTail = this.pushQueueTail.cdr;
    }
  }

  pull(): Promise<T> {
    if (this.pushQueueHead !== this.pushQueueTail) {
      this.pushQueueHead = this.pushQueueHead.cdr!;
      return Promise.resolve(this.pushQueueHead.car);
    }
    return new Promise<T>(resolve => {
      this.pullQueueTail.cdr = new LinkedList(resolve as (item?: T) => void);
      this.pullQueueTail = this.pullQueueTail.cdr;
    });
  }

  peek(): T | undefined {
    if (this.pushQueueHead === this.pushQueueTail) return undefined;
    return this.pushQueueHead.cdr!.car;
  }

  flush(): void {
    while (this.pushQueueHead !== this.pushQueueTail) {
      this.pushQueueHead = this.pushQueueHead.cdr!;
    }
  }

  get size(): number {
    let n = 0;
    let curr = this.pushQueueHead;
    while (curr !== this.pushQueueTail) {
      n += 1;
      curr = curr.cdr!;
    }
    return n;
  }

  async *getServer(): AsyncIterable<T> {
    while (true) {
      yield this.pull();
    }
  }
}

export class AsyncQueueWithPriority<T> extends AsyncQueue<T> {
  // sort by highest priority first, then by pull order
  // this is needed because PQ is not a stable sort
  private pullQueue = new OrderedSet<
    [(item?: T) => void, number, number]
  >().setComparator((a, b) => b[1] - a[1] || a[2] - b[2]);
  private pullCount = 0;

  push(v: T): void {
    if (this.pullQueue.size > 0) {
      const [resolve] = this.pullQueue.pop()!;
      resolve(v);
    } else {
      this.pushQueueTail.cdr = new LinkedList(v);
      this.pushQueueTail = this.pushQueueTail.cdr;
    }
  }

  pull(): Promise<T> {
    return this.pullWithPriority(0);
  }

  pullWithPriority(priority: number): Promise<T> {
    if (this.pushQueueHead !== this.pushQueueTail) {
      this.pushQueueHead = this.pushQueueHead.cdr!;
      return Promise.resolve(this.pushQueueHead.car);
    }
    return new Promise<T>(resolve => {
      const pullOrder = this.pullCount;
      this.pullCount += 1;
      this.pullQueue.add([resolve as (item?: T) => void, priority, pullOrder]);
    });
  }
}