job-allocation - npm Package Compare versions

Comparing version 2.0.5 to 2.0.6

lib/src/helloWorldRecovery.js

@@ -6,2 +6,3 @@ "use strict";
 const recoveryUnfinishedJobs_1 = require("./helpers/recoveryUnfinishedJobs");
+const startJobActivity_1 = require("./helpers/startJobActivity");
 const remoteQueue = new index_1.JARedisRemoteQueue({
@@ -12,10 +13,15 @@ name: 'redis-key-queue',
 });
-const remoteQueueTemp = remoteQueue.factoryNestedRemoteQueue('temp');
+const remoteQueueActiveJobs = remoteQueue.factoryNestedRemoteQueue('temp');
 const action = async (job) => {
-    await (0, delay_1.delay)(1000);
+    const stopJobActivity = (0, startJobActivity_1.startJobActivity)({
+        queueJobs: remoteQueueActiveJobs,
+        job,
+        interval: 100,
+    });
+    await (0, delay_1.delay)(6000);
     console.log(`Hello world, ${job.data.name}`);
-    // await remoteQueueTemp.removeJob(job);
+    await stopJobActivity();
 };
 const worker = new index_1.JAWorker({
-    getJob: () => remoteQueue.popMove(remoteQueueTemp.key),
+    getJob: () => remoteQueue.popMove(remoteQueueActiveJobs.key),
     action,
@@ -29,19 +35,16 @@ concurrency: 2,
     await remoteQueue.clear();
-    await remoteQueueTemp.clear();
+    await remoteQueueActiveJobs.clear();
+    console.log('remoteQueue', await remoteQueue.count(), await remoteQueueActiveJobs.count());
+    worker.start();
     await remoteQueue.add({ name: 'foo' });
-    await (0, delay_1.delay)(1000);
-    await remoteQueue.add({ name: 'bar' });
+    // await remoteQueue.add({ name: 'bar' });
     setInterval(() => {
         (0, recoveryUnfinishedJobs_1.recoveryUnfinishedJobs)({
             queueJobs: remoteQueue,
-            queueRunningJobs: remoteQueueTemp,
-            retryTimeout: 4000,
+            queueRunningJobs: remoteQueueActiveJobs,
+            retryTimeout: 1000,
         });
-    }, 1000);
+    }, 500);
 };
 /**
- * Run in nodes that perform tasks
- */
-worker.start();
-/**
  * Run in the control node
@@ -48,0 +51,0 @@ */

lib/src/helpers/recoveryUnfinishedJobs.js

@@ -8,36 +8,41 @@ "use strict";
         const dateNow = Date.now();
-        const [jobStr] = await redisClient.lrange(queueRunningJobs.key, 0, 0);
-        if (jobStr) {
-            const job = JSON.parse(jobStr);
-            const lastRetryTimestamp = job.lastRetryTimestamp ?? job.onCreated;
-            const jobRetryTimeout = job.retryTimeout ?? retryTimeout;
-            const jobRetryCount = job.retryCount ?? retryCount;
-            const isRetryJob = Boolean(jobRetryCount > 0
-                && jobRetryTimeout
-                && (lastRetryTimestamp + jobRetryTimeout) < dateNow);
-            const isRemoveJob = Boolean(isRetryJob
-                || jobRetryCount <= 0
-                || !jobRetryTimeout);
-            if (isRemoveJob) {
-                let multi = await redisClient
-                    .multi()
-                    .lrem(queueRunningJobs.key, 1, jobStr);
-                if (isRetryJob) {
-                    job.lastRetryTimestamp = dateNow;
-                    job.retryCount = jobRetryCount - 1;
-                    multi = multi.rpush(queueJobs.key, JSON.stringify(job));
+        try {
+            const jobStr = await redisClient.lindex(queueRunningJobs.key, 0);
+            if (jobStr) {
+                const job = JSON.parse(jobStr);
+                const lastRetryTimestamp = job.lastRetryTimestamp ?? job.onCreated;
+                const jobRetryTimeout = job.retryTimeout ?? retryTimeout;
+                const jobRetryCount = job.retryCount ?? retryCount;
+                const isRetryJob = Boolean(jobRetryCount > 0
+                    && jobRetryTimeout
+                    && (lastRetryTimestamp + jobRetryTimeout) < dateNow);
+                const isRemoveJob = Boolean(isRetryJob
+                    || jobRetryCount <= 0
+                    || !jobRetryTimeout);
+                if (isRemoveJob) {
+                    let multi = await redisClient
+                        .multi()
+                        .lrem(queueRunningJobs.key, 1, jobStr);
+                    if (isRetryJob) {
+                        job.lastRetryTimestamp = dateNow;
+                        job.retryCount = jobRetryCount - 1;
+                        multi = multi.rpush(queueJobs.key, JSON.stringify(job));
+                    }
+                    await multi.exec();
+                    await (0, exports.recoveryUnfinishedJobs)({
+                        queueJobs,
+                        queueRunningJobs,
+                        retryTimeout,
+                        retryCount,
+                    });
                 }
-                await multi.exec();
-                await (0, exports.recoveryUnfinishedJobs)({
-                    queueJobs,
-                    queueRunningJobs,
-                    retryTimeout,
-                    retryCount,
-                });
+                else {
+                    await Promise.reject(new Error('unwatch'));
+                }
             }
             else {
-                await redisClient.unwatch();
+                await Promise.reject(new Error('unwatch'));
             }
         }
-        else {
+        catch (e) {
             await redisClient.unwatch();
@@ -44,0 +49,0 @@ }

lib/src/index.d.ts

@@ -10,1 +10,2 @@ export * from './worker/types';
 export { recoveryUnfinishedJobs } from './helpers/recoveryUnfinishedJobs';
+export { startJobActivity } from './helpers/startJobActivity';

lib/src/index.js

@@ -17,3 +17,3 @@ "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-exports.recoveryUnfinishedJobs = exports.generateStorageKey = exports.createJAJob = exports.createWaitJobsCompleted = exports.createWorkerWatch = exports.JARedisRemoteQueue = exports.JAWorker = void 0;
+exports.startJobActivity = exports.recoveryUnfinishedJobs = exports.generateStorageKey = exports.createJAJob = exports.createWaitJobsCompleted = exports.createWorkerWatch = exports.JARedisRemoteQueue = exports.JAWorker = void 0;
 __exportStar(require("./worker/types"), exports);
@@ -35,2 +35,4 @@ __exportStar(require("./remoteQueue/types"), exports);
 Object.defineProperty(exports, "recoveryUnfinishedJobs", { enumerable: true, get: function () { return recoveryUnfinishedJobs_1.recoveryUnfinishedJobs; } });
+var startJobActivity_1 = require("./helpers/startJobActivity");
+Object.defineProperty(exports, "startJobActivity", { enumerable: true, get: function () { return startJobActivity_1.startJobActivity; } });
 //# sourceMappingURL=index.js.map

package.json

 {
   "name": "job-allocation",
-  "version": "2.0.5",
+  "version": "2.0.6",
   "description": "Library for distributing jobs between nodes",
@@ -11,3 +11,3 @@ "main": "lib/src/index.js",
   "scripts": {
-    "dev": "nodemon src/helloWorld.ts",
+    "dev": "nodemon src/helloWorldRecovery.ts",
     "build": "tsc"
@@ -14,0 +14,0 @@ },

No alert changes

Improved metrics

Total package byte prevSize

76857

increased by6.37%

Number of package files

84

increased by7.69%

Lines of code

1158

increased by5.66%

No dependency changes