Merge pull request #27 from EYBlockchain/aleandrom/TAX-2490-support-trackingID-for-breadcrumbs

feat: rabbitmq reconnection logic & support for trackingIDs

Author: aleandrom

src/queues/generateProof.mjs

@@ -1,6 +1,7 @@
 import rabbitmq from '../utils/rabbitmq.mjs';
 import logger from '../utils/logger.mjs';
 import generateProof from '../services/generateProof.mjs';
+import { formatTrackingID } from '../utils/formatter.mjs';
 
 export default function receiveMessage() {
   rabbitmq.receiveMessage('generate-proof', async message => {
@@ -11,17 +12,21 @@ export default function receiveMessage() {
       data: null,
     };
 
+    let trackingID = '';
     try {
-      response.data = await generateProof(JSON.parse(
-        message.content.toString(),
-      ));
+      trackingID = JSON.parse(message.content.toString()).trackingID;
+      response.data = await generateProof(JSON.parse(message.content.toString()));
     } catch (err) {
-      logger.error('Error in generate-proof', err);
-      response.error = 'Proof generation failed';
+      logger.error(`${formatTrackingID(trackingID)} Error in generate-proof:`, err);
+      response.error = `${formatTrackingID(trackingID)} Proof generation failed`;
     }
 
 
     rabbitmq.sendMessage(replyTo, response, { correlationId });
-    rabbitmq.sendACK(message);
+    try {
+      rabbitmq.sendACK(message);
+    } catch (error) {
+      logger.warn(`The acknowledgement failed, the channel was closed: ${error}`);
+    }
   });
 }

src/services/generateKeys.mjs

@@ -1,7 +1,6 @@
 import fs from 'fs';
 import path from 'path';
 import zokrates from '@eyblockchain/zokrates-zexe.js';
