Merge pull request #60 from taskforcesh/feat-job-progress-getter-for-…

…sanboxed-processors

job.progress() return last progress for sandboxed processors

src/classes/master.ts

@@ -92,17 +92,45 @@ process.on('uncaughtException', err => {
   throw err;
 });
 
+/**
+ * Enhance the given job argument with some functions
+ * that can be called from the sandboxed job processor.
+ *
+ * Note, the `job` argument is a JSON deserialized message
+ * from the main node process to this forked child process,
+ * the functions on the original job object are not in tact.
+ * The wrapped job adds back some of those original functions.
+ */
 function wrapJob(job: any) {
   job.data = JSON.parse(job.data || '{}');
   job.opts = JSON.parse(job.opts || '{}');
 
+  /*
+   * Emulate the real job `progress` function.
+   * If no argument is given, it behaves as a sync getter.
+   * If an argument is given, it behaves as an async setter.
+   */
+  let progressValue = job.progress;
   job.progress = function(progress: any) {
-    process.send({
-      cmd: 'progress',
-      value: progress,
-    });
-    return Promise.resolve();
+    if (progress) {
+      // Locally store reference to new progress value
+      // so that we can return it from this process synchronously.
+      progressValue = progress;
+      // Send message to update job progress.
+      process.send({
+        cmd: 'progress',
+        value: progress,
+      });
+      return Promise.resolve();
+    } else {
+      // Return the last known progress value.
+      return progressValue;
+    }
   };
+
+  /*
+   * Emulate the real job `log` function.
+   */
   job.log = function(row: any) {
     process.send({
       cmd: 'log',

src/test/fixtures/fixture_processor_progress.js

@@ -10,18 +10,22 @@ module.exports = function(job) {
   return delay(50)
     .then(() => {
       job.progress(10);
+      job.log(job.progress());
       return delay(100);
     })
     .then(() => {
       job.progress(27);
+      job.log(job.progress());
       return delay(150);
     })
     .then(() => {
       job.progress(78);
+      job.log(job.progress());
       return delay(100);
     })
     .then(() => {
-      return job.progress(100);
+      job.progress(100);
+      job.log(job.progress());
     })
     .then(() => {
       return 37;

src/test/test_sandboxed_process.ts

@@ -205,15 +205,20 @@ describe('sandboxed process', () => {
 
     const progresses: any[] = [];
 
-    const completting = new Promise((resolve, reject) => {
-      worker.on('completed', (job, value) => {
+    const completing = new Promise((resolve, reject) => {
+      worker.on('completed', async (job, value) => {
         try {
           expect(job.data).to.be.eql({ foo: 'bar' });
           expect(value).to.be.eql(37);
           expect(job.progress).to.be.eql(100);
           expect(progresses).to.be.eql([10, 27, 78, 100]);
           expect(Object.keys(worker['childPool'].retained)).to.have.lengthOf(0);
           expect(worker['childPool'].getAllFree()).to.have.lengthOf(1);
+          const logs = await queue.getJobLogs(job.id);
+          expect(logs).to.be.eql({
+            logs: ['10', '27', '78', '100'],
+            count: 4,
+          });
           resolve();
         } catch (err) {
           reject(err);
@@ -227,7 +232,7 @@ describe('sandboxed process', () => {
 
     await queue.add('test', { foo: 'bar' });
 
-    await completting;
+    await completing;
     await worker.close();
   });