feat: do not rely on comma to encode jobid in progress fixes #2003

lib/queue.js

@@ -398,10 +398,17 @@ Queue.prototype._setupQueueEventListeners = function() {
     const key = channel.split('@')[0];
     switch (key) {
       case progressKey: {
-        const commaPos = message.indexOf(',');
-        const jobId = message.substring(0, commaPos);
-        const progress = message.substring(commaPos + 1);
-        this.emit('global:progress', jobId, JSON.parse(progress));
+        // New way to send progress message data
+        try {
+          const { progress, jobId } = JSON.parse(message);
+          this.emit('global:progress', jobId, progress);
+        } catch (err) {
+          // If we fail we should try to parse the data using the deprecated method
+          const commaPos = message.indexOf(',');
+          const jobId = message.substring(0, commaPos);
+          const progress = message.substring(commaPos + 1);
+          this.emit('global:progress', jobId, JSON.parse(progress));
+        }
         break;
       }
       case delayedKey: {

lib/scripts.js

@@ -97,7 +97,10 @@ const scripts = {
 
     const progressJson = JSON.stringify(progress);
     return queue.client
-      .updateProgress(keys, [progressJson, job.id + ',' + progressJson])
+      .updateProgress(keys, [
+        progressJson,
+        JSON.stringify({ jobId: job.id, progress })
+      ])
       .then(() => {
         queue.emit('progress', job, progress);
       });

test/test_queue.js

@@ -875,6 +875,52 @@ describe('Queue', () => {
       });
     });
 
+    it('process a job that updates progress with an object', done => {
+      queue.process((job, jobDone) => {
+        expect(job.data.foo).to.be.equal('bar');
+        job.progress({ myvalue: 42 });
+        jobDone();
+      });
+
+      queue
+        .add({ foo: 'bar' })
+        .then(job => {
+          expect(job.id).to.be.ok;
+          expect(job.data.foo).to.be.eql('bar');
+        })
+        .catch(done);
+
+      queue.on('progress', (job, progress) => {
+        expect(job).to.be.ok;
+        expect(progress).to.be.eql({ myvalue: 42 });
+        done();
+      });
+    });
+
+    it('process a job that updates progress with an object emits a global event', done => {
+      let jobId;
+      queue.process((job, jobDone) => {
+        expect(job.data.foo).to.be.equal('bar');
+        job.progress({ myvalue: 42 });
+        jobDone();
+      });
+
+      queue
+        .add({ foo: 'bar' })
+        .then(job => {
+          expect(job.id).to.be.ok;
+          expect(job.data.foo).to.be.eql('bar');
+          jobId = job.id;
+        })
+        .catch(done);
+
+      queue.on('global:progress', (_jobId, progress) => {
+        expect(jobId).to.be.eql(_jobId);
+        expect(progress).to.be.eql({ myvalue: 42 });
+        done();
+      });
+    });
+
     it('process a job that returns data in the process handler', done => {
       queue.process((job, jobDone) => {
         expect(job.data.foo).to.be.equal('bar');
@@ -1901,7 +1947,10 @@ describe('Queue', () => {
       queue.add({});
       queue.add({});
 
-      queue.on('completed', _.after(2, () => done()));
+      queue.on(
+        'completed',
+        _.after(2, () => done())
+      );
     });
 
     //This job use delay to check that at any time we have 4 process in parallel.