cleaning...

Author: Sachin Maheshwari

src/producer.js

@@ -14,25 +14,21 @@ const express = require('express')
 const app = express()
 const port = 3000
 const isFailover = process.argv[2] != undefined ? (process.argv[2] === 'failover' ? true : false) : false
-var pl_processid;
-//var pl_randonseq = 'err-' + (new Date()).getTime().toString(36) + Math.random().toString(36).slice(2)
 async function setupPgClient() {
   try {
     await pgClient.connect()
     for (const triggerFunction of pgOptions.triggerFunctions) {
       await pgClient.query(`LISTEN ${triggerFunction}`)
     }
     pgClient.on('notification', async (message) => {
-      // need to take care if empty message coming 
-      pl_processid = message.processId
       try {
         const payload = JSON.parse(message.payload)
         const validTopicAndOriginator = (pgOptions.triggerTopics.includes(payload.topic)) && (pgOptions.triggerOriginators.includes(payload.originator)) // Check if valid topic and originator
         if (validTopicAndOriginator) {
           if (!isFailover) {
             logger.info('trying to push on kafka topic')
             await pushToKafka(payload)
-            audit(payload)
+            audit(message)
             logger.info('pushed to kafka and added for audit trail')
           } else {
             logger.info('taking backup on dynamodb for reconciliation')
@@ -70,14 +66,16 @@ async function run() {
 // execute  
 run()
 
-async function audit() {
-  var pl_seqid = payload.payload.payloadseqid
-  var pl_topic = payload.topic
-  var pl_table = payload.payload.table
-  var pl_uniquecolumn = payload.payload.Uniquecolumn
-  var pl_operation = payload.payload.operation
-  var pl_timestamp = payload.timestamp
-  var pl_payload = JSON.stringify(payload.payload)
+async function audit(message) {
+  const pl_processid = message.processId
+  const payload = JSON.parse(message.payload)
+  const pl_seqid = payload.payload.payloadseqid
+  const pl_topic = payload.topic
+  const pl_table = payload.payload.table
+  const pl_uniquecolumn = payload.payload.Uniquecolumn
+  const pl_operation = payload.payload.operation
+  const pl_timestamp = payload.timestamp
+  const pl_payload = JSON.stringify(payload.payload)
   logger.debug(`producer : ${pl_seqid} ${pl_processid} ${pl_table} ${pl_uniquecolumn} ${pl_operation} ${payload.timestamp}`);
   auditTrail([pl_seqid, pl_processid, pl_table, pl_uniquecolumn, pl_operation, "push-to-kafka", "", "", "", pl_payload, pl_timestamp, pl_topic], 'producer')
 }