code rewiev

services/utils/RabbitMQConnector.js

@@ -0,0 +1,37 @@
+const amqp = require('amqplib');
+
+class RabbitMQConnector {
+  constructor() {
+    this.connection = null;
+    this.channel = null;
+  }
+
+  async connect() {
+    try {
+      this.connection = await amqp.connect('amqp://username:password@messageBroker');
+      this.channel = await this.connection.createChannel();
+      console.log('Connected to RabbitMQ');
+    } catch (error) {
+      console.error('Error connecting to RabbitMQ:', error);
+    }
+  }
+
+  async disconnect() {
+    try {
+      if (this.channel) {
+        await this.channel.close();
+        console.log('Channel closed');
+      }
+
+      if (this.connection) {
+        await this.connection.close();
+        console.log('Connection closed');
+      }
+    } catch (error) {
+      console.error('Error disconnecting from RabbitMQ:', error);
+      throw error;
+    }
+  }
+}
+
+module.exports = RabbitMQConnector;

services/utils/RabbitMQListener.js

@@ -1,13 +1,45 @@
-const RabbitMQWrapper = require('./RabbitMQWrapper');
+const RabbitMQConnector = require('./RabbitMQConnector');
 
 class RabbitMQListener {
   constructor() {
-    this.rabbitmq = new RabbitMQWrapper();
+    this.rabbitmqConnector = new RabbitMQConnector();
+  }
+
+  async connect() { 
+    await this.rabbitmqConnector.connect();
+  }
+
+  async consumeFromQueue(queueName, callback) {
+    await this.connect();
+    try {
+      if (!this.rabbitmqConnector.channel) {
+        throw new Error('Channel is not initialized. Call connect() first.');
+      }
+
+      let result = await this.rabbitmqConnector.channel.assertQueue(queueName, { durable: true });
+      console.log('Queue asserted.',result);
+  
+      this.rabbitmqConnector.channel.consume(queueName, async (msg) => {
+        if (msg !== null) {
+          const content = msg.content.toString();
+          try {
+            await callback(content);
+            this.rabbitmqConnector.channel.ack(msg);
+          } catch (callbackError) {
+            console.error('Callback error:', callbackError);
+          }
+        }
+      });
+  
+      console.log(`Listening for messages on queue "${queueName}"`);
+    } catch (error) {
+      console.error('Error consuming messages:', error.message);
+    }
   }
 
   async listenToQueue(queueName, callback) {
     try {
-      await this.rabbitmq.consumeFromQueue(queueName, async (message) => {
+      await this.consumeFromQueue(queueName, async (message) => {
         await callback(message);
       });
     } catch (error) {

services/utils/RabbitMQProducer.js

@@ -1,28 +1,20 @@
-const amqp = require('amqplib');
+const RabbitMQConnector = require('./RabbitMQConnector');
 
 class RabbitMQProducer {
   
   constructor() {
-    this.connectionURL = 'amqp://username:password@messageBroker';
-    this.connection = null;
-    this.channel = null;
     this.connect();
   }
 
   async connect() {
-    try {
-      this.connection = await amqp.connect(this.connectionURL);
-      this.channel = await this.connection.createChannel();
-      console.log('Connected to RabbitMQ');
-    } catch (error) {
-      console.error('Error connecting to RabbitMQ:', error);
-    }
+    this.rabbitmqConnector = new RabbitMQConnector();
+    this.rabbitmqConnector.connect();
   }
 
   async publishToQueue(queueName, message) {
     try {
-      await this.channel.assertQueue(queueName, { durable: true });
-      this.channel.sendToQueue(queueName, Buffer.from(message));
+      await this.rabbitmqConnector.channel.assertQueue(queueName, { durable: true });
+      this.rabbitmqConnector.channel.sendToQueue(queueName, Buffer.from(message));
       console.log(`Published to ${queueName}: ${message}`);
     } catch (error) {
       console.error(`Error publishing to ${queueName}:`, error);

services/utils/RabbitMQWrapper.js

@@ -1,48 +0,0 @@
-const amqp = require('amqplib');
-
-class RabbitMQWrapper {
-  constructor() {
-    this.connection = null;
-    this.channel = null;
-  }
-
-  async connect() {
-    try {
-      this.connection = await amqp.connect('amqp://username:password@messageBroker');
-      this.channel = await this.connection.createChannel();
-      console.log('Connected to RabbitMQ');
-    } catch (error) {
-      console.error('Error connecting to RabbitMQ:', error);
-    }
-  }
-
-  async consumeFromQueue(queueName, callback) {
-    await this.connect();
-    try {
-      if (!this.channel) {
-        throw new Error('Channel is not initialized. Call connect() first.');
-      }
-
-      let result = await this.channel.assertQueue(queueName, { durable: true });
-      console.log('Queue asserted.',result);
-  
-      this.channel.consume(queueName, async (msg) => {
-        if (msg !== null) {
-          const content = msg.content.toString();
-          try {
-            await callback(content);
-            this.channel.ack(msg);
-          } catch (callbackError) {
-            console.error('Callback error:', callbackError);
-          }
-        }
-      });
-  
-      console.log(`Listening for messages on queue "${queueName}"`);
-    } catch (error) {
-      console.error('Error consuming messages:', error.message);
-    }
-  }
-}
-
-module.exports = RabbitMQWrapper;