wip: test segments

Author: rakibansary

src/app.js

@@ -3,6 +3,10 @@
  */
 
 require('./bootstrap')
+
+const AWSXRay = require('aws-xray-sdk')
+const ns = AWSXRay.getNamespace();
+
 const _ = require('lodash')
 const config = require('config')
 const Kafka = require('no-kafka')
@@ -11,8 +15,6 @@ const logger = require('./common/logger')
 const helper = require('./common/helper')
 const ProcessorService = require('./services/ProcessorService')
 
-const AWSXRay = require('aws-xray-sdk')
-const segment = new AWSXRay.Segment('legacy-challenge-processor');
 
 // Start kafka consumer
 logger.info('Starting kafka consumer')
@@ -26,6 +28,9 @@ const consumer = new Kafka.GroupConsumer(helper.getKafkaOptions())
  * this function will be invoked
  */
 const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, async (m) => {
+  const segment = new AWSXRay.Segment('legacy-challenge-processor');
+  AWSXRay.setSegment(segment);
+  
   const message = m.message.value.toString('utf8')
   logger.info(`Handle Kafka event message; Topic: ${topic}; Partition: ${partition}; Offset: ${m.offset}; Message: ${message}.`)
 
@@ -57,67 +62,57 @@ const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, a
     return
   }
 
-  const ns = AWSXRay.getNamespace();
-  console.log('Running within context', ns);
-  ns.run(async () => {
-    console.log('creating segment');
-    console.log('created segment');
-    AWSXRay.setSegment(segment);
-    console.log('set segment');
-
-    const { traceInformation: {
-      traceId,
-      parentSegmentId,
-    } = {
-      traceId: null,
-      parentSegmentId: null
-    } } = messageJSON.payload;
-
-    console.log('tracing information', traceId, parentSegmentId);
-
-    if (traceId) {
-      segment.trace_id = traceId;
-      segment.id = parentSegmentId;
-    }
+  const { traceInformation: {
+    traceId,
+    parentSegmentId,
+  } = {
+    traceId: null,
+    parentSegmentId: null
+  } } = messageJSON.payload;
 
+  console.log('tracing information', traceId, parentSegmentId);
 
-    // do not trust the message payload
-    // the message.payload will be replaced with the data from the API
-    try {
-      console.log('Fetch challenge details');
-      const challengeUuid = _.get(messageJSON, 'payload.id')
-      if (_.isEmpty(challengeUuid)) {
-        segment.close();
-        segment.addError(new Error(err));
-        throw new Error('Invalid payload')
-      }
-      const m2mToken = await helper.getM2MToken()
-      const v5Challenge = await helper.getRequest(`${config.V5_CHALLENGE_API_URL}/${challengeUuid}`, m2mToken)
-      // TODO : Cleanup. Pulling the billingAccountId from the payload, it's not part of the challenge object
-      messageJSON.payload = { billingAccountId: messageJSON.payload.billingAccountId, ...v5Challenge.body }
-    } catch (err) {
-      segment.addError(new Error(err));
-      logger.debug('Failed to fetch challenge information')
-      logger.logFullError(err)
-    }
+  if (traceId) {
+    segment.trace_id = traceId;
+    segment.id = parentSegmentId;
+  }
 
-    try {
-      console.log('Process challenge')
-      await ProcessorService.processMessage(messageJSON)
 
-      // logger.debug('Successfully processed message')
-    } catch (err) {
+  // do not trust the message payload
+  // the message.payload will be replaced with the data from the API
+  try {
+    console.log('Fetch challenge details');
+    const challengeUuid = _.get(messageJSON, 'payload.id')
+    if (_.isEmpty(challengeUuid)) {
+      segment.close();
       segment.addError(new Error(err));
-      logger.error(`Error processing message ${JSON.stringify(messageJSON)}`)
-      logger.logFullError(err)
-    } finally {
-      // Commit offset regardless of error
-      await consumer.commitOffset({ topic, partition, offset: m.offset })
+      throw new Error('Invalid payload')
     }
+    const m2mToken = await helper.getM2MToken()
+    const v5Challenge = await helper.getRequest(`${config.V5_CHALLENGE_API_URL}/${challengeUuid}`, m2mToken)
+    // TODO : Cleanup. Pulling the billingAccountId from the payload, it's not part of the challenge object
+    messageJSON.payload = { billingAccountId: messageJSON.payload.billingAccountId, ...v5Challenge.body }
+  } catch (err) {
+    segment.addError(new Error(err));
+    logger.debug('Failed to fetch challenge information')
+    logger.logFullError(err)
+  }
 
-    segment.close();
-  })
+  try {
+    console.log('Process challenge')
+    await ProcessorService.processMessage(messageJSON)
+
+    // logger.debug('Successfully processed message')
+  } catch (err) {
+    segment.addError(new Error(err));
+    logger.error(`Error processing message ${JSON.stringify(messageJSON)}`)
+    logger.logFullError(err)
+  } finally {
+    // Commit offset regardless of error
+    await consumer.commitOffset({ topic, partition, offset: m.offset })
+  }
 
+  segment.close();
 })
 
 // check if there is kafka connection alive
@@ -135,20 +130,24 @@ const check = () => {
 
 const topics = [config.CREATE_CHALLENGE_TOPIC, config.UPDATE_CHALLENGE_TOPIC]
 
-consumer
-  .init([{
-    subscriptions: topics,
-    handler: dataHandler
-  }])
-  // consume configured topics
-  .then(() => {
-    logger.info('Initialized.......')
-    healthcheck.init([check])
-    logger.info('Adding topics successfully.......')
-    logger.info(topics)
-    logger.info('Kick Start.......')
+(() => {
+  ns.run(() => {
+    consumer
+    .init([{
+      subscriptions: topics,
+      handler: dataHandler
+    }])
+    // consume configured topics
+    .then(() => {
+      logger.info('Initialized.......')
+      healthcheck.init([check])
+      logger.info('Adding topics successfully.......')
+      logger.info(topics)
+      logger.info('Kick Start.......')
+    })
+    .catch((err) => logger.error(err))
   })
-  .catch((err) => logger.error(err))
+})();
 
 if (process.env.NODE_ENV === 'test') {
   module.exports = consumer