feat(queue-events): launch without launching process (#750)

.eslintrc.json

@@ -48,6 +48,7 @@
       }
     ],
     "arrow-parens": [2, "as-needed", { "requireForBlockBody": false }],
+    "curly": "error",
     "no-async-promise-executor": 0,
     "no-extraneous-class": 0,
     "@typescript-eslint/no-inferrable-types": 2,

src/classes/queue-events.ts

@@ -182,7 +182,12 @@ export declare interface QueueEvents {
  *
  */
 export class QueueEvents extends QueueBase {
-  constructor(name: string, { connection, ...opts }: QueueEventsOptions = {}) {
+  private running = false;
+
+  constructor(
+    name: string,
+    { connection, autorun = true, ...opts }: QueueEventsOptions = {},
+  ) {
     super(name, {
       ...opts,
       connection: isRedisInstance(connection)
@@ -197,12 +202,28 @@ export class QueueEvents extends QueueBase {
       this.opts,
     );
 
-    this.consumeEvents().catch(err => this.emit('error', err));
+    if (autorun) {
+      this.run().catch(error => this.emit('error', error));
+    }
   }
 
-  private async consumeEvents() {
-    const client = await this.client;
+  async run(): Promise<void> {
+    if (!this.running) {
+      try {
+        this.running = true;
+        const client = await this.client;
+
+        await this.consumeEvents(client);
+      } catch (error) {
+        this.running = false;
+        throw error;
+      }
+    } else {
+      throw new Error('Queue Events is already running.');
+    }
+  }
 
+  private async consumeEvents(client: RedisClient) {
     const opts: QueueEventsOptions = this.opts;
 
     const key = this.keys.events;
@@ -253,6 +274,7 @@ export class QueueEvents extends QueueBase {
         if (isNotConnectionError(err)) {
           throw err;
         }
+
         await delay(DELAY_TIME_5);
       }
     }

src/classes/redis-connection.ts

@@ -118,7 +118,7 @@ export class RedisConnection extends EventEmitter {
     if (client.status !== 'end') {
       let _resolve, _reject;
 
-      const disconnecting = new Promise((resolve, reject) => {
+      const disconnecting = new Promise<void>((resolve, reject) => {
         client.once('end', resolve);
         client.once('error', reject);
         _resolve = resolve;

src/interfaces/queue-options.ts

@@ -63,6 +63,7 @@ export interface QueueOptions extends QueueBaseOptions {
  * Options for QueueEvents
  */
 export interface QueueEventsOptions extends QueueBaseOptions {
+  autorun?: boolean;
   /**
    * Last event Id. If provided it is possible to continue
    * consuming events from a known Id instead of from the last

src/test/test_events.ts

@@ -6,7 +6,7 @@ import { FlowProducer, Queue, QueueEvents, Worker } from '../classes';
 import { delay, removeAllQueueData } from '../utils';
 
 describe('events', function() {
-  this.timeout(4000);
+  this.timeout(6000);
   let queue: Queue;
   let queueEvents: QueueEvents;
   let queueName: string;
@@ -24,6 +24,52 @@ describe('events', function() {
     await removeAllQueueData(new IORedis(), queueName);
   });
 
+  describe('when autorun option is provided as false', function() {
+    it('emits waiting when a job has been added', async () => {
+      const queueName2 = `test-${v4()}`;
+      const queue2 = new Queue(queueName2);
+      const queueEvents2 = new QueueEvents(queueName2, { autorun: false });
+      await queueEvents2.waitUntilReady();
+
+      const waiting = new Promise(resolve => {
+        queue2.on('waiting', resolve);
+      });
+
+      const running = queueEvents2.run();
+
+      await queue2.add('test', { foo: 'bar' });
+
+      await waiting;
+
+      await queue2.close();
+      await queueEvents2.close();
+      await expect(running).to.have.been.fulfilled;
+      await removeAllQueueData(new IORedis(), queueName2);
+    });
+
+    describe('when run method is called when queueEvent is running', function() {
+      it('throws error', async () => {
+        const queueName2 = `test-${v4()}`;
+        const queue2 = new Queue(queueName2);
+        const queueEvents2 = new QueueEvents(queueName2, { autorun: false });
+        await queueEvents2.waitUntilReady();
+
+        const running = queueEvents2.run();
+
+        await queue2.add('test', { foo: 'bar' });
+
+        await expect(queueEvents2.run()).to.be.rejectedWith(
+          'Queue Events is already running.',
+        );
+
+        await queue2.close();
+        await queueEvents2.close();
+        await expect(running).to.have.been.fulfilled;
+        await removeAllQueueData(new IORedis(), queueName2);
+      });
+    });
+  });
+
   it('should emit waiting when a job has been added', async function() {
     const waiting = new Promise(resolve => {
       queue.on('waiting', resolve);

src/test/test_worker.ts

@@ -518,7 +518,7 @@ describe('workers', function() {
   });
 
   describe('when autorun option is provided as false', function() {
-    it('process several jobs serially using process option as false', async () => {
+    it('processes several jobs serially using process option as false', async () => {
       let counter = 1;
       const maxJobs = 10;