Repository: taskforcesh/bullmq
Pull request #562 (2021-05-23 21:18:10)

Feat expose client connection in flow-producer
Just like the queue exposes the client with the same method


Patch:
diff --git a/docs/gitbook/CHANGELOG.md b/docs/gitbook/CHANGELOG.md
index 8571285af1..6037aff0bb 100644
--- a/docs/gitbook/CHANGELOG.md
+++ b/docs/gitbook/CHANGELOG.md
@@ -1,3 +1,10 @@
+# [1.27.0](https://github.com/taskforcesh/bullmq/compare/v1.26.5...v1.27.0) (2021-05-24)
+
+
+### Features
+
+* **repeat:** add immediately opt for repeat ([d095573](https://github.com/taskforcesh/bullmq/commit/d095573f8e7ce5911f777df48368382eceb99d6a))
+
 ## [1.26.5](https://github.com/taskforcesh/bullmq/compare/v1.26.4...v1.26.5) (2021-05-21)
 
 
diff --git a/docs/gitbook/api/bullmq.repeat._keytodata.md b/docs/gitbook/api/bullmq.repeat._keytodata.md
deleted file mode 100644
index b2e2656755..0000000000
--- a/docs/gitbook/api/bullmq.repeat._keytodata.md
+++ /dev/null
@@ -1,29 +0,0 @@
-<!-- Do not edit this file. It is automatically generated by API Documenter. -->
-
-[Home](./index.md) &gt; [bullmq](./bullmq.md) &gt; [Repeat](./bullmq.repeat.md) &gt; [\_keyToData](./bullmq.repeat._keytodata.md)
-
-## Repeat.\_keyToData() method
-
-<b>Signature:</b>
-
-```typescript
-_keyToData(key: string): {
-        key: string;
-        name: string;
-        id: string;
-        endDate: number;
-        tz: string;
-        cron: string;
-    };
-```
-
-## Parameters
-
-|  Parameter | Type | Description |
-|  --- | --- | --- |
-|  key | string |  |
-
-<b>Returns:</b>
-
-{ key: string; name: string; id: string; endDate: number; tz: string; cron: string; }
-
diff --git a/docs/gitbook/api/bullmq.repeat.md b/docs/gitbook/api/bullmq.repeat.md
index 658f1618e5..be7644e7e0 100644
--- a/docs/gitbook/api/bullmq.repeat.md
+++ b/docs/gitbook/api/bullmq.repeat.md
@@ -15,7 +15,6 @@ export declare class Repeat extends QueueBase
 
 |  Method | Modifiers | Description |
 |  --- | --- | --- |
-|  [\_keyToData(key)](./bullmq.repeat._keytodata.md) |  |  |
 |  [addNextRepeatableJob(name, data, opts, skipCheckExists)](./bullmq.repeat.addnextrepeatablejob.md) |  |  |
 |  [getRepeatableCount()](./bullmq.repeat.getrepeatablecount.md) |  |  |
 |  [getRepeatableJobs(start, end, asc)](./bullmq.repeat.getrepeatablejobs.md) |  |  |
diff --git a/docs/gitbook/api/bullmq.repeatoptions.immediately.md b/docs/gitbook/api/bullmq.repeatoptions.immediately.md
new file mode 100644
index 0000000000..90360a6fe5
--- /dev/null
+++ b/docs/gitbook/api/bullmq.repeatoptions.immediately.md
@@ -0,0 +1,13 @@
+<!-- Do not edit this file. It is automatically generated by API Documenter. -->
+
+[Home](./index.md) &gt; [bullmq](./bullmq.md) &gt; [RepeatOptions](./bullmq.repeatoptions.md) &gt; [immediately](./bullmq.repeatoptions.immediately.md)
+
+## RepeatOptions.immediately property
+
+Repeated job should start right now ( work only with every settings)
+
+<b>Signature:</b>
+
+```typescript
+immediately?: boolean;
+```
diff --git a/docs/gitbook/api/bullmq.repeatoptions.md b/docs/gitbook/api/bullmq.repeatoptions.md
index 1b8844e703..8d67ea8606 100644
--- a/docs/gitbook/api/bullmq.repeatoptions.md
+++ b/docs/gitbook/api/bullmq.repeatoptions.md
@@ -20,8 +20,10 @@ export interface RepeatOptions
 |  [cron?](./bullmq.repeatoptions.cron.md) | string | <i>(Optional)</i> A cron pattern |
 |  [endDate?](./bullmq.repeatoptions.enddate.md) | Date \| string \| number | <i>(Optional)</i> End date when the repeat job should stop repeating. |
 |  [every?](./bullmq.repeatoptions.every.md) | number | <i>(Optional)</i> Repeat after this amount of milliseconds (<code>cron</code> setting cannot be used together with this setting.) |
+|  [immediately?](./bullmq.repeatoptions.immediately.md) | boolean | <i>(Optional)</i> Repeated job should start right now ( work only with every settings) |
 |  [jobId?](./bullmq.repeatoptions.jobid.md) | string | <i>(Optional)</i> |
 |  [limit?](./bullmq.repeatoptions.limit.md) | number | <i>(Optional)</i> Number of times the job should repeat at max. |
+|  [offset?](./bullmq.repeatoptions.offset.md) | number | <i>(Optional)</i> |
 |  [prevMillis?](./bullmq.repeatoptions.prevmillis.md) | number | <i>(Optional)</i> |
 |  [startDate?](./bullmq.repeatoptions.startdate.md) | Date \| string \| number | <i>(Optional)</i> Start date when the repeat job should start repeating (only with <code>cron</code>). |
 |  [tz?](./bullmq.repeatoptions.tz.md) | string | <i>(Optional)</i> Timezone |
diff --git a/docs/gitbook/api/bullmq.repeatoptions.offset.md b/docs/gitbook/api/bullmq.repeatoptions.offset.md
new file mode 100644
index 0000000000..9f66edda28
--- /dev/null
+++ b/docs/gitbook/api/bullmq.repeatoptions.offset.md
@@ -0,0 +1,11 @@
+<!-- Do not edit this file. It is automatically generated by API Documenter. -->
+
+[Home](./index.md) &gt; [bullmq](./bullmq.md) &gt; [RepeatOptions](./bullmq.repeatoptions.md) &gt; [offset](./bullmq.repeatoptions.offset.md)
+
+## RepeatOptions.offset property
+
+<b>Signature:</b>
+
+```typescript
+offset?: number;
+```
diff --git a/package.json b/package.json
index 648a11d303..381e2777f8 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "bullmq",
-  "version": "1.26.5",
+  "version": "1.27.0",
   "description": "Queue for messages and jobs based on Redis",
   "main": "dist/index.js",
   "types": "dist/index.d.ts",
diff --git a/src/classes/flow-producer.ts b/src/classes/flow-producer.ts
index bddee7a13f..85026eaf8c 100644
--- a/src/classes/flow-producer.ts
+++ b/src/classes/flow-producer.ts
@@ -4,7 +4,7 @@ import { Redis, Pipeline } from 'ioredis';
 
 import { EventEmitter } from 'events';
 import { QueueBaseOptions } from '../interfaces';
-import { RedisConnection } from './redis-connection';
+import { RedisClient, RedisConnection } from './redis-connection';
 import { KeysMap, QueueKeys } from './queue-keys';
 import { FlowJob } from '../interfaces/flow-job';
 import { Job } from './job';
@@ -87,6 +87,10 @@ export class FlowProducer extends EventEmitter {
     return jobsTree;
   }
 
+  get client(): Promise<RedisClient> {
+    return this.connection.client;
+  }
+
   /**
    * Add a node (job) of a flow to the queue. This method will recursively
    * add all its children as well. Note that a given job can potentially be
diff --git a/src/classes/job.ts b/src/classes/job.ts
index cdb1b0a1ed..56e0f28e8c 100644
--- a/src/classes/job.ts
+++ b/src/classes/job.ts
@@ -98,8 +98,19 @@ export class Job<T = any, R = any, N extends string = string> {
 
   constructor(
     private queue: MinimalQueue,
+    /**
+     * The name of the Job
+     */
     public name: N,
+
+    /**
+     * The payload for this job.
+     */
     public data: T,
+
+    /**
+     * The options object for this job.
+     */
     public opts: JobsOptions = {},
     public id?: string,
   ) {
@@ -120,6 +131,15 @@ export class Job<T = any, R = any, N extends string = string> {
     this.toKey = queue.toKey.bind(queue);
   }
 
+  /**
+   * Creates a new job and adds it to the queue.
+   * 
+   * @param queue the queue where to add the job.
+   * @param name  the name of the job.
+   * @param data  the payload of the job.
+   * @param opts the options bag for this job.
+   * @returns 
+   */
   static async create<T = any, R = any, N extends string = string>(
     queue: MinimalQueue,
     name: N,
@@ -140,6 +160,13 @@ export class Job<T = any, R = any, N extends string = string> {
     return job;
   }
 
+  /**
+   * Creates a bulk of jobs and adds them atomically to the given queue.
+   * 
+   * @param queue the queue were to add the jobs.
+   * @param jobs an array of jobs to be added to the queue.
+   * @returns 
+   */
   static async createBulk<T = any, R = any, N extends string = string>(
     queue: MinimalQueue,
     jobs: {
@@ -175,6 +202,13 @@ export class Job<T = any, R = any, N extends string = string> {
     return jobInstances;
   }
 
+  /**
+   * Instantiates a Job from a JobJsonRaw object (coming from a deserialized JSON object)
+   * @param queue the queue where the job belongs to.
+   * @param json the plain object containing the job.
+   * @param jobId an optional job id (overrides the id coming from the JSON object)
+   * @returns 
+   */
   static fromJSON(queue: MinimalQueue, json: JobJsonRaw, jobId?: string) {
     const data = JSON.parse(json.data || '{}');
     const opts = JSON.parse(json.opts || '{}');
@@ -210,6 +244,13 @@ export class Job<T = any, R = any, N extends string = string> {
     return job;
   }
 
+  /**
+   * Fetches a Job from the queue given the passed job id.
+   * 
+   * @param queue the queue where the job belongs to.
+   * @param jobId the job id.
+   * @returns 
+   */
   static async fromId(
     queue: MinimalQueue,
     jobId: string,
@@ -229,6 +270,10 @@ export class Job<T = any, R = any, N extends string = string> {
     return withoutQueue;
   }
 
+  /**
+   * Prepares a job to be serialized for storage in Redis.
+   * @returns 
+   */
   asJSON(): JobJson {
     return {
       id: this.id,
@@ -246,6 +291,11 @@ export class Job<T = any, R = any, N extends string = string> {
     };
   }
 
+  /**
+   * Updates a job's data
+   * 
+   * @param data the data that will replace the current jobs data.
+   */
   async update(data: T) {
     const client = await this.queue.client;
 
@@ -270,6 +320,11 @@ export class Job<T = any, R = any, N extends string = string> {
     return client.rpush(logsKey, logRow);
   }
 
+  /**
+   * Completely remove the job from the queue.
+   * Note, this call will throw an exception if the job 
+   * is being processed when the call is performed.
+   */
   async remove() {
     await this.queue.waitUntilReady();
 
@@ -402,26 +457,50 @@ export class Job<T = any, R = any, N extends string = string> {
     }
   }
 
+  /**
+   * 
+   * @returns true if the job has completed.
+   */
   isCompleted() {
     return this.isInZSet('completed');
   }
 
+  /**
+   * 
+   * @returns true if the job has failed.
+   */
   isFailed() {
     return this.isInZSet('failed');
   }
 
+  /**
+   * 
+   * @returns true if the job is delayed.
+   */
   isDelayed() {
     return this.isInZSet('delayed');
   }
 
+  /**
+   * 
+   * @returns true if the job is waiting for children.
+   */
   isWaitingChildren() {
     return this.isInZSet('waiting-children');
   }
 
+  /**
+   * 
+   * @returns true of the job is active.
+   */
   isActive() {
     return this.isInList('active');
   }
 
+  /**
+   * 
+   * @returns true if the job is waiting.
+   */
   async isWaiting() {
     return (await this.isInList('wait')) || (await this.isInList('paused'));
   }
@@ -539,10 +618,19 @@ export class Job<T = any, R = any, N extends string = string> {
     });
   }
 
+  /**
+   * Moves the job to the delay set.
+   * 
+   * @param timestamp timestmap where the job should be moved back to "wait"
+   * @returns 
+   */
   moveToDelayed(timestamp: number) {
     return Scripts.moveToDelayed(this.queue, this.id, timestamp);
   }
 
+  /**
+   * Promotes a delayed job so that it starts to be processed as soon as possible.
+   */
   async promote() {
     const queue = this.queue;
     const jobId = this.id;
@@ -584,6 +672,9 @@ export class Job<T = any, R = any, N extends string = string> {
     }
   }
 
+  /**
+   * Marks a job to not be retried if it fails (even if attempts has been configured)
+   */
   discard() {
     this.discarded = true;
   }
@@ -599,6 +690,13 @@ export class Job<T = any, R = any, N extends string = string> {
     return Scripts.isJobInList(this.queue, this.queue.toKey(list), this.id);
   }
 
+  /**
+   * Adds the job to Redis.
+   * 
+   * @param client 
+   * @param parentOpts 
+   * @returns 
+   */
   addJob(client: RedisClient, parentOpts?: ParentOpts): string {
     const queue = this.queue;
 
diff --git a/src/classes/repeat.ts b/src/classes/repeat.ts
index ac9f5099d7..41a0a6db59 100644
--- a/src/classes/repeat.ts
+++ b/src/classes/repeat.ts
@@ -28,6 +28,8 @@ export class Repeat extends QueueBase {
 
     const nextMillis = getNextMillis(now, repeatOpts);
 
+    const hasImmediately = repeatOpts.every && repeatOpts.immediately;
+    const offset = hasImmediately ? now - nextMillis : undefined;
     if (nextMillis) {
       // We store the undecorated opts.jobId into the repeat options
       if (!prevMillis && opts.jobId) {
@@ -47,6 +49,7 @@ export class Repeat extends QueueBase {
           repeatJobKey,
         ));
       }
+      const { immediately, ...filteredRepeatOpts } = repeatOpts;
 
       // The job could have been deleted since this check
       if (repeatableExists) {
@@ -54,9 +57,10 @@ export class Repeat extends QueueBase {
           name,
           nextMillis,
           repeatJobKey,
-          { ...opts, repeat: repeatOpts },
+          { ...opts, repeat: { offset, ...filteredRepeatOpts } },
           data,
           currentCount,
+          hasImmediately,
         );
       }
     }
@@ -69,6 +73,7 @@ export class Repeat extends QueueBase {
     opts: JobsOptions,
     data: T,
     currentCount: number,
+    hasImmediately: boolean,
   ) {
     const client = await this.client;
 
@@ -82,12 +87,13 @@ export class Repeat extends QueueBase {
       opts.repeat.jobId,
     );
     const now = Date.now();
-    const delay = nextMillis - now;
+    const delay =
+      nextMillis + (opts.repeat.offset ? opts.repeat.offset : 0) - now;
 
     const mergedOpts = {
       ...opts,
       jobId,
-      delay: delay < 0 ? 0 : delay,
+      delay: delay < 0 || hasImmediately ? 0 : delay,
       timestamp: now,
       prevMillis: nextMillis,
     };
@@ -124,7 +130,7 @@ export class Repeat extends QueueBase {
   async removeRepeatableByKey(repeatJobKey: string) {
     const client = await this.client;
 
-    const data = this._keyToData(repeatJobKey);
+    const data = this.keyToData(repeatJobKey);
     const queueKey = this.keys[''];
 
     const repeatJobId = getRepeatJobId(
@@ -143,7 +149,7 @@ export class Repeat extends QueueBase {
     );
   }
 
-  _keyToData(key: string) {
+  private keyToData(key: string) {
     const data = key.split(':');
 
     return {
@@ -215,7 +221,10 @@ function getNextMillis(millis: number, opts: RepeatOptions) {
   }
 
   if (opts.every) {
-    return Math.floor(millis / opts.every) * opts.every + opts.every;
+    return (
+      Math.floor(millis / opts.every) * opts.every +
+      (opts.immediately ? 0 : opts.every)
+    );
   }
 
   const currentDate =
@@ -235,7 +244,5 @@ function getNextMillis(millis: number, opts: RepeatOptions) {
 }
 
 function md5(str: string) {
-  return createHash('md5')
-    .update(str)
-    .digest('hex');
+  return createHash('md5').update(str).digest('hex');
 }
diff --git a/src/interfaces/repeat-options.ts b/src/interfaces/repeat-options.ts
index 1935f62e21..66ce565be7 100644
--- a/src/interfaces/repeat-options.ts
+++ b/src/interfaces/repeat-options.ts
@@ -29,11 +29,16 @@ export interface RepeatOptions {
    * (`cron` setting cannot be used together with this setting.)
    */
   every?: number;
-
+  /**
+   * Repeated job should start right now
+   * ( work only with every settings)
+   */
+  immediately?: boolean;
   /**
    * The start value for the repeat iteration count.
    */
   count?: number;
   prevMillis?: number;
+  offset?: number;
   jobId?: string;
 }


Test patch:
diff --git a/src/test/fixtures/fixture_processor_crash.js b/src/test/fixtures/fixture_processor_crash.js
index 75ef5b6e01..cb5a7af48b 100644
--- a/src/test/fixtures/fixture_processor_crash.js
+++ b/src/test/fixtures/fixture_processor_crash.js
@@ -12,5 +12,7 @@ module.exports = function(job) {
     process.exit(job.data.exitCode);
   }, 100);
 
-  return new Promise(() => {});
+  return new Promise(() => {
+    // do nothing
+  });
 };
diff --git a/src/test/test_repeat.ts b/src/test/test_repeat.ts
index 96444a190f..4e87328462 100644
--- a/src/test/test_repeat.ts
+++ b/src/test/test_repeat.ts
@@ -274,6 +274,55 @@ describe('repeat', function() {
     await worker.close();
   });
 
+  it('should repeat every 2 seconds and start immediately', async function() {
+    const queueScheduler = new QueueScheduler(queueName);
+    await queueScheduler.waitUntilReady();
+
+    const date = new Date('2017-02-07 9:24:00');
+    this.clock.setSystemTime(date);
+    const nextTick = 2 * ONE_SECOND;
+
+    const worker = new Worker(queueName, async () => {});
+
+    await queue.add(
+      'repeat',
+      { foo: 'bar' },
+      {
+        repeat: {
+          every: 2000,
+          immediately: true,
+        },
+      },
+    );
+
+    this.clock.tick(100);
+
+    let prev: Job;
+    let counter = 0;
+
+    const completing = new Promise<void>(resolve => {
+      worker.on('completed', async job => {
+        this.clock.tick(nextTick);
+        if (prev && counter === 1) {
+          expect(prev.timestamp).to.be.lt(job.timestamp);
+          expect(job.timestamp - prev.timestamp).to.be.gte(100);
+        } else if (prev) {
+          expect(prev.timestamp).to.be.lt(job.timestamp);
+          expect(job.timestamp - prev.timestamp).to.be.gte(2000);
+        }
+        prev = job;
+        counter++;
+        if (counter === 5) {
+          resolve();
+        }
+      });
+    });
+
+    await completing;
+    await queueScheduler.close();
+    await worker.close();
+  });
+
   // Skipped until we find a way of simulating time to avoid waiting 5 days
   it.skip('should repeat once a day for 5 days', async function() {
     const queueScheduler = new QueueScheduler(queueName);