bullmq - npm Package Compare versions

Comparing version 5.2.1 to 5.3.0

dist/cjs/classes/job.js

@@ -177,2 +177,5 @@ "use strict";
         }
+        if (json.pb) {
+            job.processedBy = json.pb;
+        }
         return job;
@@ -179,0 +182,0 @@ }

dist/cjs/classes/queue-events.js

@@ -54,2 +54,3 @@ "use strict";
                 const client = await this.client;
+                // TODO: Planed for deprecation as it has no really a use case
                 try {
@@ -56,0 +57,0 @@ await client.client('SETNAME', this.clientName(utils_1.QUEUE_EVENT_SUFFIX));

dist/cjs/classes/queue-getters.js

@@ -311,7 +311,7 @@ /*eslint-env node */
     }
-    async baseGetClients(suffix) {
+    async baseGetClients(matcher) {
         const client = await this.client;
         const clients = (await client.client('LIST'));
         try {
-            const list = this.parseClientList(clients, suffix);
+            const list = this.parseClientList(clients, matcher);
             return list;
@@ -334,3 +334,8 @@ }
     getWorkers() {
-        return this.baseGetClients(utils_1.WORKER_SUFFIX);
+        const unnamedWorkerClientName = `${this.clientName()}`;
+        const namedWorkerClientName = `${this.clientName()}:w:`;
+        const matcher = (name) => name &&
+            (name === unnamedWorkerClientName ||
+                name.startsWith(namedWorkerClientName));
+        return this.baseGetClients(matcher);
     }
@@ -341,6 +346,9 @@ /**
      *
+     * @deprecated do not use this method, it will be removed in the future.
+     *
      * @returns - Returns an array with queue events info.
      */
     async getQueueEvents() {
-        return this.baseGetClients(utils_1.QUEUE_EVENT_SUFFIX);
+        const clientName = `${this.clientName()}${utils_1.QUEUE_EVENT_SUFFIX}`;
+        return this.baseGetClients((name) => name === clientName);
     }
@@ -386,3 +394,3 @@ /**
     }
-    parseClientList(list, suffix = '') {
+    parseClientList(list, matcher) {
         const lines = list.split('\n');
@@ -400,4 +408,5 @@ const clients = [];
             const name = client['name'];
-            if (name && name === `${this.clientName()}${suffix ? `${suffix}` : ''}`) {
+            if (matcher(name)) {
                 client['name'] = this.name;
+                client['rawname'] = name;
                 clients.push(client);
@@ -404,0 +413,0 @@ }

dist/cjs/classes/scripts.js

@@ -606,3 +606,3 @@ /**
     }
-    async moveToActive(client, token) {
+    async moveToActive(client, token, name) {
         const opts = this.queue.opts;
@@ -630,2 +630,3 @@ const queueKeys = this.queue.keys;
                 limiter: opts.limiter,
+                name,
             }),
@@ -632,0 +633,0 @@ ];

dist/cjs/classes/worker.js

@@ -72,6 +72,9 @@ "use strict";
                 }
-                const mainFile = this.opts.useWorkerThreads
-                    ? 'main-worker.js'
-                    : 'main.js';
-                let mainFilePath = path.join(path.dirname(module.filename), `${mainFile}`);
+                // Separate paths so that bundling tools can resolve dependencies easier
+                const dirname = path.dirname(module.filename || __filename);
+                const workerThreadsMainFile = path.join(dirname, 'main-worker.js');
+                const spawnProcessMainFile = path.join(dirname, 'main.js');
+                let mainFilePath = this.opts.useWorkerThreads
+                    ? workerThreadsMainFile
+                    : spawnProcessMainFile;
                 try {
@@ -81,2 +84,5 @@ fs.statSync(mainFilePath); // would throw if file not exists
                 catch (_) {
+                    const mainFile = this.opts.useWorkerThreads
+                        ? 'main-worker.js'
+                        : 'main.js';
                     mainFilePath = path.join(process.cwd(), `dist/cjs/classes/${mainFile}`);
@@ -95,3 +101,3 @@ fs.statSync(mainFilePath);
         }
-        const connectionName = this.clientName(utils_1.WORKER_SUFFIX);
+        const connectionName = this.clientName() + (this.opts.name ? `:w:${this.opts.name}` : '');
         this.blockingConnection = new redis_connection_1.RedisConnection((0, utils_1.isRedisInstance)(opts.connection)
@@ -243,3 +249,3 @@ ? opts.connection.duplicate({ connectionName })
                 if (this.blockUntil <= 0 || this.blockUntil - Date.now() < 10) {
-                    return this.moveToActive(client, token);
+                    return this.moveToActive(client, token, this.opts.name);
                 }
@@ -264,3 +270,3 @@ }
             }
-            return this.moveToActive(client, token);
+            return this.moveToActive(client, token, this.opts.name);
         }
@@ -276,4 +282,4 @@ }
     }
-    async moveToActive(client, token) {
-        const [jobData, id, limitUntil, delayUntil] = await this.scripts.moveToActive(client, token);
+    async moveToActive(client, token, name) {
+        const [jobData, id, limitUntil, delayUntil] = await this.scripts.moveToActive(client, token, name);
         this.updateDelays(limitUntil, delayUntil);
@@ -280,0 +286,0 @@ return this.nextJobFromJobData(jobData, id, token);

dist/cjs/scripts/moveToActive-11.js

@@ -116,2 +116,6 @@ "use strict";
   end
+  if opts['name'] then
+    -- Set "processedBy" field to the worker name
+    rcall("HSET", jobKey, "pb", opts['name'])
+  end
   rcall("XADD", eventStreamKey, "*", "event", "active", "jobId", jobId, "prev", "waiting")
@@ -118,0 +122,0 @@ rcall("HSET", jobKey, "processedOn", processedOn)

dist/cjs/scripts/moveToFinished-14.js

@@ -302,2 +302,6 @@ "use strict";
   end
+  if opts['name'] then
+    -- Set "processedBy" field to the worker name
+    rcall("HSET", jobKey, "pb", opts['name'])
+  end
   rcall("XADD", eventStreamKey, "*", "event", "active", "jobId", jobId, "prev", "waiting")
@@ -304,0 +308,0 @@ rcall("HSET", jobKey, "processedOn", processedOn)

dist/cjs/utils.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-exports.QUEUE_EVENT_SUFFIX = exports.WORKER_SUFFIX = exports.errorToJSON = exports.parseObjectValues = exports.isRedisVersionLowerThan = exports.childSend = exports.asyncSend = exports.isNotConnectionError = exports.DELAY_TIME_1 = exports.DELAY_TIME_5 = exports.clientCommandMessageReg = exports.getParentKey = exports.removeAllQueueData = exports.decreaseMaxListeners = exports.increaseMaxListeners = exports.isRedisCluster = exports.isRedisInstance = exports.delay = exports.array2obj = exports.isEmpty = exports.lengthInUtf8Bytes = exports.tryCatch = exports.errorObject = void 0;
+exports.QUEUE_EVENT_SUFFIX = exports.errorToJSON = exports.parseObjectValues = exports.isRedisVersionLowerThan = exports.childSend = exports.asyncSend = exports.isNotConnectionError = exports.DELAY_TIME_1 = exports.DELAY_TIME_5 = exports.clientCommandMessageReg = exports.getParentKey = exports.removeAllQueueData = exports.decreaseMaxListeners = exports.increaseMaxListeners = exports.isRedisCluster = exports.isRedisInstance = exports.delay = exports.array2obj = exports.isEmpty = exports.lengthInUtf8Bytes = exports.tryCatch = exports.errorObject = void 0;
 const ioredis_1 = require("ioredis");
@@ -168,4 +168,3 @@ // eslint-disable-next-line @typescript-eslint/ban-ts-comment
 exports.errorToJSON = errorToJSON;
-exports.WORKER_SUFFIX = '';
 exports.QUEUE_EVENT_SUFFIX = ':qe';
 //# sourceMappingURL=utils.js.map

dist/esm/classes/job.d.ts

@@ -99,2 +99,6 @@ import { ChainableCommander } from 'ioredis';
     token?: string;
+    /**
+     * The worker name that is processing or processed this job.
+     */
+    processedBy?: string;
     protected toKey: (type: string) => string;
@@ -101,0 +105,0 @@ protected discarded: boolean;

dist/esm/classes/job.js

@@ -174,2 +174,5 @@ import { __rest } from "tslib";
         }
+        if (json.pb) {
+            job.processedBy = json.pb;
+        }
         return job;
@@ -176,0 +179,0 @@ }

