Repository: taskforcesh/bullmq
Pull request #82 (2019-12-06 23:59:15)

$ as default queue event id #76
closes: #76

manast: thanks for the PR. Why can't you use ```waitUntilReady``` from the base class? seems like the effect will be the same.
manast: regarding consumer groups, I do not see how that would be useful in this case. Exactly once semantics is already provided by the current implementation, you just need to save the last event id and use it the next time you start listening to events.
jbr: @manast I was hesitant to use QueueBase#waitUntilReady because of this comment: https://github.com/taskforcesh/bullmq/blob/master/src/classes/queue-base.ts#L59. Will it make whatever refactor is going to be necessary on deprecation easier to use the same interface? What will waitUntilReady be replaced with on deprecation?

Patch:
diff --git a/src/classes/compat.ts b/src/classes/compat.ts
index fed33bb84d..42f95d8759 100644
--- 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/classes/queue-events.ts b/src/classes/queue-events.ts
index 653e81b538..9e0bab5b34 100644
--- a/src/classes/queue-events.ts
+++ b/src/classes/queue-events.ts
@@ -25,7 +25,7 @@ export class QueueEvents extends QueueBase {
     const opts: QueueEventsOptions = this.opts;
 
     const key = this.keys.events;
-    let id = opts.lastEventId || '0-0';
+    let id = opts.lastEventId || '$';
 
     while (!this.closing) {
       try {


Test patch:
diff --git a/src/test/test_compat.ts b/src/test/test_compat.ts
index 051179a7fc..0adbcb4192 100644
--- a/src/test/test_compat.ts
+++ b/src/test/test_compat.ts
@@ -369,23 +369,37 @@ 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();
-      });
+    it('should listen to global events with .once', 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.add('test', {});
+      await queue.process(() => null);
+      await delay(50);
+      expect(events).to.eql(['waiting', 'active', 'completed']);
+    });
 
-      queue.add('test', {});
-      queue.process(async () => {});
+    it('should listen to global events with .on', async function() {
+      const events: string[] = [];
+      queue.on('global:waiting', () => events.push('waiting'));
+      queue.on('global:active', () => events.push('active'));
+      queue.on('global:completed', () => events.push('completed'));
+      await queue.queueEvents.waitUntilReady();
+      await queue.add('test', {});
+      await queue.add('test', {});
+      await queue.process(() => null);
+      await delay(50);
+      expect(events).to.eql([
+        'waiting',
+        'waiting',
+        'active',
+        'completed',
+        'active',
+        'completed',
+      ]);
     });
   });
 
@@ -443,6 +457,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 +493,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 +635,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 +649,11 @@ describe('Compat', function() {
           }
         });
       });
+
+      await queue.queueEvents.waitUntilReady();
+
+      await queue.add('test', {});
+      return promise;
     });
   });
 });