CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## v.3.18.0
+
+- feat: make pause forward compatible with bullmq (#1818) (@manast)
+
+[Changes](https://github.com/OptimalBits/bull/compare/v3.17.0...v3.18.0)
+
 ## v.3.17.0
 
 - feat: better rate limiter (#1816) (@manast)

lib/commands/index.js

@@ -27,15 +27,12 @@ const _fs = {
 module.exports = (function() {
   let scripts;
 
-  return function(client) {
-    return utils.isRedisReady(client).then(() => {
-      scripts = scripts || loadScripts(__dirname);
+  return async function(client) {
+    await utils.isRedisReady(client);
+    scripts = await (scripts || loadScripts(__dirname));
 
-      return scripts.then(_scripts => {
-        return _scripts.forEach(command => {
-          return client.defineCommand(command.name, command.options);
-        });
-      });
+    return scripts.map(command => {
+      return client.defineCommand(command.name, command.options);
     });
   };
 })();

lib/commands/pause-4.lua → lib/commands/pause-5.lua

@@ -6,6 +6,7 @@
       KEYS[2] 'paused' or 'wait'
       KEYS[3] 'meta-paused'
       KEYS[4] 'paused' o 'resumed' event.
+      KEYS[5] 'meta' this key is only used in BullMQ and above.
 
       ARGV[1] 'paused' or 'resumed'
 
@@ -20,8 +21,15 @@ end
 
 if ARGV[1] == "paused" then
   rcall("SET", KEYS[3], 1)
+
+  -- for forwards compatibility
+  rcall("HSET", KEYS[5], "paused", 1)
 else
   rcall("DEL", KEYS[3])
+
+  -- for forwards compatibility
+  rcall("HDEL", KEYS[5], "paused")
+
 end
 
 rcall("PUBLISH", KEYS[4], ARGV[1])

lib/queue.js

@@ -90,20 +90,15 @@ const Queue = function Queue(name, url, opts) {
   }
 
   if (_.isString(url)) {
-    opts = _.extend(
-      {},
-      {
-        redis: redisOptsFromUrl(url)
-      },
-      opts
-    );
+    opts = {
+      ...opts,
+      redis: { ...redisOptsFromUrl(url), ...(opts || {}).redis }
+    };
   } else {
-    opts = url;
+    opts = url || {};
   }
 
-  opts = _.cloneDeep(opts || {});
-
-  if (opts && !_.isObject(opts)) {
+  if (!_.isObject(opts)) {
     throw Error('Options must be a valid object');
   }
 
@@ -142,6 +137,7 @@ const Queue = function Queue(name, url, opts) {
   delete opts.redis.keyPrefix;
 
   this.clients = [];
+
   const lazyClient = redisClientGetter(this, opts, (type, client) => {
     // bubble up Redis error events
     const handler = this.emit.bind(this, 'error');
@@ -490,10 +486,9 @@ Queue.prototype._registerEvent = function(eventName) {
 
 Queue.ErrorMessages = errors.Messages;
 
-Queue.prototype.isReady = function() {
-  return this._initializing.then(() => {
-    return this;
-  });
+Queue.prototype.isReady = async function() {
+  await this._initializing;
+  return this;
 };
 
 async function redisClientDisconnect(client) {
@@ -540,13 +535,11 @@ Queue.prototype.close = function(doNotWaitJobs) {
   }
 
   return (this.closing = this.isReady()
-    .then(
-      () => this._initializingProcess,
-      err => {
-        console.error(err);
-        isReady = false;
-      }
-    )
+    .then(this._initializingProcess)
+    .catch(err => {
+      console.error(err);
+      isReady = false;
+    })
     .finally(() => this._clearTimers())
     .then(() => isReady && this.pause(true, doNotWaitJobs))
     .then(() => {
@@ -843,17 +836,21 @@ Queue.prototype.resume = function(isLocal /* Optional */) {
     });
 };
 
-Queue.prototype.isPaused = function(isLocal) {
+Queue.prototype.isPaused = async function(isLocal) {
   if (isLocal) {
-    return Promise.resolve(!!this.paused);
+    return !!this.paused;
   } else {
-    return this.isReady()
-      .then(() => {
-        return this.client.exists(this.keys['meta-paused']);
-      })
-      .then(result => {
-        return result === 1;
-      });
+    await this.isReady();
+    const multi = this.multi();
+
+    multi.exists(this.keys['meta-paused']);
+
+    // For forward compatibility with BullMQ.
+    multi.hexists(this.toKey('meta'), 'paused');
+
+    const [[, isPaused], [, isPausedNew]] = await multi.exec();
+
+    return !!(isPaused || isPausedNew);
   }
 };
 

lib/scripts.js

@@ -50,7 +50,7 @@ const scripts = {
     }
 
     const keys = _.map(
-      [src, dst, 'meta-paused', pause ? 'paused' : 'resumed'],
+      [src, dst, 'meta-paused', pause ? 'paused' : 'resumed', 'meta'],
       name => {
         return queue.toKey(name);
       }

package.json

@@ -1,6 +1,6 @@
 {
   "name": "bull",
-  "version": "3.17.0",
+  "version": "3.18.0",
   "description": "Job manager",
   "engines": {
     "node": ">=10"

test/test_queue.js

@@ -136,7 +136,9 @@ describe('Queue', () => {
     });
 
     it('should create a queue with a redis connection string', () => {
-      const queue = new Queue('connstring', 'redis://123.4.5.67:1234/2');
+      const queue = new Queue('connstring', 'redis://123.4.5.67:1234/2', {
+        redis: { connectTimeout: 1000 }
+      });
 
       expect(queue.client.options.host).to.be.eql('123.4.5.67');
       expect(queue.eclient.options.host).to.be.eql('123.4.5.67');
@@ -148,10 +150,12 @@ describe('Queue', () => {
       expect(queue.eclient.options.db).to.be.eql(2);
 
       return queue.close();
-    }).timeout(5000);
+    });
 
     it('should create a queue with only a hostname', () => {
-      const queue = new Queue('connstring', 'redis://127.2.3.4');
+      const queue = new Queue('connstring', 'redis://127.2.3.4', {
+        redis: { connectTimeout: 1000 }
+      });
 
       expect(queue.client.options.host).to.be.eql('127.2.3.4');
       expect(queue.eclient.options.host).to.be.eql('127.2.3.4');
@@ -162,7 +166,7 @@ describe('Queue', () => {
       expect(queue.client.condition.select).to.be.eql(0);
       expect(queue.eclient.condition.select).to.be.eql(0);
 
-      queue.close().catch((/*err*/) => {
+      return queue.close().catch((/*err*/) => {
         // Swallow error.
       });
     });