Merge pull request #10 from not-empty/feature-check-complection

feature check completion

Author: leodisarli

.env.example

@@ -1,3 +1,6 @@
+AUX_REDIS_HOST=taurus-redis
+AUX_REDIS_PORT=6379
+
 REDIS_HOST=taurus-redis
 REDIS_MAX_RETRIES=null
 REDIS_PORT=6379

README.md

@@ -201,6 +201,43 @@ If you inside the container or want to run locally:
 ```sh
 node sample/multi-producer.js default 60 '{"data":"mydata"}'
 ```
+### Discover when all parallel async jobs are done
+
+If you use parallel processing with multiple workers, finding out when all jobs have completed successfully can be a complicated task due to asynchrony.
+
+To deal with this, Taurus has a functionality that uses a Redis + Lua solution to ensure that the last job in a group was executed, to use it you need to increment a list using a unique specific key when sending to each queue, and decrement this key to each queue that runs successfully.
+
+The last queue will know it is last and will allow you to perform finishing actions.
+
+You need:
+
+1) Fill in the Redis data responsible for control in the .env file (we recommend not being the same person who manages the queues)
+
+```
+AUX_REDIS_HOST=taurus-redis
+AUX_REDIS_PORT=6379
+```
+
+2) When inserting into each queue you can use the CheckCompletion class to increment each job: (If you are outside the Taurus ecosystem you can just create a key in redis with [INCR command](https://redis.io/commands/incr/) with your favorite language.)
+
+```js
+const CheckCompletion = require('../core/check-completion.js');
+...
+const checkCompletion = new CheckCompletion();
+await checkCompletion.increment(uniqueKeyToRepresenTheGroup);
+```
+
+3) When each queue has finished executing, just call the decrement command and check if it returned zero, if it returned zero, this queue is the last one of this group and you can execute any command you find necessary
+
+```js
+const CheckCompletion = require('../core/check-completion.js');
+...
+const checkCompletion = new CheckCompletion();
+const taksRemaining = await checkCompletion.decrement(sameUniqueKeyToRepresenTheGroup);
+if (taksRemaining === 0) {
+   console.log('All jobs are finished');
+}
+```
 
 ### Development
 

config/redis.js

@@ -0,0 +1,4 @@
+module.exports = {
+  auxRedisHost: process.env.AUX_REDIS_HOST || 'localhost',
+  auxRedisPort: Number(process.env.AUX_REDIS_PORT) || 6379,
+};

core/check-completion.js

@@ -0,0 +1,48 @@
+const Redis = require('ioredis');
+const configRedis = require('../config/redis');
+
+class CheckCompletion {
+  constructor() {
+    this.options = {
+        host: configRedis.auxRedisHost,
+        port: configRedis.auxRedisPort,
+    };
+  }
+
+  async increment(
+      key
+  ) {
+    const redis = new Redis(
+        this.options,
+    );
+
+    const result = await redis.incr(key);
+    console.log(`Job counter key: ${key}, increased, total jobs: ${result}`);
+
+    await redis.disconnect();
+    return result;
+  }
+
+  async decrement(
+    key
+  ) {
+    const redis = new Redis(
+      this.options,
+    );
+    const luaScript = `
+        local count = redis.call('DECR', KEYS[1])
+        return count
+    `;
+
+    const result = await redis.eval(luaScript, 1, key);
+    if (result === 0) {
+        console.log('Last job completed.');
+        await redis.del(key);
+    } else {
+        console.log(`Jobs remaining: ${result}`);
+    }
+    return result;
+  }
+}
+
+module.exports = CheckCompletion;