reader-prefetch.js

/**
 * Small priority scheduler shared by reader prefetch tasks.
 */

(function() {
  'use strict';

  if (window.MGR_READER_PREFETCH) return;

  function create(options = {}) {
    const maxConcurrent = Math.max(1, Number(options.maxConcurrent) || 2);
    const run = typeof options.run === 'function' ? options.run : async function() {};
    const shouldSkip = typeof options.shouldSkip === 'function' ? options.shouldSkip : function() { return false; };
    const onError = typeof options.onError === 'function' ? options.onError : function() {};
    const queue = [];
    const queued = new Set();
    const active = new Map();

    function pump() {
      while (active.size < maxConcurrent && queue.length > 0) {
        const key = queue.shift();
        queued.delete(key);
        if (active.has(key) || shouldSkip(key)) continue;

        const controller = new AbortController();
        active.set(key, controller);
        let task;
        try {
          task = run(key, controller.signal);
        } catch (error) {
          task = Promise.reject(error);
        }
        Promise.resolve(task)
          .catch((error) => {
            if (!controller.signal.aborted) onError(error, key);
          })
          .finally(() => {
            active.delete(key);
            pump();
          });
      }
    }

    function enqueue(keys, enqueueOptions = {}) {
      if (!Array.isArray(keys) || keys.length === 0) return;
      const additions = [];
      for (const key of keys) {
        if (queued.has(key) || active.has(key) || shouldSkip(key)) continue;
        queued.add(key);
        additions.push(key);
      }

      if (enqueueOptions.priority === true) {
        queue.unshift(...additions);
      } else {
        queue.push(...additions);
      }
      pump();
    }

    function cancelExcept(keep) {
      const predicate = typeof keep === 'function' ? keep : function(key) { return key === keep; };
      for (let index = queue.length - 1; index >= 0; index--) {
        if (predicate(queue[index])) continue;
        queued.delete(queue[index]);
        queue.splice(index, 1);
      }
      active.forEach((controller, key) => {
        if (!predicate(key)) controller.abort('prefetch-cancel');
      });
    }

    function clear() {
      queue.length = 0;
      queued.clear();
      active.forEach((controller) => controller.abort('prefetch-clear'));
    }

    function snapshot() {
      return {
        queued: queue.slice(),
        active: Array.from(active.keys()),
        maxConcurrent
      };
    }

    return Object.freeze({ enqueue, cancelExcept, clear, snapshot });
  }

  window.MGR_READER_PREFETCH = Object.freeze({ create });
})();