CHANGELOG.md

@@ -1,5 +1,13 @@
 # Changelog
 
+## v.3.13.0
+
+feat: add "preventParsingData" job option to prevent data parsing
+fix: queue.clean clean job logs as well
+fix: whenCurrentJobsFinished should wait for all jobs
+
+[Changes](https://github.com/OptimalBits/bull/compare/v3.12.1...v3.13.0)
+
 ## v.3.12.1
 
 - fix: catch errors parsing invalid progress data

PATTERNS.md

@@ -95,15 +95,15 @@ var queueQux = new Queue('quxbaz', opts);
 Redis cluster
 -------------
 
-Bull internals requires atomic operations that spans different keys. This fact breaks Redis'
-rules for cluster configurations. However it is still possible to use a cluster environment
+Bull internals require atomic operations that span different keys. This behavior breaks Redis's
+rules for cluster configurations. However, it is still possible to use a cluster environment
 by using the proper bull prefix option as a cluster "hash tag". Hash tags are used to guarantee
 that certain keys are placed in the same hash slot, read more about hash tags in the [redis cluster
-tutorial](https://redis.io/topics/cluster-tutorial).
+tutorial](https://redis.io/topics/cluster-tutorial). A hash tag is defined with brackets. I.e. a key that has a substring inside brackets will use that
+substring to determine in which hash slot the key will be placed.
 
-A hash tag is defined with brackets. I.e. a key that has a substring inside brackets will use that
-substring to determine in which hash slot the key will be placed. So to make bull compatible with
-cluster, just use a queue prefix inside brackets, for example:
+In summary, to make bull compatible with Redis cluster, use a queue prefix inside brackets.
+For example:
 
 ```js
   var queue = new Queue('cluster', {

lib/commands/cleanJobsInSet-1.lua

@@ -37,6 +37,7 @@ for _, job in ipairs(jobs) do
         redis.call("ZREM", KEYS[1], job)
       end
       redis.call("DEL", jobKey)
+      redis.call("DEL", jobKey .. ":logs")
       deletedCount = deletedCount + 1
       table.insert(deleted, job)
     end

lib/job.js

@@ -556,8 +556,10 @@ Job.prototype._saveAttempt = function(multi, err) {
 };
 
 Job.fromJSON = function(queue, json, jobId) {
-  const data = JSON.parse(json.data || '{}');
   const opts = JSON.parse(json.opts || '{}');
+  const data = opts.preventParsingData
+    ? json.data
+    : JSON.parse(json.data || '{}');
 
   const job = new Job(queue, json.name || Job.DEFAULT_JOB_NAME, data, opts);
 

lib/queue.js

@@ -670,7 +670,8 @@ interface JobOptions
   repeat: {
     tz?: string,
     endDate?: Date | string | number
-  }
+  },
+  preventParsingData: boolean;
 }
 */
 
@@ -1191,7 +1192,7 @@ Queue.prototype.whenCurrentJobsFinished = function() {
     return this.bclient.connect();
   });
 
-  return Promise.all([this.processing[0]]).then(() => {
+  return Promise.all(this.processing).then(() => {
     return forcedReconnection;
   });
 };

package.json

@@ -1,6 +1,6 @@
 {
   "name": "bull",
-  "version": "3.12.1",
+  "version": "3.13.0",
   "description": "Job manager",
   "engines": {
     "node": ">=8"
@@ -22,34 +22,34 @@
   "dependencies": {
     "cron-parser": "^2.13.0",
     "debuglog": "^1.0.0",
-    "get-port": "^5.0.0",
+    "get-port": "^5.1.1",
     "ioredis": "^4.14.1",
     "lodash": "^4.17.15",
-    "p-timeout": "^3.1.0",
-    "promise.prototype.finally": "^3.1.1",
+    "p-timeout": "^3.2.0",
+    "promise.prototype.finally": "^3.1.2",
     "semver": "^6.3.0",
-    "util.promisify": "^1.0.0",
-    "uuid": "^3.3.3"
+    "util.promisify": "^1.0.1",
+    "uuid": "^3.4.0"
   },
   "devDependencies": {
-    "@commitlint/cli": "^7.5.2",
-    "@commitlint/config-conventional": "^7.5.0",
+    "@commitlint/cli": "^7.6.1",
+    "@commitlint/config-conventional": "^7.6.0",
     "chai": "^4.2.0",
-    "coveralls": "^3.0.6",
+    "coveralls": "^3.0.9",
     "delay": "^4.3.0",
-    "eslint": "^5.12.1",
+    "eslint": "^5.16.0",
     "eslint-plugin-mocha": "^6.2.1",
     "eslint-plugin-node": "^8.0.1",
     "expect.js": "^0.3.1",
     "husky": "^1.3.1",
     "istanbul": "^0.4.5",
     "lint-staged": "^8.2.1",
-    "mocha": "^6.2.0",
+    "mocha": "^6.2.2",
     "mocha-lcov-reporter": "^1.3.0",
     "moment": "^2.24.0",
     "p-reflect": "^1.0.0",
-    "prettier": "^1.18.2",
-    "sinon": "^7.4.2"
+    "prettier": "^1.19.1",
+    "sinon": "^7.5.0"
   },
   "scripts": {
     "lint": "eslint lib test *.js",

test/test_job.js

@@ -976,4 +976,27 @@ describe('Job', () => {
         );
     });
   });
+
+  describe('.fromJSON', () => {
+    let data;
+
+    beforeEach(() => {
+      data = { foo: 'bar' };
+    });
+
+    it('should parse JSON data by default', async () => {
+      const job = await Job.create(queue, data, {});
+      const jobParsed = Job.fromJSON(queue, job.toData());
+
+      expect(jobParsed.data).to.eql(data);
+    });
+
+    it('should not parse JSON data if "preventParsingData" option is specified', async () => {
+      const job = await Job.create(queue, data, { preventParsingData: true });
+      const jobParsed = Job.fromJSON(queue, job.toData());
+      const expectedData = JSON.stringify(data);
+
+      expect(jobParsed.data).to.be(expectedData);
+    });
+  });
 });

test/test_when_current_jobs_finished.js

@@ -63,6 +63,55 @@ describe('.whenCurrentJobsFinished', () => {
     );
   });
 
+  it('should wait for all jobs to complete', async () => {
+    const queue = await utils.newQueue();
+
+    // add multiple jobs to queue
+    await queue.add({});
+    await queue.add({});
+
+    let finishJob1;
+    let finishJob2;
+
+    // wait for all jobs to be active
+    await new Promise(resolve => {
+      let callCount = 0;
+      queue.process(2, () => {
+        callCount++;
+        if (callCount === 1) {
+          return new Promise(resolve => {
+            finishJob1 = resolve;
+          });
+        }
+
+        resolve();
+        return new Promise(resolve => {
+          finishJob2 = resolve;
+        });
+      });
+    });
+
+    let isFulfilled = false;
+    const finished = queue.whenCurrentJobsFinished().then(() => {
+      isFulfilled = true;
+    });
+
+    finishJob2();
+    await delay(100);
+
+    expect(isFulfilled).to.equal(
+      false,
+      'should not fulfill until all jobs are finished'
+    );
+
+    finishJob1();
+    await delay(100);
+    expect(await finished).to.equal(
+      undefined,
+      'whenCurrentJobsFinished should resolve once all jobs are finished'
+    );
+  });
+
   it('should wait for job to fail', async () => {
     const queue = await utils.newQueue();
     await queue.add({});