fix: ensure QE ready before adding test events

taskforcesh · Dec 9, 2019 · fd190f4 · fd190f4
1 parent 3c5b01d
commit fd190f4
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 31 deletions.
diff --git a/src/classes/compat.ts b/src/classes/compat.ts
@@ -42,11 +42,10 @@ export class Queue3<T = any> extends EventEmitter {
    * The name of the queue
    */
   name: string;
+  queueEvents: QueueEvents;
 
   private opts: CommonOptions;
-
   private readonly queue: Queue;
-  private queueEvents: QueueEvents;
   private worker: Worker;
   private queueScheduler: QueueScheduler;
 

diff --git a/src/test/test_compat.ts b/src/test/test_compat.ts
@@ -369,23 +369,16 @@ describe('Compat', function() {
       });
     });
 
-    it('should listen to global events', function(done) {
-      let state: string;
-      queue.on('global:waiting', function() {
-        expect(state).to.be.undefined;
-        state = 'waiting';
-      });
-      queue.once('global:active', function() {
-        expect(state).to.be.equal('waiting');
-        state = 'active';
-      });
-      queue.once('global:completed', async function() {
-        expect(state).to.be.equal('active');
-        done();
-      });
-
-      queue.add('test', {});
-      queue.process(async () => {});
+    it('should listen to global events', async function() {
+      const events: string[] = [];
+      queue.once('global:waiting', () => events.push('waiting'));
+      queue.once('global:active', () => events.push('active'));
+      queue.once('global:completed', () => events.push('completed'));
+      await queue.queueEvents.waitUntilReady();
+      await queue.add('test', {});
+      await queue.process(() => null);
+      await delay(50);
+      expect(events).to.eql(['waiting', 'active', 'completed']);
     });
   });
 
@@ -443,6 +436,17 @@ describe('Compat', function() {
         isResumed = true,
         first = true;
 
+      queue.on('global:paused', async () => {
+        isPaused = false;
+        await queue.resume();
+      });
+
+      queue.on('global:resumed', () => {
+        isResumed = true;
+      });
+
+      await queue.queueEvents.waitUntilReady();
+
       const processPromise = new Promise((resolve, reject) => {
         process = async (job: Job) => {
           try {
@@ -468,15 +472,6 @@ describe('Compat', function() {
       queue.add('test', { foo: 'paused' });
       queue.add('test', { foo: 'paused' });
 
-      queue.on('global:paused', async () => {
-        isPaused = false;
-        await queue.resume();
-      });
-
-      queue.on('global:resumed', () => {
-        isResumed = true;
-      });
-
       return processPromise;
     });
 
@@ -619,9 +614,7 @@ describe('Compat', function() {
     it('pauses fast when queue is drained', async function() {
       await queue.process(async () => {});
 
-      await queue.add('test', {});
-
-      return new Promise((resolve, reject) => {
+      const promise = new Promise((resolve, reject) => {
         queue.on('global:drained', async () => {
           try {
             const start = new Date().getTime();
@@ -635,6 +628,11 @@ describe('Compat', function() {
           }
         });
       });
+
+      await queue.queueEvents.waitUntilReady();
+
+      await queue.add('test', {});
+      return promise;
     });
   });
 });