Implement keyed priority queue

src/index.ts

@@ -1,7 +1,8 @@
 import Plugin from '@swup/plugin';
 import { getCurrentUrl, Handler, Location } from 'swup';
 import type { DelegateEvent, DelegateEventHandler, DelegateEventUnsubscribe, PageData } from 'swup';
-import createQueue, { Queue } from './queue.js';
+
+import Queue from './queue.js';
 
 declare module 'swup' {
 	export interface Swup {
@@ -73,8 +74,7 @@ export default class SwupPreloadPlugin extends Plugin {
 
 	options: PluginOptions;
 
-	protected queue: Queue;
-	protected preloadPromises = new Map<string, Promise<PageData | void>>();
+	protected preloadQueue: Queue<PageData | void>;
 	protected preloadObserver?: { stop: () => void; update: () => void };
 
 	protected mouseEnterDelegate?: DelegateEventUnsubscribe;
@@ -101,8 +101,8 @@ export default class SwupPreloadPlugin extends Plugin {
 		// Bind public methods
 		this.preload = this.preload.bind(this);
 
-		// Create global priority queue
-		this.queue = createQueue(this.options.throttle);
+		// Create priority queue
+		this.preloadQueue = new Queue<PageData | void>(this.options.throttle);
 	}
 
 	mount() {
@@ -152,7 +152,7 @@ export default class SwupPreloadPlugin extends Plugin {
 		this.swup.preload = undefined;
 		this.swup.preloadLinks = undefined;
 
-		this.preloadPromises.clear();
+		this.preloadQueue.clear();
 
 		this.mouseEnterDelegate?.destroy();
 		this.touchStartDelegate?.destroy();
@@ -166,8 +166,8 @@ export default class SwupPreloadPlugin extends Plugin {
 	 */
 	protected onPageLoad: Handler<'page:load'> = (visit, args, defaultHandler) => {
 		const { url } = visit.to;
-		if (url && this.preloadPromises.has(url)) {
-			return this.preloadPromises.get(url);
+		if (url && this.preloadQueue.has(url)) {
+			return this.preloadQueue.get(url);
 		}
 		return defaultHandler?.(visit, args);
 	};
@@ -255,8 +255,8 @@ export default class SwupPreloadPlugin extends Plugin {
 		}
 
 		// Already preloading? Return existing promise
-		if (this.preloadPromises.has(url)) {
-			return this.preloadPromises.get(url);
+		if (this.preloadQueue.has(url)) {
+			return this.preloadQueue.get(url);
 		}
 
 		// Should we preload?
@@ -266,21 +266,7 @@ export default class SwupPreloadPlugin extends Plugin {
 
 		// Queue the preload with either low or high priority
 		// The actual preload will happen when a spot in the queue is available
-		const queuedPromise = new Promise<PageData | void>((resolve) => {
-			this.queue.add(() => {
-				this.performPreload(url)
-					.catch(() => {})
-					.then((page) => resolve(page))
-					.finally(() => {
-						this.queue.next();
-						this.preloadPromises.delete(url);
-					});
-			}, priority);
-		});
-
-		this.preloadPromises.set(url, queuedPromise);
-
-		return queuedPromise;
+		return this.preloadQueue.add(url, () => this.performPreload(url), priority);
 	}
 
 	/**
@@ -380,7 +366,7 @@ export default class SwupPreloadPlugin extends Plugin {
 		// Already in cache?
 		if (this.swup.cache.has(url)) return false;
 		// Already preloading?
-		if (this.preloadPromises.has(url)) return false;
+		if (this.preloadQueue.has(url)) return false;
 		// Should be ignored anyway?
 		if (this.swup.shouldIgnoreVisit(href, { el })) return false;
 		// Special condition for links: points to current page?
@@ -392,10 +378,14 @@ export default class SwupPreloadPlugin extends Plugin {
 	/**
 	 * Perform the actual preload fetch and trigger the preload hook.
 	 */
-	protected async performPreload(url: string): Promise<PageData> {
-		const page = await this.swup.fetchPage(url);
-		await this.swup.hooks.call('page:preload', { page });
-		return page;
+	protected async performPreload(url: string): Promise<PageData | void> {
+		try {
+			const page = await this.swup.fetchPage(url);
+			await this.swup.hooks.call('page:preload', { page });
+			return page;
+		} catch (error) {
+			return;
+		}
 	}
 
 	/**

src/queue.ts

@@ -1,59 +1,93 @@
-type QueueFunction = {
-	(): void;
-	__queued?: boolean;
-};
-
-export type Queue = {
-	add: (fn: QueueFunction, highPriority?: boolean) => void;
-	next: () => void;
-};
-
-export default function createQueue(limit: number = 1): Queue {
-	const qlow: QueueFunction[] = [];
-	const qhigh: QueueFunction[] = [];
-	let total = 0;
-	let running = 0;
-
-	function add(fn: QueueFunction, highPriority: boolean = false): void {
-		// Already added before?
-		if (fn.__queued) {
-			// Move from low to high-priority queue
-			if (highPriority) {
-				const idx = qlow.indexOf(fn);
-				if (idx >= 0) {
-					const removed = qlow.splice(idx, 1);
-					total = total - removed.length;
-				}
-			} else {
-				return;
-			}
+type QueueFunction<T> = { (): Promise<T>; };
+
+/**
+ * A priority queue that runs a limited number of jobs at a time.
+ */
+export default class Queue<T extends any = unknown> {
+	/** The number of jobs to run at a time */
+	private limit: number;
+
+	/** The queue of low-priority jobs */
+	private qlow: Map<string, Promise<T>> = new Map();
+
+	/** The queue of high-priority jobs */
+	private qhigh: Map<string, Promise<T>> = new Map();
+
+	/** The list of currently running jobs */
+	private qactive: Map<string, Promise<T>> = new Map();
+
+	constructor(limit: number = 1) {
+		this.limit = limit;
+	}
+
+	/** The total number of jobs in the queue */
+	get total(): number {
+		return this.qlow.size + this.qhigh.size;
+	}
+
+	/** Add a job to queue */
+	async add(key: string, fn: QueueFunction<T>, highPriority: boolean = false): Promise<T|void> {
+		// Short-circuit if already running
+		if (this.qactive.has(key)) {
+			return this.qactive.get(key);
+		}
+
+		if (this.qlow.has(key) && highPriority) {
+			// Promote from low to high-priority queue
+			this.qlow.delete(key);
+		} else if (this.qhigh.has(key)) {
+			// Skip if already in queue
+			return;
 		}
 
-		// Mark as processed
-		fn.__queued = true;
-		// Push to queue: high or low
-		(highPriority ? qhigh : qlow).push(fn);
-		// Increment total
-		total++;
-		// Initialize queue if first item
-		if (total <= 1) {
-			run();
+		(highPriority ? this.qhigh : this.qlow).set(key, fn);
+
+		if (this.total <= 1) {
+			this.run();
 		}
 	}
 
-	function next(): void {
-		running--; // make room for next
-		run();
+	active(key: string): boolean {
+		return this.qactive.has(key);
+	}
+
+	queued(key: string): boolean {
+		return this.qlow.has(key) || this.qhigh.has(key);
+	}
+
+	has(key: string): boolean {
+		return this.active(key) || this.queued(key);
 	}
 
-	function run(): void {
-		if (running < limit && total > 0) {
-			const fn = qhigh.shift() || qlow.shift() || (() => {});
-			fn();
-			total--;
-			running++; // is now WIP
+	clear(): void {
+		this.qlow.clear();
+		this.qhigh.clear();
+	}
+
+	/** Run the next available job */
+	protected async run(): Promise<void> {
+		if (!this.total) return;
+		if (this.qactive.size >= this.limit) return;
+
+		const next = this.next();
+		if (next) {
+			this.qactive.set(next.key);
+			this.run();
+			await next.fn();
+			this.qactive.delete(next.key);
+			this.run();
 		}
 	}
 
-	return { add, next };
+	/** Get the next available job */
+	protected next(): { key: string; fn: QueueFunction } | null {
+		return [this.qhigh, this.qlow].reduce((acc, queue) => {
+			if (!acc) {
+				const [key, fn] = queue.entries().next().value || [];
+				queue.delete(key);
+				return key ? { key, fn } : null;
+			}
+			return acc;
+		}, null as { key: string; fn: QueueFunction } | null);
+	}
 }