feat: support job.discard function in sandboxed processors

* Implement Job.prototype.isDiscarded

lib/job.js

@@ -419,6 +419,10 @@ Job.prototype.isStuck = function() {
   });
 };
 
+Job.prototype.isDiscarded = function() {
+  return this._discarded;
+};
+
 Job.prototype.getState = function() {
   const fns = [
     { fn: 'isCompleted', state: 'completed' },

lib/process/master.js

@@ -183,5 +183,13 @@ function wrapJob(job) {
     });
     job.data = data;
   };
+  /*
+   * Emulate the real job `discard` function.
+   */
+  job.discard = function() {
+    process.send({
+      cmd: 'discard'
+    });
+  };
   return job;
 }

lib/process/sandbox.js

@@ -33,6 +33,9 @@ module.exports = function(processFile, childPool) {
             case 'update':
               job.update(msg.value);
               break;
+            case 'discard':
+              job.discard();
+              break;
             case 'log':
               job.log(msg.value);
               break;

test/fixtures/fixture_processor_discard.js

@@ -0,0 +1,14 @@
+/**
+ * A processor file to be used in tests.
+ *
+ */
+'use strict';
+
+const delay = require('delay');
+
+module.exports = function(job) {
+  return delay(500).then(() => {
+    job.discard();
+    throw new Error('Manually discarded processor');
+  });
+};

test/test_sandboxed_process.js

@@ -282,6 +282,27 @@ describe('sandboxed process', () => {
     queue.add({ foo: 'bar' });
   });
 
+  it('should process, discard and fail without retry', done => {
+    queue.process(__dirname + '/fixtures/fixture_processor_discard.js');
+
+    queue.on('failed', (job, err) => {
+      try {
+        expect(job.data).eql({ foo: 'bar' });
+        expect(job.isDiscarded()).to.be.true;
+        expect(job.failedReason).eql('Manually discarded processor');
+        expect(err.message).eql('Manually discarded processor');
+        expect(err.stack).include('fixture_processor_discard.js');
+        expect(Object.keys(queue.childPool.retained)).to.have.lengthOf(0);
+        expect(queue.childPool.getAllFree()).to.have.lengthOf(1);
+        done();
+      } catch (err) {
+        done(err);
+      }
+    });
+
+    queue.add({ foo: 'bar' });
+  });
+
   it('should process and fail', done => {
     queue.process(__dirname + '/fixtures/fixture_processor_fail.js');