Refactor queue management to a service class. (#94)

Author: demiankatz

api/src/models/Job.ts

@@ -6,31 +6,31 @@ import Config from "./Config";
 import { execSync } from "child_process";
 import JobMetadata from "./JobMetadata";
 import ImageFile from "./ImageFile";
-import { Queue } from "bullmq";
+import QueueManager from "../services/QueueManager";
 
 class Job {
     dir: string;
     name: string;
     _metadata: JobMetadata = null;
     config: Config;
+    queue: QueueManager;
 
-    constructor(dir: string, config: Config) {
+    constructor(dir: string, config: Config, queue: QueueManager) {
         this.dir = dir;
         this.name = path.basename(dir);
         this.config = config;
+        this.queue = queue;
     }
 
     public static build(dir: string): Job {
-        return new Job(dir, Config.getInstance());
+        return new Job(dir, Config.getInstance(), QueueManager.getInstance());
     }
 
     async ingest(): Promise<void> {
         const lockfile = this.metadata.ingestLockfile;
         if (this.metadata.published && !fileExists(lockfile)) {
             closeSync(openSync(lockfile, "w")); // touch
-            const q = new Queue("vudl");
-            await q.add("ingest", { dir: this.dir });
-            q.close();
+            this.queue.ingestJob(this.dir);
         }
     }
 
@@ -48,9 +48,7 @@ class Job {
 
         if (status.expected > status.processed && !fileExists(lockfile)) {
             closeSync(openSync(lockfile, "w")); // touch
-            const q = new Queue("vudl");
-            await q.add("derivatives", { dir: this.dir });
-            q.close();
+            this.queue.buildDerivatives(this.dir);
         }
     }
 

api/src/routes/messenger.ts

@@ -2,15 +2,13 @@ import SolrIndexer from "../services/SolrIndexer";
 import bodyParser = require("body-parser");
 import express = require("express");
 import Config from "../models/Config";
-import { Queue } from "bullmq";
+import QueueManager from "../services/QueueManager";
 import { requireToken } from "./auth";
 import { pidSanitizer } from "./sanitize";
 const messenger = express.Router();
 
 messenger.post("/pdfgenerator/:pid", pidSanitizer, requireToken, async function (req, res) {
-    const q = new Queue("vudl");
-    await q.add("generatepdf", { pid: req.params.pid });
-    q.close();
+    QueueManager.getInstance().generatePdf(req.params.pid);
     res.status(200).send("ok");
 });
 
@@ -38,27 +36,6 @@ messenger.post("/solrindex/:pid", pidSanitizer, requireToken, async function (re
     }
 });
 
-async function queueIndexOperation(pid, action): Promise<void> {
-    // Fedora often fires many change events about the same object in rapid succession;
-    // we don't want to index more times than we have to, so let's not re-queue anything
-    // that is already awaiting indexing.
-    const q = new Queue("vudl");
-    const jobs = await q.getJobs("wait");
-    let lastPidAction = null;
-    for (let i = 0; i < jobs.length; i++) {
-        if (jobs[i].name === "index" && jobs[i].data.pid === pid) {
-            lastPidAction = jobs[i].data.action;
-            break;
-        }
-    }
-    if (action === lastPidAction) {
-        console.log("Skipping queue; " + pid + " is already awaiting " + action + ".");
-    } else {
-        await q.add("index", { pid: pid, action: action });
-    }
-    q.close();
-}
-
 messenger.post("/camel", bodyParser.json(), async function (req, res) {
     const fedoraBase = Config.getInstance().restBaseUrl;
     const idParts = req?.body?.id.replace(fedoraBase, "").split("/");
@@ -84,10 +61,10 @@ messenger.post("/camel", bodyParser.json(), async function (req, res) {
     switch (action) {
         case "Create":
         case "Update":
-            await queueIndexOperation(pid, "index");
+            await QueueManager.getInstance().performIndexOperation(pid, "index");
             break;
         case "Delete":
-            await queueIndexOperation(pid, "delete");
+            await QueueManager.getInstance().performIndexOperation(pid, "delete");
             break;
         default: {
             const msg = "Unexpected action: " + action + " (on PID: " + pid + ")";

api/src/services/QueueManager.ts

@@ -0,0 +1,54 @@
+import { Queue } from "bullmq";
+
+class QueueManager {
+    private static instance: QueueManager;
+    protected defaultQueueName = "vudl";
+
+    public static getInstance(): QueueManager {
+        if (!QueueManager.instance) {
+            QueueManager.instance = new QueueManager();
+        }
+        return QueueManager.instance;
+    }
+
+    protected async addToQueue(jobName: string, data: Record<string, string>, queueName: string = null): Promise<void> {
+        const q = new Queue(queueName ?? this.defaultQueueName);
+        await q.add(jobName, data);
+        q.close();
+    }
+
+    public async buildDerivatives(dir: string): Promise<void> {
+        return await this.addToQueue("derivatives", { dir });
+    }
+
+    public async generatePdf(pid: string): Promise<void> {
+        return await this.addToQueue("generatepdf", { pid });
+    }
+
+    public async ingestJob(dir: string): Promise<void> {
+        return await this.addToQueue("ingest", { dir });
+    }
+
+    public async performIndexOperation(pid: string, action: string): Promise<void> {
+        // Fedora often fires many change events about the same object in rapid succession;
+        // we don't want to index more times than we have to, so let's not re-queue anything
+        // that is already awaiting indexing.
+        const q = new Queue(this.defaultQueueName);
+        const jobs = await q.getJobs("wait");
+        let lastPidAction = null;
+        for (let i = 0; i < jobs.length; i++) {
+            if (jobs[i].name === "index" && jobs[i].data.pid === pid) {
+                lastPidAction = jobs[i].data.action;
+                break;
+            }
+        }
+        if (action === lastPidAction) {
+            console.log("Skipping queue; " + pid + " is already awaiting " + action + ".");
+        } else {
+            await q.add("index", { pid, action });
+        }
+        q.close();
+    }
+}
+
+export default QueueManager;