w33ble/elastiq
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

refactoring failures

Browse Source
This commit is contained in:
Joe Fleming
2016-05-02 14:42:27 -07:00
parent 60d486b804
commit 7dadeeb111
1 changed files with 31 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
src/worker.js
View File

@@ -71,17 +71,17 @@ export default class Job extends events.EventEmitter {
}); });
} }
_failJob(job, msg = false) { _failJob(job, output = false) {
this.debug(`Failing job ${job._id}`); this.debug(`Failing job ${job._id}`);
const completedTime = moment().toISOString();
const doc = { const doc = {
status: jobStatuses.JOB_STATUS_FAILED status: jobStatuses.JOB_STATUS_FAILED,
completed_at: completedTime,
}; };
if (msg) { if (output) {
doc.output = { doc.output = this._formatOutput(output);
content_type: 'text/plain',
content: msg
};
} }
return this.client.update({ return this.client.update({
@@ -97,22 +97,30 @@ export default class Job extends events.EventEmitter {
}); });
} }
_formatOutput(output) {
const unknownMime = false;
const docOutput = {};
if (typeof output === 'object' && output.content) {
docOutput.content = output.content;
docOutput.content_type = output.content_type || unknownMime;
} else {
docOutput.content = output || '';
docOutput.content_type = unknownMime;
}
return docOutput;
}
_performJob(job) { _performJob(job) {
this.debug(`Starting job ${job._id}`); this.debug(`Starting job ${job._id}`);
return Bluebird.fromCallback((cb) => this.workerFn(job._source.payload, cb)) return Bluebird.fromCallback((cb) => this.workerFn(job._source.payload, cb))
.then((output) => { .then((output) => {
const completedTime = moment().toISOString(); this.debug(`Completed job ${job._id}`);
const unknownMime = false;
const docOutput = {};
if (typeof output === 'object' && output.content) { const completedTime = moment().toISOString();
docOutput.content = output.content; const docOutput = this._formatOutput(output);
docOutput.content_type = output.content_type || unknownMime;
} else {
docOutput.content = output || '';
docOutput.content_type = unknownMime;
}
const doc = { const doc = {
status: jobStatuses.JOB_STATUS_COMPLETED, status: jobStatuses.JOB_STATUS_COMPLETED,
@@ -132,25 +140,14 @@ export default class Job extends events.EventEmitter {
throw err; throw err;
}); });
}, (jobErr) => { }, (jobErr) => {
const completedTime = moment().toISOString(); this.debug(`Failure occurred during job ${job._id}`);
const doc = {
status: jobStatuses.JOB_STATUS_FAILED,
completed_at: completedTime,
output: {
content_type: false,
content: jobErr.toString()
}
};
return this.client.update({ return this._failJob(job, jobErr.toString())
index: job._index,
type: job._type,
id: job._id,
version: job._version,
body: { doc }
})
.catch(() => false) .catch(() => false)
.then(() => { throw jobErr; }); .then(() => {
this.emit('job_error', jobErr);
throw jobErr;
});
}); });
} }