track poller running state

use setTimeout, and wait for the search and job processing to complete before polling for jobs again
This commit is contained in:
2017-05-23 20:10:38 -07:00
parent 01adab4174
commit 128905cdb4

View File

@@ -37,6 +37,7 @@ export default class Worker extends events.EventEmitter {
this._poller = { this._poller = {
timer: false, timer: false,
enabled: true, enabled: true,
running: false,
}; };
this.debug(`Created worker for job type ${this.jobtype}`); this.debug(`Created worker for job type ${this.jobtype}`);
@@ -250,23 +251,33 @@ export default class Worker extends events.EventEmitter {
} }
_startJobPolling() { _startJobPolling() {
if (!this._poller.enabled) { if (!this._poller.enabled || this._poller.running) {
return; return;
} }
this._poller.timer = setInterval(() => { this._poller.timer = setTimeout(() => {
this._getPendingJobs().then((jobs) => this._claimPendingJobs(jobs)); this._getPendingJobs().then((jobs) => {
if (!this._poller.running) return;
const foundJobs = (!jobs || jobs.length === 0);
const task = foundJobs ? Promise.resolve() : this._claimPendingJobs(jobs);
task.then(() => {
this._poller.running = false;
this._startJobPolling();
});
});
} , this.checkInterval); } , this.checkInterval);
this._poller.running = true;
} }
_stopJobPolling() { _stopJobPolling() {
clearInterval(this._poller.timer); this._poller.running = false;
clearTimeout(this._poller.timer);
} }
_claimPendingJobs(jobs) { _claimPendingJobs(jobs = []) {
if (!jobs || jobs.length === 0) return;
this._stopJobPolling();
let claimed = false; let claimed = false;
// claim a single job, stopping after first successful claim // claim a single job, stopping after first successful claim
@@ -292,10 +303,8 @@ export default class Worker extends events.EventEmitter {
this.debug(`Claimed job ${claimedJob._id}`); this.debug(`Claimed job ${claimedJob._id}`);
return this._performJob(claimedJob); return this._performJob(claimedJob);
}) })
.then(() => this._startJobPolling())
.catch((err) => { .catch((err) => {
this.debug('Error claiming jobs', err); this.debug('Error claiming jobs', err);
this._startJobPolling();
}); });
} }