Obtain X-Ray daemon address before app and worker init

Author: juffalow

src/index.ts

@@ -29,6 +29,7 @@ createTerminus(server, {
 
 async function start(): Promise<void> {
   try {
+    console.log('Main.start();');
     const namespace = config.services.trace.daemonAddressNamespace;
     const name = config.services.trace.daemonAddressName;
     if (typeof namespace === 'string' && typeof name === 'string') {
@@ -47,4 +48,5 @@ async function start(): Promise<void> {
   }
 }
 
+console.log('What the kurwa?!');
 start();

src/services/trace/index.ts

@@ -4,6 +4,8 @@ import config from '../../config';
 
 const container = {
   get AWSXRay() {
+    console.log('Trace.AWSXRay()');
+
     if (typeof this._awsXRay === 'undefined') {
       this._awsXRay = new AWSXRay(config.services.trace.plugins);
     }
@@ -12,6 +14,7 @@ const container = {
   },
 
   get CLSHooked() {
+    console.log('Trace.CLSHooked()');
     if (typeof this._clsHooked === 'undefined') {
       this._clsHooked = new CLSHooked();
     }

src/services/worker/Consumer.ts

@@ -0,0 +1,78 @@
+import { Consumer } from 'sqs-consumer';
+import services from '../../services';
+import config from '../../config';
+import logger from '../../logger';
+import video from '../../services/job/video';
+import repositories from '../../repositories';
+
+function createConsumer() {
+  const consumer = Consumer.create({
+    queueUrl: config.services.queue.url,
+    handleMessage: async (message) => {
+      const namespace: any = services.Trace.getNamespace();
+      const segment: any = services.Trace.createSegment('upload-service-worker');
+      
+      await namespace.runPromise(async () => {
+        services.Trace.setSegment(segment);
+
+        logger.debug('Received message from queue!', { ...message });
+
+        const body = JSON.parse(message.Body);
+
+        if ('traceId' in body) {
+          namespace.set('traceId', body.traceId);
+        } else {
+          logger.warn('Trace ID is not present in message body!');
+        }
+    
+        if ('fileId' in body) {
+          try {
+            const file = await repositories.File.get(body.fileId, null);
+            const values = await services.Job.process(body.mimetype, file) as Record<string, string | number | boolean>;
+            await repositories.File.update({ ...values, cacheable: !body.mimetype.startsWith('video') }, { id: file.id });
+          } catch (err) {
+            logger.error('Could not process file!', { error: err.message, stack: err.stack });
+          }
+        }
+    
+        if ('detail' in body && 'jobId' in body.detail) {
+          if (body.detail.status === 'COMPLETE') {
+            const updatedFile = await video.AWSVideo.complete(body) as Record<string, string | number | boolean>;
+            await repositories.File.update({ ...updatedFile, cacheable: true }, { id: updatedFile['id'] as string });
+          }
+    
+          if (body.detail.status === 'ERROR') {
+            logger.error('Media convert job ended with error!', body);
+          }
+        }
+
+        segment.close();
+      });
+    },
+    sqs: services.AWS.sqs,
+  });
+
+  consumer.on('message_processed', (message) => {
+    logger.debug('Message successfully processed and removed from the queue!', { message });
+  });
+
+  consumer.on('error', (err) => {
+    logger.error('Unknown error occured!', { message: err.message, stack: err.stack });
+  });
+
+  consumer.on('processing_error', (err) => {
+    logger.error('Error while processing message from queue!', { message: err.message, stack: err.stack });
+
+    services.Trace.closeSegment();
+  });
+
+  consumer.on('timeout_error', (err) => {
+    logger.error('Timeout error!', { message: err.message, stack: err.stack });
+  });
+
+  return consumer;
+}
+
+export default {
+  get: createConsumer,
+}

src/worker.ts

@@ -1,73 +1,7 @@
-import { Consumer } from 'sqs-consumer';
 import services from './services';
 import config from './config';
 import logger from './logger';
-import video from './services/job/video';
-import repositories from './repositories';
-
-const app = Consumer.create({
-  queueUrl: config.services.queue.url,
-  handleMessage: async (message) => {
-    const namespace: any = services.Trace.getNamespace();
-    const segment: any = services.Trace.createSegment('upload-service-worker');
-    
-    await namespace.runPromise(async () => {
-      services.Trace.setSegment(segment);
-
-      logger.debug('Received message from queue!', { ...message });
-
-      const body = JSON.parse(message.Body);
-
-      if ('traceId' in body) {
-        namespace.set('traceId', body.traceId);
-      } else {
-        logger.warn('Trace ID is not present in message body!');
-      }
-  
-      if ('fileId' in body) {
-        try {
-          const file = await repositories.File.get(body.fileId, null);
-          const values = await services.Job.process(body.mimetype, file) as Record<string, string | number | boolean>;
-          await repositories.File.update({ ...values, cacheable: !body.mimetype.startsWith('video') }, { id: file.id });
-        } catch (err) {
-          logger.error('Could not process file!', { error: err.message, stack: err.stack });
-        }
-      }
-  
-      if ('detail' in body && 'jobId' in body.detail) {
-        if (body.detail.status === 'COMPLETE') {
-          const updatedFile = await video.AWSVideo.complete(body) as Record<string, string | number | boolean>;
-          await repositories.File.update({ ...updatedFile, cacheable: true }, { id: updatedFile['id'] as string });
-        }
-  
-        if (body.detail.status === 'ERROR') {
-          logger.error('Media convert job ended with error!', body);
-        }
-      }
-
-      segment.close();
-    });
-  },
-  sqs: services.AWS.sqs,
-});
-
-app.on('message_processed', (message) => {
-  logger.debug('Message successfully processed and removed from the queue!', { message });
-});
-
-app.on('error', (err) => {
-  logger.error('Unknown error occured!', { message: err.message, stack: err.stack });
-});
-
-app.on('processing_error', (err) => {
-  logger.error('Error while processing message from queue!', { message: err.message, stack: err.stack });
-
-  services.Trace.closeSegment();
-});
-
-app.on('timeout_error', (err) => {
-  logger.error('Timeout error!', { message: err.message, stack: err.stack });
-});
+import Consumer from './services/worker/Consumer';
 
 async function start() {
   const namespace = config.services.trace.daemonAddressNamespace;
@@ -81,7 +15,8 @@ async function start() {
 
   logger.info('Upload worker is running');
 
-  app.start();
+  const consumer = Consumer.get();
+  consumer.start();
 }
 
 start();