dist/esm/classes/queue-events.js

@@ -51,2 +51,3 @@ import { __rest } from "tslib";
                 const client = await this.client;
+                // TODO: Planed for deprecation as it has no really a use case
                 try {
@@ -53,0 +54,0 @@ await client.client('SETNAME', this.clientName(QUEUE_EVENT_SUFFIX));

dist/esm/classes/queue-getters.d.ts

@@ -187,2 +187,4 @@ import { QueueBase } from './queue-base';
      *
+     * @deprecated do not use this method, it will be removed in the future.
+     *
      * @returns - Returns an array with queue events info.
@@ -189,0 +191,0 @@ */

dist/esm/classes/queue-getters.js

@@ -5,3 +5,3 @@ /*eslint-env node */
 import { Job } from './job';
-import { clientCommandMessageReg, QUEUE_EVENT_SUFFIX, WORKER_SUFFIX, } from '../utils';
+import { clientCommandMessageReg, QUEUE_EVENT_SUFFIX } from '../utils';
 /**
@@ -310,7 +310,7 @@ *
     }
-    async baseGetClients(suffix) {
+    async baseGetClients(matcher) {
         const client = await this.client;
         const clients = (await client.client('LIST'));
         try {
-            const list = this.parseClientList(clients, suffix);
+            const list = this.parseClientList(clients, matcher);
             return list;
@@ -333,3 +333,8 @@ }
     getWorkers() {
-        return this.baseGetClients(WORKER_SUFFIX);
+        const unnamedWorkerClientName = `${this.clientName()}`;
+        const namedWorkerClientName = `${this.clientName()}:w:`;
+        const matcher = (name) => name &&
+            (name === unnamedWorkerClientName ||
+                name.startsWith(namedWorkerClientName));
+        return this.baseGetClients(matcher);
     }
@@ -340,6 +345,9 @@ /**
      *
+     * @deprecated do not use this method, it will be removed in the future.
+     *
      * @returns - Returns an array with queue events info.
      */
     async getQueueEvents() {
-        return this.baseGetClients(QUEUE_EVENT_SUFFIX);
+        const clientName = `${this.clientName()}${QUEUE_EVENT_SUFFIX}`;
+        return this.baseGetClients((name) => name === clientName);
     }
@@ -385,3 +393,3 @@ /**
     }
-    parseClientList(list, suffix = '') {
+    parseClientList(list, matcher) {
         const lines = list.split('\n');
@@ -399,4 +407,5 @@ const clients = [];
             const name = client['name'];
-            if (name && name === `${this.clientName()}${suffix ? `${suffix}` : ''}`) {
+            if (matcher(name)) {
                 client['name'] = this.name;
+                client['rawname'] = name;
                 clients.push(client);
@@ -403,0 +412,0 @@ }

dist/esm/classes/scripts.d.ts

@@ -85,3 +85,3 @@ /**
     reprocessJob<T = any, R = any, N extends string = string>(job: MinimalJob<T, R, N>, state: 'failed' | 'completed'): Promise<void>;
-    moveToActive(client: RedisClient, token: string): Promise<any[]>;
+    moveToActive(client: RedisClient, token: string, name?: string): Promise<any[]>;
     promote(jobId: string): Promise<void>;
@@ -88,0 +88,0 @@ /**

dist/esm/classes/scripts.js

@@ -604,3 +604,3 @@ /**
     }
-    async moveToActive(client, token) {
+    async moveToActive(client, token, name) {
         const opts = this.queue.opts;
@@ -628,2 +628,3 @@ const queueKeys = this.queue.keys;
                 limiter: opts.limiter,
+                name,
             }),
@@ -630,0 +631,0 @@ ];

dist/esm/classes/worker.d.ts

@@ -147,3 +147,3 @@ /// <reference types="node" />
     rateLimit(expireTimeMs: number): Promise<void>;
-    protected moveToActive(client: RedisClient, token: string): Promise<Job<DataType, ResultType, NameType>>;
+    protected moveToActive(client: RedisClient, token: string, name?: string): Promise<Job<DataType, ResultType, NameType>>;
     private waitForJob;
@@ -150,0 +150,0 @@ /**

dist/esm/classes/worker.js

@@ -7,3 +7,3 @@ import * as fs from 'fs';
 import { AbortController } from 'node-abort-controller';
-import { delay, DELAY_TIME_1, isNotConnectionError, isRedisInstance, WORKER_SUFFIX, } from '../utils';
+import { delay, DELAY_TIME_1, isNotConnectionError, isRedisInstance, } from '../utils';
 import { QueueBase } from './queue-base';
@@ -70,6 +70,9 @@ import { Repeat } from './repeat';
                 }
-                const mainFile = this.opts.useWorkerThreads
-                    ? 'main-worker.js'
-                    : 'main.js';
-                let mainFilePath = path.join(path.dirname(module.filename), `${mainFile}`);
+                // Separate paths so that bundling tools can resolve dependencies easier
+                const dirname = path.dirname(module.filename || __filename);
+                const workerThreadsMainFile = path.join(dirname, 'main-worker.js');
+                const spawnProcessMainFile = path.join(dirname, 'main.js');
+                let mainFilePath = this.opts.useWorkerThreads
+                    ? workerThreadsMainFile
+                    : spawnProcessMainFile;
                 try {
@@ -79,2 +82,5 @@ fs.statSync(mainFilePath); // would throw if file not exists
                 catch (_) {
+                    const mainFile = this.opts.useWorkerThreads
+                        ? 'main-worker.js'
+                        : 'main.js';
                     mainFilePath = path.join(process.cwd(), `dist/cjs/classes/${mainFile}`);
@@ -93,3 +99,3 @@ fs.statSync(mainFilePath);
         }
-        const connectionName = this.clientName(WORKER_SUFFIX);
+        const connectionName = this.clientName() + (this.opts.name ? `:w:${this.opts.name}` : '');
         this.blockingConnection = new RedisConnection(isRedisInstance(opts.connection)
@@ -241,3 +247,3 @@ ? opts.connection.duplicate({ connectionName })
                 if (this.blockUntil <= 0 || this.blockUntil - Date.now() < 10) {
-                    return this.moveToActive(client, token);
+                    return this.moveToActive(client, token, this.opts.name);
                 }
@@ -262,3 +268,3 @@ }
             }
-            return this.moveToActive(client, token);
+            return this.moveToActive(client, token, this.opts.name);
         }
@@ -274,4 +280,4 @@ }
     }
-    async moveToActive(client, token) {
-        const [jobData, id, limitUntil, delayUntil] = await this.scripts.moveToActive(client, token);
+    async moveToActive(client, token, name) {
+        const [jobData, id, limitUntil, delayUntil] = await this.scripts.moveToActive(client, token, name);
         this.updateDelays(limitUntil, delayUntil);
@@ -278,0 +284,0 @@ return this.nextJobFromJobData(jobData, id, token);

dist/esm/interfaces/job-json.d.ts

@@ -20,2 +20,3 @@ import { RedisJobOptions } from '../types';
     repeatJobKey?: string;
+    processedBy?: string;
 }
@@ -41,2 +42,3 @@ export interface JobJsonRaw {
     ats?: string;
+    pb?: string;
 }

dist/esm/interfaces/worker-options.d.ts

@@ -13,2 +13,8 @@ import { Job } from '../classes/job';
     /**
+     * Optional worker name. The name will be stored on every job
+     * processed by this worker instance, and can be used to monitor
+     * which worker is processing or has processed a given job.
+     */
+    name?: string;
+    /**
      * Condition to start processor at instance creation.
@@ -15,0 +21,0 @@ *

dist/esm/scripts/moveToActive-11.js

@@ -113,2 +113,6 @@ const content = `--[[
   end
+  if opts['name'] then
+    -- Set "processedBy" field to the worker name
+    rcall("HSET", jobKey, "pb", opts['name'])
+  end
   rcall("XADD", eventStreamKey, "*", "event", "active", "jobId", jobId, "prev", "waiting")
@@ -115,0 +119,0 @@ rcall("HSET", jobKey, "processedOn", processedOn)

dist/esm/scripts/moveToFinished-14.js

@@ -299,2 +299,6 @@ const content = `--[[
   end
+  if opts['name'] then
+    -- Set "processedBy" field to the worker name
+    rcall("HSET", jobKey, "pb", opts['name'])
+  end
   rcall("XADD", eventStreamKey, "*", "event", "active", "jobId", jobId, "prev", "waiting")
@@ -301,0 +305,0 @@ rcall("HSET", jobKey, "processedOn", processedOn)

dist/esm/utils.d.ts

@@ -45,4 +45,3 @@ /// <reference types="node" />
 export declare const errorToJSON: (value: any) => Record<string, any>;
-export declare const WORKER_SUFFIX = "";
 export declare const QUEUE_EVENT_SUFFIX = ":qe";
 export {};

dist/esm/utils.js

@@ -148,4 +148,3 @@ import { Cluster } from 'ioredis';
 };
-export const WORKER_SUFFIX = '';
 export const QUEUE_EVENT_SUFFIX = ':qe';
 //# sourceMappingURL=utils.js.map

package.json

 {
   "name": "bullmq",
-  "version": "5.2.1",
+  "version": "5.3.0",
   "description": "Queue for messages and jobs based on Redis",
@@ -5,0 +5,0 @@ "homepage": "https://bullmq.io/",

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

1508159

increased by0.32%

Lines of code

23956

increased by0.28%

No dependency changes