0.3.2

.travis.yml

@@ -15,4 +15,5 @@ after_success: npm run coverage
 
 branches:
   only:
-    - master
+    - master
+    - /^v[0-9].*$/

package.json

@@ -1,6 +1,6 @@
 {
   "name": "esqueue",
-  "version": "0.3.0",
+  "version": "0.3.2",
   "description": "Job queue, powered by Elasticsearch",
   "main": "lib/index.js",
   "scripts": {

src/worker.js

@@ -180,7 +180,7 @@ export default class Job extends events.EventEmitter {
   }
 
   _claimPendingJobs(jobs) {
-    if (jobs.length === 0) return;
+    if (!jobs || jobs.length === 0) return;
 
     this._stopJobPolling();
     let claimed = false;
@@ -267,6 +267,9 @@ export default class Job extends events.EventEmitter {
       return jobs;
     })
     .catch((err) => {
+      // ignore missing indices errors
+      if (err.status === 404) return [];
+
       this.debug('job querying failed', err);
       this.emit('error', err);
       this.queue.emit('worker_error', {

test/src/worker.js

@@ -105,15 +105,9 @@ describe('Worker class', function () {
     });
   });
 
-  describe('searching for jobs', function () {
+  describe('polling for jobs', function () {
     let searchSpy;
 
-    function getSearchParams(jobtype, params = {}) {
-      new Worker(mockQueue, jobtype, noop, params);
-      clock.tick(defaults.interval);
-      return searchSpy.firstCall.args[0];
-    }
-
     beforeEach(() => {
       anchorMoment = moment(anchor);
       clock = sinon.useFakeTimers(anchorMoment.valueOf());
@@ -124,37 +118,91 @@ describe('Worker class', function () {
       clock.restore();
     });
 
-    describe('polling interval', function () {
+    it('should start polling for jobs after interval', function () {
-      it('should start polling for jobs after interval', function () {
+      new Worker(mockQueue, 'test', noop);
-        new Worker(mockQueue, 'test', noop);
+      sinon.assert.notCalled(searchSpy);
-        sinon.assert.notCalled(searchSpy);
+      clock.tick(defaults.interval);
-        clock.tick(defaults.interval);
+      sinon.assert.calledOnce(searchSpy);
-        sinon.assert.calledOnce(searchSpy);
+    });
+
+    it('should use interval option to control polling', function () {
+      const interval = 567;
+      new Worker(mockQueue, 'test', noop, { interval });
+      sinon.assert.notCalled(searchSpy);
+      clock.tick(interval);
+      sinon.assert.calledOnce(searchSpy);
+    });
+  });
+
+  describe('query for pending jobs', function () {
+    let worker;
+    let searchStub;
+
+    function getSearchParams(jobtype = 'test', params = {}) {
+      worker = new Worker(mockQueue, jobtype, noop, params);
+      worker._getPendingJobs();
+      return searchStub.firstCall.args[0];
+    }
+
+    describe('error handling', function () {
+      beforeEach(() => {
       });
 
-      it('should use interval option to control polling', function () {
+      it('should pass search errors', function (done) {
-        const interval = 567;
+        searchStub = sinon.stub(mockQueue.client, 'search', () => Promise.reject());
-        new Worker(mockQueue, 'test', noop, { interval });
+        worker = new Worker(mockQueue, 'test', noop);
-        sinon.assert.notCalled(searchSpy);
+        worker._getPendingJobs()
-        clock.tick(interval);
+        .then(() => done(new Error('should not resolve')))
-        sinon.assert.calledOnce(searchSpy);
+        .catch(() => { done(); });
+      });
+
+      it('should swollow index missing errors', function (done) {
+        searchStub = sinon.stub(mockQueue.client, 'search', () => Promise.reject({
+          status: 404
+        }));
+        worker = new Worker(mockQueue, 'test', noop);
+        worker._getPendingJobs()
+        .then(() => { done(); })
+        .catch(() => done(new Error('should not reject')));
+      });
+
+      it('should return an empty array on missing index', function (done) {
+        searchStub = sinon.stub(mockQueue.client, 'search', () => Promise.reject({
+          status: 404
+        }));
+        worker = new Worker(mockQueue, 'test', noop);
+        worker._getPendingJobs()
+        .then((res) => {
+          try {
+            expect(res).to.be.an(Array);
+            expect(res).to.have.length(0);
+            done();
+          } catch (e) {
+            done(e);
+          }
+        })
+        .catch(() => done(new Error('should not reject')));
       });
     });
 
     describe('query parameters', function () {
+      beforeEach(() => {
+        searchStub = sinon.stub(mockQueue.client, 'search', () => Promise.resolve());
+      });
+
       it('should query with version', function () {
-        const params = getSearchParams('test');
+        const params = getSearchParams();
         expect(params).to.have.property('version', true);
       });
 
       it('should query by default doctype', function () {
-        const params = getSearchParams('test');
+        const params = getSearchParams();
         expect(params).to.have.property('type', constants.DEFAULT_SETTING_DOCTYPE);
       });
 
       it('should query by custom doctype', function () {
         const doctype = 'custom_test';
-        const params = getSearchParams('test', { doctype });
+        const params = getSearchParams('type', { doctype });
         expect(params).to.have.property('type', doctype);
       });
     });
@@ -163,6 +211,16 @@ describe('Worker class', function () {
       const conditionPath = 'query.constant_score.filter.bool';
       const jobtype = 'test_jobtype';
 
+      beforeEach(() => {
+        searchStub = sinon.stub(mockQueue.client, 'search', () => Promise.resolve());
+        anchorMoment = moment(anchor);
+        clock = sinon.useFakeTimers(anchorMoment.valueOf());
+      });
+
+      afterEach(() => {
+        clock.restore();
+      });
+
       it('should search by job type', function () {
         const { body } = getSearchParams(jobtype);
         const conditions = get(body, conditionPath);
@@ -201,9 +259,9 @@ describe('Worker class', function () {
         expect(body).to.have.property('size', size);
       });
     });
-
   });
 
+
   describe('claiming a job', function () {
     let params;
     let job;