🔧 Bull + Redis task queue first draft

services/api.service.ts

@@ -99,39 +99,24 @@ export default class ApiService extends Service {
 					this.logger.info("Client connected via websocket!");
 
 					client.on("action", (action, done) => {
-						console.log(action);
 						switch (action.type) {
 							case "LS_IMPORT":
 								this.broker
 									.call(
-										"import.processAndImportToDB",
+										"libraryqueue.enqueue",
 										action.data,
 										{}
 									)
 									.then((res) => {
-										if (done) done(res);
+										if (done) {
+											done(res);
+										}
 									})
 									.catch((err) => this.logger.error(err));
 								break;
 						}
 					});
-
+					// Add a disconnect listener
-					// client.on("call", ({ action, params, opts }, done) => {
-					// 	this.logger.info(
-					// 		"Received request from client! Action:",
-					// 		action,
-					// 		", Params:",
-					// 		params
-					// 	);
-
-					// 	this.broker
-					// 		.call(action, params, opts)
-					// 		.then((res) => {
-					// 			if (done) done(res);
-					// 		})
-					// 		.catch((err) => this.logger.error(err));
-					// });
-
 					client.on("disconnect", () => {
 						this.logger.info("Client disconnected");
 					});

services/import.service.ts

@@ -91,7 +91,6 @@ export default class ImportService extends Service {
 								try {
 									const { extractionOptions, walkedFolders } =
 										ctx.params;
-									// map(walkedFolders, async (folder, idx) => {
 										let comicExists = await Comic.exists({
 											"rawFileDetails.name": `${walkedFolders.name}`,
 										});
@@ -127,12 +126,12 @@ export default class ImportService extends Service {
 													{}
 												);
 											
+												return await dbImportResult;
 										} else {
 											logger.info(
 												`Comic: \"${walkedFolders.name}\" already exists in the database`
 											);
 										}
-									// });
 								} catch (error) {
 									logger.error(
 										"Error importing comic books",

services/library.queue.service.ts

@@ -6,7 +6,6 @@ import {
 	ServiceSchema,
 	Errors,
 } from "moleculer";
-
 import BullMQMixin from "moleculer-bull";
 
 export default class LibraryQueueService extends Service {
@@ -23,13 +22,13 @@ export default class LibraryQueueService extends Service {
 					settings: {},
 					hooks: {},
 					queues: {
-						"mail.send": {
+						"process.import": {
 							async process(job) {
 								this.logger.info("New job received!", job.data);
 								this.logger.info(`Processing queue...`);
-								// const accounts = await this.broker.call('v1.users.list');
+								const result = await this.broker.call('import.processAndImportToDB', job.data);
-								// this.logger.info(accounts);
 								return Promise.resolve({
+                                    result,
 									id: job.id,
 									worker: process.pid,
 								});
@@ -40,36 +39,40 @@ export default class LibraryQueueService extends Service {
 						enqueue: {
 							rest: "POST /enqueue",
 							params: {},
-							async handler(ctx: Context<{}>) {
+							async handler(ctx: Context<{ extractionOptions: object, walkedFolders: object}>) {
+                                console.log(ctx.params);
 								const job = await this.createJob("mail.send", {
-									blah: "blah",
+                                    extractionOptions: ctx.params.extractionOptions,
-								});
+									walkedFolders: ctx.params.walkedFolders,
-								const failed = await this.getQueue(
-									"mail.send"
-								).on("failed", async (job, error) => {
-									this.logger.error(
-										`An error occured in 'mail.send' queue on job id '${job.id}': ${error.message}`
-									);
-								});
-								const completed = await this.getQueue(
-									"mail.send"
-								).on("completed", async (job, res) => {
-									this.logger.info(
-										`Job with the id '${job.id}' completed.`
-									);
-								});
-								const stalled = await this.getQueue(
-									"mail.send"
-								).on("stalled", async (job) => {
-									this.logger.warn(
-										`The job with the id '${job} got stalled!`
-									);
 								});
+								
 							},
 						},
 					},
 					methods: {},
-					async started(): Promise<any> {},
+					async started(): Promise<any> {
+                        const failed = await this.getQueue(
+                            "process.import"
+                        ).on("failed", async (job, error) => {
+                            this.logger.error(
+                                `An error occured in 'mail.send' queue on job id '${job.id}': ${error.message}`
+                            );
+                        });
+                        const completed = await this.getQueue(
+                            "process.import"
+                        ).on("completed", async (job, res) => {
+                            this.logger.info(
+                                `Job with the id '${job.id}' completed.`
+                            );
+                        });
+                        const stalled = await this.getQueue(
+                            "process.import"
+                        ).on("stalled", async (job) => {
+                            this.logger.warn(
+                                `The job with the id '${job} got stalled!`
+                            );
+                        });
+                    },
 				},
 				schema
 			)