 *        The function to be called on each element of the iterable.
   *
   *        Returning true from the function will stop the iteration.
   *
   * @param [yieldEvery = 50] {number|object}
   *        The number of iterations to complete before yielding back to the event
   *        loop.
   *
   * @return {boolean}
   *         Whether or not the function returned early.
   */
  async yieldingForEach(iterable, fn, yieldEvery = 50) {
    const yieldState =
      typeof yieldEvery === "number"
        ? Async.yieldState(yieldEvery)
        : yieldEvery;
    let iteration = 0;

    for (const item of iterable) {
      let result = fn(item, iteration++);
      if (typeof result !== "undefined" && typeof result.then !== "undefined") {
        // If we await result when it is not a Promise, we create an
        // automatically resolved promise, which is exactly the case that we
        // are trying to avoid.
        result = await result;
      }

      if (result === true) {
        return true;
      }

      if (yieldState.shouldYield()) {
        await Async.promiseYield();
        Async.checkAppReady();
      }
    }

    return false;
  },

  asyncQueueCaller(log) {
    return new AsyncQueueCaller(log);
  },

  asyncObserver(log, obj) {
    return new AsyncObserver(log, obj);
  },

  watchdog() {
    return new Watchdog();
  },
};

/**
 * Allows consumers to enqueue asynchronous callbacks to be called in order.
 * Typically this is used when providing a callback to a caller that doesn't
 * await on promises.
 */
class AsyncQueueCaller {
  constructor(log) {
    this._log = log;
    this._queue = Promise.resolve();
    this.QueryInterface = ChromeUtils.generateQI([
      "nsIObserver",
      "nsISupportsWeakReference",
    ]);
  }

  /**
   * /!\ Never await on another function that calls enqueueCall /!\
   *     on the same queue or we will deadlock.
   */
  enqueueCall(func) {
    this._queue = (async () => {
      await this._queue;
      try {
        return await func();
      } catch (e) {
        this._log.error(e);
        return false;
      }
    })();
  }

  promiseCallsComplete() {
    return this._queue;
  }
}

/*
 * Subclass of AsyncQueueCaller that can be used with Services.obs directly.
 * When this observe() is called, it will enqueue a call to the consumers's
 * observe().
 */
class AsyncObserver extends AsyncQueueCaller {
  constructor(obj, log) {
    super(log);
    this.obj = obj;
  }

  observe(subject, topic, data) {
    this.enqueueCall(() => this.obj.observe(subject, topic, data));
  }

  promiseObserversComplete() {
    return this.promiseCallsComplete();
  }
}

/**
 * Woof! Signals an operation to abort, either at shutdown or after a timeout.
 * The buffered engine uses this to abort long-running merges, so that they
 * don't prevent Firefox from quitting, or block future syncs.
 */
class Watchdog {
  constructor() {
    this.controller = new AbortController();
    this.timer = new Timer();

    /**
     * The reason for signaling an abort. `null` if not signaled,
     * `"timeout"` if the watchdog timer fired, or `"shutdown"` if the app is
     * is quitting.
     *
     * @type {string?}
     */
    this.abortReason = null;
  }

  /**
   * Returns the abort signal for this watchdog. This can be passed to APIs
   * that take a signal for cancellation, like `SyncedBookmarksMirror::apply`
   * or `fetch`.
   *
   * @type {AbortSignal}
   */
  get signal() {
    return this.controller.signal;
  }

  /**
   * Starts the watchdog timer, and listens for the app quitting.
   *
   * @param {number} delay
   *                 The time to wait before signaling the operation to abort.
   */
  start(delay) {
    if (!this.signal.aborted) {
      Services.obs.addObserver(this, "quit-application");
      this.timer.init(this, delay, Ci.nsITimer.TYPE_ONE_SHOT);
    }
  }

  /**
   * Stops the watchdog timer and removes any listeners. This should be called
   * after the operation finishes.
   */
  stop() {
    if (!this.signal.aborted) {
      Services.obs.removeObserver(this, "quit-application");
      this.timer.cancel();
    }
  }

  observe(subject, topic) {
    if (topic == "timer-callback") {
      this.abortReason = "timeout";
    } else if (topic == "quit-application") {
      this.abortReason = "shutdown";
    }
    this.stop();
    this.controller.abort();
  }
}
PK