feat: implementing addAtomic

Signed-off-by: Fawzi Abdulfattah <[email protected]>

Author: iifawzi

.gitignore

@@ -2,4 +2,5 @@
 node_modules
 
 .vscode
-package-lock.json
+package-lock.json
+testing.ts

core/connectors/ioredis.connector.ts

@@ -1,5 +1,6 @@
 /* eslint-disable @typescript-eslint/no-unused-vars */
 import { IClientConnector, ProducerConfigs } from "@core/types";
+import { GetPrefixedQueue } from "@core/helpers";
 import { Redis } from "ioredis";
 
 export class RedisIOClient implements IClientConnector<Redis> {
@@ -32,28 +33,29 @@ export class RedisIOClient implements IClientConnector<Redis> {
     return this.client;
   }
 
-  async add(queue: string, data: unknown): Promise<string> {
+  async add(queue: string, data: string): Promise<string> {
     if (this.client.status === "wait") {
       await this.client.connect();
     }
-    // it will never return null given NOMKSTREAM is not set.
-    return (await this.client.xadd(
-      queue,
-      "*",
-      "data",
-      JSON.stringify(data)
-    )) as unknown as string;
+    const clientResponse = await this.client.xadd(queue, "*", "data", data);
+    return clientResponse as string;
   }
 
   async addAtomic(
     queue: string,
     additionalQueues: string[],
-    data: unknown
+    data: string
   ): Promise<string[]> {
     if (this.client.status === "wait") {
       await this.client.connect();
     }
-    throw new Error("Method not implemented.");
+    const multi = this.client.multi();
+    multi.xadd(queue, "*", "data", data);
+    additionalQueues.forEach((additionalQueueName: string) => {
+      multi.xadd(GetPrefixedQueue(additionalQueueName), "*", "data", data);
+    });
+    const clientMultiResponse = await multi.exec();
+    return clientMultiResponse?.map((res) => res[1]) as string[];
   }
 
   async bulk(queue: string, data: unknown[]): Promise<string[]> {

core/helpers/get.prefixed.queue.ts

@@ -0,0 +1,5 @@
+import { CONSTANTS } from "@core/constants";
+
+export function GetPrefixedQueue(queue: string): string {
+  return CONSTANTS.QUEUE_PREFIX + queue;
+}

core/helpers/index.ts

@@ -0,0 +1 @@
+export * from "./get.prefixed.queue"

core/producer/producer.abstract.ts

@@ -1,4 +1,4 @@
-import { CONSTANTS } from "@core/constants";
+import { GetPrefixedQueue } from "@core/helpers";
 import { IClientConnector } from "@core/types";
 /**
  * AbstractProducer is an abstract class that provides basic functionalities for adding data to queues.
@@ -12,7 +12,7 @@ export default abstract class AbstractProducer<K> {
    * @param client - An instance of IClientConnector to interact with the queue.
    */
   constructor(private name: string, private client: IClientConnector<K>) {
-    this.queueName = CONSTANTS.QUEUE_PREFIX + this.name;
+    this.queueName = GetPrefixedQueue(this.name);
     process.once("SIGINT", this.handleShutdown.bind(this));
     process.once("SIGTERM", this.handleShutdown.bind(this));
   }
@@ -21,14 +21,23 @@ export default abstract class AbstractProducer<K> {
    * Adds data to the main queue and optionally to additional queues atomically.
    * @param data - The data to be added to the queue.
    * @param additionalQueues - Optional. Additional queues to which the data should be added atomically.
-   * @returns A promise that resolves when the data has been added.
+   * @returns A promise that resolves to the job ID(s) when the data has been added.
+   *          If data is added to a single queue, it returns a string representing the job ID.
+   *          If data is added to multiple queues, it returns an array of strings representing the jobs IDs.
    */
-  public async add(data: unknown, additionalQueues?: string[]) {
+  public async add(
+    data: unknown,
+    additionalQueues?: string[]
+  ): Promise<string | string[]> {
+    const stringifiedData = JSON.stringify(data);
     if (additionalQueues && additionalQueues.length) {
-      await this.client.addAtomic(this.queueName, additionalQueues, data);
-      return;
+      return await this.client.addAtomic(
+        this.queueName,
+        additionalQueues,
+        stringifiedData
+      );
     }
-    await this.client.add(this.queueName, data);
+    return await this.client.add(this.queueName, stringifiedData);
   }
 
   /**