Merge pull request #14 from knation/node-redis-v4

Improve internal handling of redis clients

Author: knation

.github/workflows/node.js.yml

@@ -16,7 +16,7 @@ jobs:
 
     strategy:
       matrix:
-        node-version: [12.x, 14.x, 16.x]
+        node-version: [18.x]
         # See supported Node.js release schedule at https://nodejs.org/en/about/releases/
 
     steps:
@@ -28,10 +28,3 @@ jobs:
         cache: 'npm'
     - run: npm ci
     - run: npm run build --if-present
-    - run: npm run coverage
-
-    # Save coverage report in Coveralls
-    - name: Coveralls
-      uses: coverallsapp/github-action@master
-      with:
-        github-token: ${{ secrets.GITHUB_TOKEN }}

README.md

@@ -95,7 +95,7 @@ try {
   this.add(5000, data);
 }
 
-// Don't forget to clean up later: `dt.stop()`
+// Don't forget to clean up later: `dt.stop()` or `dt.close()`, as applicable
 ```
 
 
@@ -113,21 +113,25 @@ The constructor takes a single object. Properties are as follows:
 | `callback`               | The function to call when tasks are due. <br><br>When a task is due or past-due, your callback method is called asynchronously, passing the `data` you provided when adding, the generated `taskId`, and the time (in ms) that the task was due.<br><br>The context of `this` is the `DelayedTasks` object.                                                                                                                           | Yes      |         |
 | `options.pollIntervalMs` | How often to poll redis for tasks due (in milliseconds). The shorter the interval, the sooner after being due a task will be processed, but the more load in redis. | No       | 1000    |
 
-<sup>1</sup> This module uses [`node-redis`](https://github.com/redis/node-redis) version 4 under the hood in legacy mode.
+<sup>1</sup> This module uses [`node-redis`](https://github.com/redis/node-redis) version 4 under the hood in legacy mode. If you provide your own client, you're responsible for connecting, disconnecting, and creating an error handler.
 
 ### Connect
 
-Before doing anything, call `await dt.connect()` to connect to redis.
+If you provided an object with options for connecting to redis, you must call `await dt.connect()` to connect to redis before doing anything else. If you've provided an existing, connected redis client, this is not necessary.
 
 ### start / stop polling
 
-To begin polling for tasks, call `dt.start()`. Call `dt.stop()` to stop future polling.
+To begin polling for tasks, call `dt.start()`. This returns a boolean with the status of starting. If `false`, it's because the redis client hasn't been connected yet. If this was a self-supplied client, call `await client.connect()`. Otherwise, call `await dt.connect()` to create the connection.
+
+Call `dt.stop()` to stop future polling.
 
 ### close()
 
-Calling `dt.close()` will stop polling and close the redis client being used.
+Calling `await dt.close()` will stop polling. If a new redis client was created for the object instance (that is, it was passed an object of configuration details), that redis client will be closed, too (using `disconnect()` to abort pending requests). If you passed an existing redis client to the constructor, it will be left open and you'll have to close it yourself when you're finished with it.
 
-**CAUTION:** If you passed your own redis client in the constructor, that client will be closed with this command. If you just want to stop polling, but leave the connection open, call `dt.stop()` instead.
+This returns a promise that resolves when the client connection is confirmed closed.
+
+If you just want to stop polling, but leave the connection open, call `dt.stop()` instead.
 
 ### add(_delayMs_, _data_)
 
@@ -151,7 +155,7 @@ To force a poll outside of the poll interval, call `dt.poll()`. This should be u
 
 ## Testing
 
-Start a local redis server on port 6379. You can run `docker-compose up` to launch one from this repo. Once redis is running, run `npm test` or `npm coverage`.
+The test suite requires a local redis server on port 6379. You can run `docker-compose up` to launch one from this repo. Once redis is running, run `npm test` or `npm coverage`.
 
 ## Notes
 
@@ -181,5 +185,7 @@ catch everything, including redis errors.
 
 * Better handling of watch conflicts in `dt.poll`. Right now it just quits, but if this happens a lot, nothing would end up getting processed.
 
+* Find a redis mock that works with `node-redis` v4 and later versions of redis server.
+
 ## License
 MIT License

index.js

@@ -17,18 +17,24 @@ class DelayedTasks {
       throw new TypeError('Invalid queue ID specified');
     }
 
+    // Will be set to true if the redis instance is self-contained
+    this.selfContainedResis = false;
+
     if (typeof settings.redis?.connect === 'function') {
       this.redisClient = settings.redis;
+
     } else if (typeof settings.redis === 'object') {
       settings.redis.legacyMode = true;
       this.redisClient = redis.createClient(settings.redis);
+      this.selfContainedResis = true;
+      this.redisClient.on("error", function(error) {
+        // todo: do something with this
+      });
+
     } else {
       throw new TypeError('Invalid redis connection options');
     }
 
-    this.redisClient.on("error", function(error) {
-      // todo: do something with this
-    });
 
     // Callback function for all delayed tasks
     if (typeof settings.callback === 'function') {
@@ -54,14 +60,30 @@ class DelayedTasks {
   }
 
   connect() {
-    return this.redisClient.connect();
+    if (!this.redisClient.isReady) {
+      const p = new Promise(resolve => this.redisClient.on('ready', resolve));
+
+      this.redisClient.connect();
+
+      return p;
+
+    } else {
+      // Already connected -- nothing to do and no failure
+      return Promise.resolve();
+    }
   }
 
   /**
    * Start polling.
    */
   start() {
-    this.pollIntervalId = setInterval(this.poll.bind(this), this.pollIntervalMs);
+    if (this.redisClient.isReady) {
+      this.pollIntervalId = setInterval(this.poll.bind(this), this.pollIntervalMs);
+      return true;
+
+    } else {
+      return false;
+    }
   }
 
   /**
@@ -73,11 +95,16 @@ class DelayedTasks {
   }
 
   /**
-   * Closes up shop.
+   * Closes up shop. If the redis instance is self contained (it was created,
+   * just for this object instance), it will be deleted.
    */
-  close() {
+  async close() {
     this.stop();
-    this.redisClient.quit();
+
+    if (this.selfContainedResis && this.redisClient.isReady) {
+      await this.redisClient.disconnect();
+      this.redisClient = null;
+    }
   }
 
   /**
@@ -102,12 +129,19 @@ class DelayedTasks {
               .exec((execError, results) => {
                 /* istanbul ignore next */
                 if (execError) {
-                  resolve(0);
-                  return;
+                  if (execError instanceof redis.WatchError) {
+                    /**
+                     * If execError is a "WatchError", it means that a concurrent client
+                     * changed the key while we were processing it and thus
+                     * the execution of the MULTI command was not performed. We'll fail
+                     * silently as those jobs will be picked up on the next `poll()`
+                     */
+                    return resolve(0);
+                  } else {
+                    return reject(execError)
+                  }
                 }
 
-                // Success, either that the update was made, or the key changed
-
                 if (results && results[0] !== null) {
                   // Process tasks
                   tasks
@@ -116,16 +150,11 @@ class DelayedTasks {
                 }
 
                 resolve((!results || results[0] === null) ? 0 : results[0]);
-
-                /**
-                 * If execError is a "WatchError", it means that a concurrent client
-                 * changed the key while we were processing it and thus
-                 * the execution of the MULTI command was not performed.
-                 */
               });
 
             } else {
               // No changes to make
+              this.redisClient.unwatch();
               resolve(0);
             }
         });