Fix job events

compwright · compwright · commit dfc3931c0b70 · 2019-06-29T00:14:59.000-04:00
diff --git a/lib/JobService.js b/lib/JobService.js
@@ -13,10 +13,37 @@ class JobService extends AdapterService {
       sorter,
       paginate: {
         max: 25
-      }
+      },
+      events: ['succeeded', 'retrying', 'failed', 'progress']
     }, options));
   }
 
+  setup (app, path) {
+    if (this.options.events.indexOf('succeeded') > -1) {
+      this.queue.on('job succeeded',
+        (id, result) => this.emit('succeeded', { id, result })
+      );
+    }
+
+    if (this.options.events.indexOf('retrying') > -1) {
+      this.queue.on('job retrying',
+        (id, error) => this.emit('retrying', { id, error })
+      );
+    }
+
+    if (this.options.events.indexOf('failed') > -1) {
+      this.queue.on('job failed',
+        (id, error) => this.emit('failed', { id, error })
+      );
+    }
+
+    if (this.options.events.indexOf('progress') > -1) {
+      this.queue.on('job progress',
+        (id, progress) => this.emit('progress', { id, progress })
+      );
+    }
+  }
+
   get queue () {
     return this.options.queue;
   }
diff --git a/lib/index.js b/lib/index.js
@@ -4,23 +4,5 @@ module.exports = (options = {}) => {
   if (!options.queue) {
     throw new Error('Missing queue parameter');
   }
-
-  const { queue } = options;
-  const service = new JobService(options);
-
-  // Subscribe the service to the Queue PubSub events
-  queue.on('job succeeded',
-    (id, result) => service.emit('succeeded', { id, result })
-  );
-  queue.on('job retrying',
-    (id, error) => service.emit('retrying', { id, error })
-  );
-  queue.on('job failed',
-    (id, error) => service.emit('failed', { id, error })
-  );
-  queue.on('job progress',
-    (id, progress) => service.emit('progress', { id, progress })
-  );
-
-  return service;
+  return new JobService(options);
 };
diff --git a/test/index.test.js b/test/index.test.js
@@ -88,6 +88,68 @@ describe('Feathers Bee-Queue Service', () => {
     );
   });
 
+  describe('pubsub events', () => {
+    describe('when enabled', () => {
+      let app, queue, service;
+
+      beforeEach(() => {
+        app = feathers();
+        queue = new Queue('pubsub');
+        service = app.use('/pubsub', plugin({
+          queue,
+          events: [ 'progress' ]
+        })).service('/pubsub');
+        service.setup(app, '/pubsub');
+      });
+
+      afterEach(() => queue.destroy());
+
+      it('emits progress event', done => {
+        service.create({})
+          .then(job => queue.getJob(job.id))
+          .then(job => {
+            service.on('progress', event => {
+              assert.strictEqual(event.id, job.id);
+              assert.strictEqual(event.progress, 50);
+              done();
+            });
+            job.reportProgress(50);
+          })
+          .catch(done);
+      });
+    });
+
+    describe('when disabled', () => {
+      let app, queue, service;
+
+      beforeEach(() => {
+        app = feathers();
+        queue = new Queue('pubsub2');
+        service = app.use('/pubsub2', plugin({
+          queue,
+          events: []
+        })).service('/pubsub2');
+        service.setup(app, '/pubsub2');
+      });
+
+      afterEach(() => queue.destroy());
+
+      it('does not emit progress event', done => {
+        service.create({})
+          .then(job => queue.getJob(job.id))
+          .then(job => {
+            service.on('progress', () => {
+              done(new Error('It should not emit an event'));
+            });
+            job.reportProgress(50);
+            assert.ok(true);
+            done();
+          })
+          .catch(done);
+      });
+    });
+  });
+
   describe('find', () => {
     it('validates the query status', async () => {
       try {
