ADD: loggers and event listeners

Author: sb-vinayak

lib/events/JobFailed.ts

@@ -1,5 +1,9 @@
-import { EmitsEvent, Event } from '@squareboat/nest-events';
-import { events } from '../constants';
+import { EmitsEvent, Event } from "@squareboat/nest-events";
+import { events } from "../constants";
 
 @Event(events.jobFailed)
-export class JobFailed extends EmitsEvent {}
+export class JobFailed extends EmitsEvent {
+  constructor(public message: any, public job: any) {
+    super();
+  }
+}

lib/events/JobProcessed.ts

@@ -1,5 +1,9 @@
-import { EmitsEvent, Event } from '@squareboat/nest-events';
-import { events } from '../constants';
+import { EmitsEvent, Event } from "@squareboat/nest-events";
+import { events } from "../constants";
 
 @Event(events.jobProcessed)
-export class JobProcessed extends EmitsEvent {}
+export class JobProcessed extends EmitsEvent {
+  constructor(public message: any, public job: any) {
+    super();
+  }
+}

lib/events/JobProcessing.ts

@@ -1,5 +1,9 @@
-import { EmitsEvent, Event } from '@squareboat/nest-events';
-import { events } from '../constants';
+import { EmitsEvent, Event } from "@squareboat/nest-events";
+import { events } from "../constants";
 
 @Event(events.jobProcessing)
-export class JobProcessing extends EmitsEvent {}
+export class JobProcessing extends EmitsEvent {
+  constructor(public message: any, public job: any) {
+    super();
+  }
+}

lib/interfaces/options.ts

@@ -32,4 +32,5 @@ export interface ListenerOptions {
   connection?: string;
   queue?: string;
   schedulerInterval?: number;
+  logger?: boolean;
 }

lib/jobrunner.ts

@@ -6,33 +6,73 @@ import {
 import { ListenerOptions } from "./interfaces";
 import { QueueMetadata } from "./metadata";
 import { Dispatch } from "./queue";
+import { EmitEvent } from "@squareboat/nest-events";
+import { JobFailed, JobProcessed, JobProcessing } from "./events";
+import { ConsoleIO, Logger } from "@squareboat/nest-console";
 
 export class JobRunner {
+  private consoleIo: ConsoleIO;
+
   constructor(
     private options: ListenerOptions,
     private connection: QueueDriver
-  ) {}
+  ) {
+    this.consoleIo = new ConsoleIO("", {});
+  }
 
   async run(job: DriverJob) {
     const message = this.fetchMessage(job);
     const { data } = message;
     try {
+      this.log("info", `LOG [${message.job}] Job Processing...`);
       const targetJob = QueueMetadata.getJob(message.job);
       if (!targetJob || !targetJob.target) return;
-
+      const event = new JobProcessing(message, job);
+      event.emit();
       await targetJob.target(data);
       await this.success(message, job);
+      this.log("success", `LOG [${message.job}] Job Processed`);
     } catch (e) {
+      const event = new JobFailed(message, job);
+      event.emit();
       await this.retry(message, job);
+      const errorMessage = (e as Error).message;
+      this.log(
+        "error",
+        `LOG [${message.job}] Job Failed | Error: ${errorMessage}`
+      );
     }
   }
 
+  log(level: string, msg: string): void {
+    if (!this.options.logger) return;
+    let logger = undefined;
+    switch (level) {
+      case "info":
+        logger = Logger.info;
+        break;
+      case "success":
+        logger = Logger.success;
+        break;
+      case "error":
+        logger = Logger.error;
+        break;
+      case "warn":
+        logger = Logger.warn;
+        break;
+    }
+
+    logger && logger(msg);
+  }
+
   /**
    * Job processed succesfully method
    * @param message
    * @param job
    */
   async success(message: InternalMessage, job: DriverJob): Promise<void> {
+    const event = new JobProcessed(message, job);
+    event.emit();
     await this.removeJobFromQueue(job);
   }
 

lib/metadata.ts

@@ -11,7 +11,10 @@ interface JobTarget {
 @Injectable()
 export class QueueMetadata {
   private static data: QueueOptions;
-  private static defaultOptions: Record<string, string | number | undefined>;
+  private static defaultOptions: Record<
+    string,
+    string | number | boolean | undefined
+  >;
   private static store: Record<string, any> = { jobs: {} };
 
   constructor(@Inject(QUEUE_OPTIONS) data: QueueOptions) {

lib/queue.ts

@@ -14,7 +14,7 @@ export class Queue {
 
 export function Dispatch(message: Message): Promise<void> {
   const job = QueueMetadata.getJob(message.job);
-  const payload = PayloadBuilder.build(message, job.options);
+  const payload = PayloadBuilder.build(message, job?.options || {});
   const connection = QueueService.getConnection(payload.connection);
   return connection.push(JSON.stringify(payload), payload);
 }

lib/worker.ts

@@ -3,6 +3,7 @@ import { QueueMetadata } from "./metadata";
 import { QueueService } from "./service";
 import { JobRunner } from "./jobrunner";
 import { DriverJob, QueueDriver } from "@squareboat/nest-queue-strategy";
+import { Logger } from "@squareboat/nest-console";
 
 export class QueueWorker {
   private options: ListenerOptions;
@@ -14,6 +15,7 @@ export class QueueWorker {
       ...defaultOptions,
       schedulerInterval: 10000,
       queue: undefined,
+      logger: true,
       ...this.options,
     };
 
@@ -38,6 +40,7 @@ export class QueueWorker {
    * Listen to the queue
    */
   async listen() {
+    this.log("info", "LOG [QueueWorker] Queue Worker Initialised");
     const connection = QueueService.getConnection(this.options.connection);
 
     // perform scheduled task of the driver
@@ -75,4 +78,25 @@ export class QueueWorker {
     const connection = QueueService.getConnection(this.options.connection);
     return await connection.count({ queue: this.options.queue });
   }
+
+  log(level: string, msg: string): void {
+    if (!this.options.logger) return;
+    let logger = undefined;
+    switch (level) {
+      case "info":
+        logger = Logger.info;
+        break;
+      case "success":
+        logger = Logger.success;
+        break;
+      case "error":
+        logger = Logger.error;
+        break;
+      case "warn":
+        logger = Logger.warn;
+        break;
+    }
+
+    logger && logger(msg);
+  }
 }