RabbitMQ exit call + correlation for AppDynamics

rabbitmq.js

const appdynamics = require("appdynamics")
appdynamics.profile({
   controllerHostName: '.saas.appdynamics.com',
   controllerPort: 443,
   controllerSslEnabled: true,
   accountName: '',
   accountAccessKey: '',
   applicationName: '',
   tierName: '',
   debug: true,
   nodeName: '' // The controller will automatically append the node name with a unique number
});


// Use this method around publish (see for an example below). Provide the following parameters:
// - appdynamics: The appdynamics instance created by require("appdynamics")
// - request: the http request object
// - identifiyngProperties: Provide, host, port, queue, routing key etc to identify the backend
// - debug: set this property to true to investiage issues.
function addExitandCorrelationHeader(appdynamics, request, identifyingProperties, debug = false) {
  return new Promise((resolve, reject) => {
    let txn = null;
    let headers = {}
    let exitCall = null;
    let correlationHeader = null;

    try {
      if (appdynamics) {
        txn = appdynamics.getTransaction(request);
        if (txn) {

            backendName = "amqp://"
            if(identifyingProperties.host) {
              txn.addSnapshotData("amqp-host", identifyingProperties.host)
              backendName += identifyingProperties.host
            }
            if(identifyingProperties.port) {
              txn.addSnapshotData("amqp-port", identifyingProperties.port)
              backendName += ":" + identifyingProperties.port
            }
            if(identifyingProperties.queue) {
              txn.addSnapshotData("amqp-queue", identifyingProperties.queue)
               backendName += "/" + identifyingProperties.queue
            }
            if(identifyingProperties['routing key']) {
              txn.addSnapshotData("amqp-routing-key", identifyingProperties['routing key'])
            }

            exitCall = txn.startExitCall({
                exitType: "EXIT_RABBITMQ",
                backendName,
                identifyingProperties
            });

            if(exitCall) {
              correlationHeader = txn.createCorrelationInfo(exitCall);
              //correlationHeader = correlationHeader.replace("etypeorder=RABBITMQ", "etypeorder=JMS*esubtype=JMS")
              if(debug) {
                console.log(exitCall)
                console.log('Adding correlation header: ' + correlationHeader);
              }
              if(correlationHeader) {
                headers.singularityheader = correlationHeader;
              }
            }
        } else if(debug) {
          console.log('Could not get txn object from appdynamics agent.');
        }
      } else if(debug) {
        console.log('No appdynamics agent provided, exit call not detected.');
      }
    } catch(e) {
      if(debug) {
        console.log(e)
      }
    }

    const endExitCall = function() {
      if(txn) {
        txn.endExitCall(exitCall);
      }
    }

    resolve({ headers, endExitCall })
  })
}

const http = require('http')
var amqp = require('amqplib/callback_api');

const port = 3000

const requestHandler = (request, response) => {
  console.log(request.url)
  try {
    amqp.connect('amqp://localhost', function(error0, connection) {
      if (error0) {
          throw error0;
      }
      connection.createChannel(function(error1, channel) {
          if (error1) {
              throw error1;
          }

          var exchange = 'logs';
          var routingKey = 'info';
          var msg = 'Hello Node.JS!';

          channel.assertExchange(exchange, 'fanout', {
              durable: false
          });

          // With confirm channel
          /* addExitandCorrelationHeader(appdynamics, request, { host: 'localhost', port: 5672, queue: 'logs', 'routing key': routingKey }, true).then(function(result) {
            const { headers, endExitCall } = result
            channel.publish(exchange, routingKey, Buffer.from(msg), { headers }, (err, ok) => {
              console.log(" [x] Sent %s", msg);
              endExitCall()
              response.end('Hello Node.js Server!')
            })
          })*/

          // Without confirm channel
          addExitandCorrelationHeader(appdynamics, request, { host: 'localhost', port: 5672, queue: 'logs', 'routing key': routingKey }, true).then(function(result) {
            const { headers, endExitCall } = result
            channel.publish(exchange, routingKey, Buffer.from(msg), { headers })
            console.log(" [x] Sent %s", msg);
            endExitCall()
            response.end('Hello Node.js Server!')
          })
      });
    });
  } catch (e) {
    console.log(e)
  }
}

const server = http.createServer(requestHandler)

server.listen(port, (err) => {
  if (err) {
    return console.log('something bad happened', err)
  }

  console.log(`server is listening on ${port}`)
})