-import rabbitmq from '../utils/rabbitmq.mjs';
 import logger from '../utils/logger.mjs';
 
 export default async function({

src/services/generateProof.mjs

@@ -5,6 +5,7 @@ import zokrates from '@eyblockchain/zokrates-zexe.js';
 import path from 'path';
 import { getProofFromFile } from '../utils/filing.mjs';
 import logger from '../utils/logger.mjs';
+import { formatTrackingID } from '../utils/formatter.mjs';
 
 const unlink = util.promisify(fs.unlink);
 
@@ -16,6 +17,7 @@ export default async function ({
   proofFileName,
   backend = 'ark', // zexe backend now named ark
   provingScheme = 'gm17',
+  trackingID,
 }) {
   const outputPath = `./output`;
   let proof, publicInputs;
@@ -29,11 +31,11 @@ export default async function ({
   const proofJsonFile = `${circuitName}_${fileNamePrefix}_proof.json`;
 
   if (fs.existsSync(`${outputPath}/${folderpath}/${witnessFile}`)) {
-    throw Error('Witness file with same name exists');
+    throw Error(`${formatTrackingID(trackingID)} Witness file with same name exists`);
   }
 
   if (fs.existsSync(`${outputPath}/${folderpath}/${proofJsonFile}`)) {
-    throw Error('proof.json file with same name exists');
+    throw Error(`${formatTrackingID(trackingID)} proof.json file with same name exists`);
   }
 
   const opts = {};
@@ -42,15 +44,15 @@ export default async function ({
   opts.fileName = proofFileName || `${proofJsonFile}`;
 
   try {
-    logger.info('Compute witness...');
+    logger.info(`${formatTrackingID(trackingID)} Compute witness...`);
     await zokrates.computeWitness(
       `${outputPath}/${folderpath}/${circuitName}_out`,
       `${outputPath}/${folderpath}/`,
       `${witnessFile}`,
       inputs,
     );
 
-    logger.info('Generate proof...');
+    logger.info(`${formatTrackingID(trackingID)} Generate proof...`);
     await zokrates.generateProof(
       `${outputPath}/${folderpath}/${circuitName}_pk.key`,
       `${outputPath}/${folderpath}/${circuitName}_out`,
@@ -63,7 +65,7 @@ export default async function ({
     ({ proof, inputs: publicInputs } = await getProofFromFile(`${folderpath}/${proofJsonFile}`));
 
     logger.info(`Complete`);
-    logger.debug(`Responding with proof and inputs:`);
+    logger.debug(`${formatTrackingID(trackingID)}Responding with proof and inputs:`);
     logger.debug(publicInputs);
   } finally {
     try {

src/utils/formatter.mjs

@@ -0,0 +1,3 @@
+export const formatTrackingID = trackingID => {
+  return trackingID ? `[${trackingID}] - ` : '';
+};

src/utils/rabbitmq.mjs

@@ -1,61 +1,89 @@
 import amqp from 'amqplib';
+import queues from '../queues/index.mjs';
+import logger from '../utils/logger.mjs';
+
+let connection;
+let channel;
+
+// connect to RabbitMQ server.
+const connect = async () => {
+  logger.info('[AMQP] Connecting...');
+  connection = await amqp.connect(`${process.env.RABBITMQ_HOST}:${process.env.RABBITMQ_PORT}`);
+  connection.on('error', err => {
+    if (err.message !== 'Connection closing') {
+      logger.error(`[AMQP] Connection error: ${err.message}`);
+      return setTimeout(connect, 1000);
+    }
+  });
+  connection.on('close', () => {
+    logger.debug('[AMQP] Reconnecting');
+    return setTimeout(connect, 1000);
+  });
+  channel = await connection.createChannel();
+  channel.prefetch(1);
+  queues();
+};
+
+// publish message to a queue
+const sendMessage = async (queue, data, options = {}) => {
+  channel.sendToQueue(queue, Buffer.from(JSON.stringify(data)), options);
+};
+
+// only called from test-suite ./test/queue.mjs
+const cancelChannelConsume = consumerTag => {
+  channel.cancel(consumerTag);
+};
+
+const sendACK = message => {
+  channel.ack(message);
+};
+
+// only called from test-suite ./test/queue.mjs
+const sendNACK = message => {
+  channel.nack(message);
+};
+
+/*
+ * Consumer: receive message from a queue
+ */
+const receiveMessage = async (queue, callback) => {
+  await channel.assertQueue(queue);
+  channel.consume(queue, callback, { noAck: false });
+};
+
+// only called from test-suite ./test/queue.mjs
+const listenToReplyQueue = async (queue, correlationId, callback) => {
+  logger.info(`[AMQP] Listening to reply queue: ${queue}`);
+  receiveMessage(queue, message => {
+    if (message.properties.correlationId !== correlationId) {
+      logger.debug(`[AMQP] Sending NACK due to different correlation id: ${JSON.stringify({ message: message.properties.correlationId, param: correlationId })}`);
+      return sendNACK(message);
+    }
+    cancelChannelConsume(message.fields.consumerTag);
+    sendACK(message);
+
+    const response = JSON.parse(message.content.toString());
+    response.type = message.properties.type;
+
+    return callback(response);
+  });
+};
+
+// only called from test-suite ./test/queue.mjs
+const close = async () => {
+  await channel.close();
+  await connection.close();
+};
 
 export default {
-  // connect to RabbitMQ server.
-  async connect() {
-    this.connection = await amqp.connect(
-      `${process.env.RABBITMQ_HOST}:${process.env.RABBITMQ_PORT}`,
-    );
-    this.channel = await this.connection.createChannel();
-    this.channel.prefetch(1);
-  },
-
-  // publish message to a queue
-  async sendMessage(queue, data, options = {}) {
-    this.channel.sendToQueue(queue, Buffer.from(JSON.stringify(data)), options);
-  },
-
-  // only called from test-suite ./test/queue.mjs
-  cancelChannelConsume(consumerTag) {
-    this.channel.cancel(consumerTag);
-  },
-
-  sendACK(message) {
-    this.channel.ack(message);
-  },
-
-  // only called from test-suite ./test/queue.mjs
-  sendNACK(message) {
-    this.channel.nack(message);
-  },
-
-  /*
-   * Consumer: receive message from a queue
-   */
-  async receiveMessage(queue, callback) {
-    await this.channel.assertQueue(queue);
-    this.channel.consume(queue, callback);
-  },
-
-  // only called from test-suite ./test/queue.mjs
-  listenToReplyQueue(queue, correlationId, callback) {
-    this.receiveMessage(queue, message => {
-      if (message.properties.correlationId !== correlationId) {
-        return this.sendNACK(message);
-      }
-      this.cancelChannelConsume(message.fields.consumerTag);
-      this.sendACK(message);
-
-      const response = JSON.parse(message.content.toString());
-      response.type = message.properties.type;
-
-      return callback(response);
-    });
-  },
-
-  // only called from test-suite ./test/queue.mjs
-  async close() {
-    await this.channel.close();
-    await this.connection.close();
-  },
+  connection,
+  channel,
+  connect,
+  sendMessage,
+  cancelChannelConsume,
+  sendACK,
+  sendNACK,
+  receiveMessage,
+  listenToReplyQueue,
+  close,
 };