Refactor queue. Still only processing 1 job.

2024-08-30 13:01:42 -07:00
parent b6066f9149
commit e7226b36d7
4 changed files with 127 additions and 73 deletions
--- a/jobs/jobsUploadMedia.ts
+++ b/jobs/jobsUploadMedia.ts
@@ -45,7 +45,7 @@ export async function jobsUploadMedia(req: Request, res: Response) {

      //Check if there's a heic in the file set. If so, modify the file set.
      await ConvertHeicFiles(req.files as Express.Multer.File[]);
-
+return;
      logger.debug(
        "Creating thumbnails for newly uploaded media",
        (req.files as Express.Multer.File[]).map((f) => f.filename)
--- a/util/heicConverter.ts
+++ b/util/heicConverter.ts
@@ -12,7 +12,7 @@ import { FolderPaths } from "./serverInit.js";

 const __dirname = path.dirname(fileURLToPath(import.meta.url));

-const HeicQueue = new Queue("HEIC Queue", { connection: { host: "localhost", port: 6379 } });
+const HeicQueue = new Queue("HEIC Queue", { connection: { host: "localhost", port: 6379 , } });

 dotenv.config({
  path: path.resolve(process.cwd(), `.env.${process.env.NODE_ENV || "development"}`)
@@ -23,23 +23,25 @@ const imageMagick = gm.subClass({ imageMagick: true });
 export async function ConvertHeicFiles(files: Express.Multer.File[]) {
  const validFiles = await filterValidHeicFiles(files);

-  await Promise.all(
-    validFiles.map(async (file) => {
-      const convertedFileName = generateUniqueHeicFilename(file);
-      await HeicQueue.add(convertedFileName, { convertedFileName, file });
+const jobs = await HeicQueue.addBulk(validFiles.map(file => ({name: file.filename, data: {convertedFileName:generateUniqueHeicFilename(file) , file} })))
+logger.log("debug", `The Jobs Object ${(JSON.stringify(jobs,null,2))}`)
+  // await Promise.all(
+  //   validFiles.map(async (file) => {
+  //     const convertedFileName = generateUniqueHeicFilename(file);
+  //     await HeicQueue.add(convertedFileName, { convertedFileName, file },{removeOnComplete: true,});

-      // try {
-      //   await ConvertToJpeg(file.path, `${file.destination}/${convertedFileName}`);
-      //   logger.log("debug", `Converted ${file.filename} image to JPEG from HEIC.`);
-      //   await handleOriginalFile(file, convertedFileName);
-      //   file.filename = convertedFileName;
-      //   file.mimetype = "image/jpeg";
-      //   file.path = `${file.destination}/${convertedFileName}`;
-      // } catch (error) {
-      //   logger.log("error", `Error converting ${file.filename} image to JPEG from HEIC. ${JSON.stringify(error)}`);
-      // }
-    })
-  );
+  //     // try {
+  //     //   await ConvertToJpeg(file.path, `${file.destination}/${convertedFileName}`);
+  //     //   logger.log("debug", `Converted ${file.filename} image to JPEG from HEIC.`);
+  //     //   await handleOriginalFile(file, convertedFileName);
+  //     //   file.filename = convertedFileName;
+  //     //   file.mimetype = "image/jpeg";
+  //     //   file.path = `${file.destination}/${convertedFileName}`;
+  //     // } catch (error) {
+  //     //   logger.log("error", `Error converting ${file.filename} image to JPEG from HEIC. ${JSON.stringify(error)}`);
+  //     // }
+  //   })
+  // );
 }

 async function filterValidHeicFiles(files: Express.Multer.File[]) {
@@ -85,30 +87,63 @@ const HeicWorker = new Worker(
  async (job: Job) => {
    const { file, convertedFileName } = job.data;
    try {
+      logger.log("debug", `Attempting to Convert ${file.filename} image to JPEG from HEIC.`);
      await ConvertToJpeg(file.path, `${file.destination}/${convertedFileName}`);
      logger.log("debug", `Converted ${file.filename} image to JPEG from HEIC.`);
      await handleOriginalFile(file, convertedFileName);
      file.filename = convertedFileName;
      file.mimetype = "image/jpeg";
      file.path = `${file.destination}/${convertedFileName}`;
+      return true;
    } catch (error) {
-      logger.log("error", `Error converting ${file.filename} image to JPEG from HEIC. ${JSON.stringify(error)}`);
+      logger.log("error", `QUEUE ERROR: Error converting ${file.filename} image to JPEG from HEIC. ${JSON.stringify(error)}`);
+      return false;
    }
  },
  {
-    connection: { host: "localhost", port: 6379 }
+    connection: { host: "localhost", port: 6379 }, 
  }
 );

+HeicQueue.on('waiting', job => {
+  logger.log("debug", `Job is waiting in queue! ${job.data.convertedFileName}`);
+})
+HeicQueue.on('error', error => {
+  logger.log("error", `Queue Error! ${error}`);
+})
+
 HeicWorker.on("ready", () => {
-  console.log(`Worker Ready`);
+  logger.log('debug',`Worker Ready`);
 });
 HeicWorker.on("active", (job, prev) => {
-  console.log(`Job ${job} is now active; previous status was ${prev}`);
+  logger.log('debug',`Job ${job.id} is now active; previous status was ${prev}`);
 });
 HeicWorker.on("completed", (jobId, returnvalue) => {
-  console.log(`${jobId} has completed and returned ${returnvalue}`);
+  logger.log('debug',`${jobId.id} has completed and returned ${returnvalue}`);
 });
 HeicWorker.on("failed", (jobId, failedReason) => {
-  console.log(`${jobId} has failed with reason ${failedReason}`);
+  logger.log('error',`${jobId} has failed with reason ${failedReason}`);
 });
+HeicWorker.on('error', error => {
+  logger.log('error', `There was a queue error! ${error}`)
+})
+HeicWorker.on('stalled', error => {
+  logger.log('error', `There was a worker stall! ${error}`)
+})
+HeicWorker.on('ioredis:close', () => {
+  logger.log('error', `Redis connection closed!`)
+})
+
+// const queueEvents = new QueueEvents(  "HEIC Queue");
+
+// queueEvents.on('completed', ( jobId, returnvalue ) => {
+//   // Called every time a job is completed by any worker.
+// });
+
+// queueEvents.on('failed', (jobId, failedReason ) => {
+//   // Called whenever a job is moved to failed by any worker.
+// });
+
+// queueEvents.on('progress', (jobId, data) => {
+//   // jobId received a progress event
+// });
--- a/util/heicQueueProcessor.ts
+++ b/util/heicQueueProcessor.ts
@@ -1,45 +1,45 @@
-import { SandboxedJob } from "bullmq";
-import { logger } from "../server.js";
-import dotenv from "dotenv";
-import fs from "fs-extra";
-import path from "path";
-import { FolderPaths } from "./serverInit.js";
-import gm from "gm";
-const imageMagick = gm.subClass({ imageMagick: true });
-export default async function (job: SandboxedJob) {
-  // Do something with job
-  console.log("*** We're in the processing Queue")
-  const { file, convertedFileName } = job.data;
-  try {
-    logger.log("debug", `Attempting to convert ${file.filename} image to JPEG from HEIC.`);
-    await ConvertToJpeg(file.path, `${file.destination}/${convertedFileName}`);
-    logger.log("debug", `Converted ${file.filename} image to JPEG from HEIC.`);
-    await handleOriginalFile(file, convertedFileName);
-    file.filename = convertedFileName;
-    file.mimetype = "image/jpeg";
-    file.path = `${file.destination}/${convertedFileName}`;
-  } catch (error) {
-    logger.log("error", `Error converting ${file.filename} image to JPEG from HEIC. ${JSON.stringify(error)}`);
-  }
-}
-async function handleOriginalFile(file: Express.Multer.File, convertedFileName: string) {
-  if (process.env.KEEP_CONVERTED_ORIGINALS) {
-    await fs.ensureDir(path.join(file.destination, FolderPaths.ConvertedOriginalSubDir));
-    await fs.move(file.path, `${path.join(file.destination, FolderPaths.ConvertedOriginalSubDir)}/${file.filename}`);
-  } else {
-    await fs.unlink(file.path);
-  }
-}
+// import { SandboxedJob } from "bullmq";
+// import { logger } from "../server.js";
+// import dotenv from "dotenv";
+// import fs from "fs-extra";
+// import path from "path";
+// import { FolderPaths } from "./serverInit.js";
+// import gm from "gm";
+// const imageMagick = gm.subClass({ imageMagick: true });
+// export default async function (job: SandboxedJob) {
+//   // Do something with job
+//   console.log("*** We're in the processing Queue")
+//   const { file, convertedFileName } = job.data;
+//   try {
+//     logger.log("debug", `Attempting to convert ${file.filename} image to JPEG from HEIC.`);
+//     await ConvertToJpeg(file.path, `${file.destination}/${convertedFileName}`);
+//     logger.log("debug", `Converted ${file.filename} image to JPEG from HEIC.`);
+//     await handleOriginalFile(file, convertedFileName);
+//     file.filename = convertedFileName;
+//     file.mimetype = "image/jpeg";
+//     file.path = `${file.destination}/${convertedFileName}`;
+//   } catch (error) {
+//     logger.log("error", `Error converting ${file.filename} image to JPEG from HEIC. ${JSON.stringify(error)}`);
+//   }
+// }
+// async function handleOriginalFile(file: Express.Multer.File, convertedFileName: string) {
+//   if (process.env.KEEP_CONVERTED_ORIGINALS) {
+//     await fs.ensureDir(path.join(file.destination, FolderPaths.ConvertedOriginalSubDir));
+//     await fs.move(file.path, `${path.join(file.destination, FolderPaths.ConvertedOriginalSubDir)}/${file.filename}`);
+//   } else {
+//     await fs.unlink(file.path);
+//   }
+// }

-async function ConvertToJpeg(file: string, newPath: string) {
-  const fileOnDisk: Buffer = await fs.readFile(file);
+// async function ConvertToJpeg(file: string, newPath: string) {
+//   const fileOnDisk: Buffer = await fs.readFile(file);

-  return new Promise<string>((resolve, reject) => {
-    imageMagick(fileOnDisk)
-      .setFormat("jpg")
-      .write(newPath, (error) => {
-        if (error) reject(error.message);
-        resolve(newPath);
-      });
-  });
-}
+//   return new Promise<string>((resolve, reject) => {
+//     imageMagick(fileOnDisk)
+//       .setFormat("jpg")
+//       .write(newPath, (error) => {
+//         if (error) reject(error.message);
+//         resolve(newPath);
+//       });
+//   });
+// }
--- a/yarn.lock
+++ b/yarn.lock
@@ -23,10 +23,29 @@
    enabled "2.0.x"
    kuler "^2.0.0"

-"@img/sharp-win32-x64@0.33.5":
+"@img/sharp-libvips-linux-x64@1.0.4":
+  version "1.0.4"
+  resolved "https://registry.npmjs.org/@img/sharp-libvips-linux-x64/-/sharp-libvips-linux-x64-1.0.4.tgz"
+  integrity sha512-MmWmQ3iPFZr0Iev+BAgVMb3ZyC4KeFc3jFxnNbEPas60e1cIfevbtuyf9nDGIzOaW9PdnDciJm+wFFaTlj5xYw==
+
+"@img/sharp-libvips-linuxmusl-x64@1.0.4":
+  version "1.0.4"
+  resolved "https://registry.npmjs.org/@img/sharp-libvips-linuxmusl-x64/-/sharp-libvips-linuxmusl-x64-1.0.4.tgz"
+  integrity sha512-viYN1KX9m+/hGkJtvYYp+CCLgnJXwiQB39damAO7WMdKWlIhmYTfHjwSbQeUK/20vY154mwezd9HflVFM1wVSw==
+
+"@img/sharp-linux-x64@0.33.5":
  version "0.33.5"
-  resolved "https://registry.npmjs.org/@img/sharp-win32-x64/-/sharp-win32-x64-0.33.5.tgz"
-  integrity sha512-MpY/o8/8kj+EcnxwvrP4aTJSWw/aZ7JIGR4aBeZkZw5B7/Jn+tY9/VNwtcoGmdT7GfggGIU4kygOMSbYnOrAbg==
+  resolved "https://registry.npmjs.org/@img/sharp-linux-x64/-/sharp-linux-x64-0.33.5.tgz"
+  integrity sha512-opC+Ok5pRNAzuvq1AG0ar+1owsu842/Ab+4qvU879ippJBHvyY5n2mxF1izXqkPYlGuP/M556uh53jRLJmzTWA==
+  optionalDependencies:
+    "@img/sharp-libvips-linux-x64" "1.0.4"
+
+"@img/sharp-linuxmusl-x64@0.33.5":
+  version "0.33.5"
+  resolved "https://registry.npmjs.org/@img/sharp-linuxmusl-x64/-/sharp-linuxmusl-x64-0.33.5.tgz"
+  integrity sha512-WT+d/cgqKkkKySYmqoZ8y3pxx7lx9vVejxW/W4DOFMYVSkErR+w7mf2u8m/y4+xHe7yY9DAXQMWQhpnMuFfScw==
+  optionalDependencies:
+    "@img/sharp-libvips-linuxmusl-x64" "1.0.4"

 "@ioredis/commands@^1.1.1":
  version "1.2.0"
@@ -51,10 +70,10 @@
    "@jridgewell/resolve-uri" "^3.0.3"
    "@jridgewell/sourcemap-codec" "^1.4.10"

-"@msgpackr-extract/msgpackr-extract-win32-x64@3.0.3":
+"@msgpackr-extract/msgpackr-extract-linux-x64@3.0.3":
  version "3.0.3"
-  resolved "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-win32-x64/-/msgpackr-extract-win32-x64-3.0.3.tgz"
-  integrity sha512-x0fWaQtYp4E6sktbsdAqnehxDgEc/VwM7uLsRCYWaiGu0ykYdZPiS8zCWdnjHwyiumousxfBm4SO31eXqwEZhQ==
+  resolved "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-linux-x64/-/msgpackr-extract-linux-x64-3.0.3.tgz"
+  integrity sha512-cvwNfbP07pKUfq1uH+S6KJ7dT9K8WOE4ZiAcsrSes+UY55E/0jLYc+vq+DO7jlmqRb5zAggExKm0H7O/CBaesg==

 "@sec-ant/readable-stream@^0.4.1":
  version "0.4.1"