feat: allow to pass on args to worker function

Author: AVVS

README.md

@@ -96,12 +96,14 @@ function onJobCompleted(err, ...args) {
 
 ### Async/Await in-flight request caching
 
-#### Fanout(jobId: String, [timeout: Number], job: Function)
+#### Fanout(jobId: String, [timeout: Number], job: Function, [...args]: any[])
 
 Use `fanout(...)` method for the easiest way to handle job subscriptions where
 one actor must perform long-running job, but as soon as it's done - everyone who
 queued for the results of this job must be notified.
 
+`args` are optional args to be passed on to the job function. If you need to preserve context - use .bind
+
 Sample of code is provided to make use of this feature:
 
 ```js

__tests__/integration.js

@@ -206,6 +206,7 @@ describe('integration tests', () => {
 
   it('#fanout: multiple jobs are completed only once', () => {
     const args = ['completed'];
+    const arg1 = 'arg1';
     const job = sinon.spy(() => args);
     const onComplete = sinon.spy();
     const unexpectedError = sinon.spy();
@@ -218,14 +219,15 @@ describe('integration tests', () => {
       const id = String(idx % 3);
 
       try {
-        onComplete(id, await queueManager.dlock.fanout(id, job));
+        onComplete(id, await queueManager.dlock.fanout(id, job, arg1));
       } catch (e) {
         unexpectedError(e);
       }
     })
       .delay(100)
       .then(() => {
         assert.equal(job.callCount, 3);
+        assert.equal(job.withArgs(arg1).callCount, 3);
         assert.equal(onComplete.withArgs('0', args).callCount, 4);
         assert.equal(onComplete.withArgs('1', args).callCount, 3);
         assert.equal(onComplete.withArgs('2', args).callCount, 3);
@@ -238,6 +240,7 @@ describe('integration tests', () => {
     const job = sinon.spy(async () => {
       await Promise.delay(3000);
     });
+    const arg1 = 'arg1';
     const onComplete = sinon.spy();
     const timeoutError = sinon.spy();
     const unexpectedError = sinon.spy();
@@ -246,7 +249,7 @@ describe('integration tests', () => {
       const id = String(idx % 3);
 
       try {
-        const result = await queueManager.dlock.fanout(id, 1500, job);
+        const result = await queueManager.dlock.fanout(id, 1500, job, arg1);
         onComplete(result);
       } catch (e) {
         if (e.message === 'queue-no-response') {
@@ -258,6 +261,7 @@ describe('integration tests', () => {
     });
 
     assert.equal(job.callCount, 3);
+    assert.equal(job.withArgs(arg1).callCount, 3);
     assert.equal(onComplete.callCount, 0);
     assert.equal(timeoutError.callCount, 10);
     assert.equal(timeoutError.withArgs(sinon.match({ message: 'queue-no-response' })).callCount, 10);

package.json

@@ -44,19 +44,19 @@
     "serialize-error": "^7.0.1"
   },
   "devDependencies": {
-    "@babel/cli": "^7.10.1",
-    "@babel/core": "^7.10.2",
+    "@babel/cli": "^7.10.3",
+    "@babel/core": "^7.10.3",
     "@babel/plugin-transform-strict-mode": "^7.10.1",
     "@makeomatic/deploy": "^10.2.1",
     "codecov": "^3.7.0",
     "cross-env": "^7.0.2",
-    "eslint": "^7.1.0",
-    "eslint-config-makeomatic": "^5.0.0",
-    "eslint-plugin-import": "^2.20.2",
+    "eslint": "^7.3.1",
+    "eslint-config-makeomatic": "^5.0.2",
+    "eslint-plugin-import": "^2.21.2",
     "eslint-plugin-promise": "^4.2.1",
     "eslint-plugin-unicorn": "^20.1.0",
     "ioredis": "^4.17.3",
-    "jest": "^26.0.1",
+    "jest": "^26.1.0",
     "sinon": "^9.0.2"
   },
   "engine": {

src/distributed-callback-queue.js

@@ -185,15 +185,33 @@ class DistributedCallbackQueue {
    * @param {String} suffix job key
    * @param {Number} [timeout] when job is considered to be failed and error is returned instead
    * @param {Function} worker async job to be performed by the party that gets the lock
+   * @param {Mixed[]} [args] passed on to worker as args
    *
    * @return {Function} job handler that must be invoked with a worker that returns a promise
    */
   async fanout(suffix, ...props) {
-    // verify we've got correct args in
-    assert(props.length >= 1 && props.length <= 2);
-    const worker = props.pop();
-    const timeout = props.pop();
+    const propsAmount = props.length;
+    assert(propsAmount >= 1, 'must have at least job function passed');
+
+    // eslint-disable-next-line prefer-const
+    let [timeout, worker, ...workerArgs] = props;
+
+    // in case of 1 arg
+    switch (propsAmount) {
+      case 1:
+        worker = timeout;
+        timeout = undefined;
+        break;
+      default:
+        if (typeof timeout === 'function') {
+          workerArgs.unshift(worker);
+          worker = timeout;
+          timeout = undefined;
+        }
+    }
+
     assert(typeof worker === 'function', 'ensure that you pass a function as a worker');
+    assert(typeof timeout === 'number' || typeof timeout === 'undefined', 'invalid timeout value');
 
     // allows us to reject-and-halt (eg. on timeout) even if the #push'ed lock has not yet been acquired
     let jobAbortReject;
@@ -244,7 +262,7 @@ class DistributedCallbackQueue {
     }
 
     // wrap so that we have concept of "cancelling" work
-    const performWork = worker();
+    const performWork = worker(...workerArgs);
 
     try {
       const result = await Promise.race([