-
-
Notifications
You must be signed in to change notification settings - Fork 151
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #122 from taskrabbit/retry
Retry
- Loading branch information
Showing
11 changed files
with
609 additions
and
179 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,102 @@ | ||
| ///////////////////////// | ||
| // REQUIRE THE PACKAGE // | ||
| ///////////////////////// | ||
|
|
||
| var NR = require(__dirname + "/../index.js"); | ||
| // In your projects: var NR = require("node-resque"); | ||
|
|
||
| /////////////////////////// | ||
| // SET UP THE CONNECTION // | ||
| /////////////////////////// | ||
|
|
||
| var connectionDetails = { | ||
| pkg: 'ioredis', | ||
| host: '127.0.0.1', | ||
| password: null, | ||
| port: 6379, | ||
| database: 0, | ||
| // namespace: 'resque', | ||
| // looping: true, | ||
| // options: {password: 'abc'}, | ||
| }; | ||
|
|
||
| ////////////////////////////// | ||
| // DEFINE YOUR WORKER TASKS // | ||
| ////////////////////////////// | ||
|
|
||
| var jobs = { | ||
| "add": { | ||
| plugins: [ 'retry' ], | ||
| pluginOptions: { | ||
| retry: { | ||
| retryLimit: 3, | ||
| // retryDelay: 1000, | ||
| backoffStrategy: [1000 * 10, 1000 * 20, 1000 * 30], | ||
| }, | ||
| }, | ||
| perform: function(a,b,callback){ | ||
| var broken = true; | ||
|
|
||
| if(broken){ | ||
| return callback(new Error('BUSTED')); | ||
| }else{ | ||
| return callback(null, (a + b)); | ||
| } | ||
| }, | ||
| } | ||
| }; | ||
|
|
||
| //////////////////// | ||
| // START A WORKER // | ||
| //////////////////// | ||
|
|
||
| var worker = new NR.worker({connection: connectionDetails, queues: ['math']}, jobs); | ||
| worker.connect(function(){ | ||
| worker.workerCleanup(); // optional: cleanup any previous improperly shutdown workers on this host | ||
| worker.start(); | ||
| }); | ||
|
|
||
| /////////////////////// | ||
| // START A SCHEDULER // | ||
| /////////////////////// | ||
|
|
||
| var scheduler = new NR.scheduler({connection: connectionDetails}); | ||
| scheduler.connect(function(){ | ||
| scheduler.start(); | ||
| }); | ||
|
|
||
| ///////////////////////// | ||
| // REGESTER FOR EVENTS // | ||
| ///////////////////////// | ||
|
|
||
| worker.on('start', function(){ console.log("worker started"); }); | ||
| worker.on('end', function(){ console.log("worker ended"); }); | ||
| worker.on('cleaning_worker', function(worker, pid){ console.log("cleaning old worker " + worker); }); | ||
| worker.on('poll', function(queue){ console.log("worker polling " + queue); }); | ||
| worker.on('job', function(queue, job){ console.log("working job " + queue + " " + JSON.stringify(job)); }); | ||
| worker.on('reEnqueue', function(queue, job, plugin){ console.log("reEnqueue job (" + plugin + ") " + queue + " " + JSON.stringify(job)); }); | ||
| worker.on('success', function(queue, job, result){ console.log("job success " + queue + " " + JSON.stringify(job) + " >> " + result); }); | ||
| worker.on('pause', function(){ console.log("worker paused"); }); | ||
| worker.on('error', function(queue, job, error){ console.log("error " + queue + " " + JSON.stringify(job) + " >> " + error); }); | ||
| worker.on('failure', function(queue, job, failure){ | ||
| console.log("job failure " + queue + " " + JSON.stringify(job) + " >> " + failure); | ||
| setTimeout(process.exit, 2000); | ||
| }); | ||
|
|
||
| scheduler.on('start', function(){ console.log("scheduler started"); }); | ||
| scheduler.on('end', function(){ console.log("scheduler ended"); }); | ||
| scheduler.on('poll', function(){ console.log("scheduler polling"); }); | ||
| scheduler.on('master', function(state){ console.log("scheduler became master"); }); | ||
| scheduler.on('error', function(error){ console.log("scheduler error >> " + error); }); | ||
| scheduler.on('working_timestamp', function(timestamp){ console.log("scheduler working timestamp " + timestamp); }); | ||
| scheduler.on('transferred_job', function(timestamp, job){ console.log("scheduler enquing job " + timestamp + " >> " + JSON.stringify(job)); }); | ||
|
|
||
| //////////////////////////////////// | ||
| // CONNECT TO A QUEUE AND WORK IT // | ||
| //////////////////////////////////// | ||
|
|
||
| var queue = new NR.queue({connection: connectionDetails}, jobs); | ||
| queue.on('error', function(error){ console.log(error); }); | ||
| queue.connect(function(){ | ||
| queue.enqueue('math', "add", [1,2]); | ||
| }); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,169 @@ | ||
| // If a job fails, retry it N times before finally placing it into the failed queue | ||
| // a port of some of the features in https://github.com/lantins/resque-retry | ||
|
|
||
| var crypto = require('crypto'); | ||
| var os = require('os'); | ||
|
|
||
| var retry = function(worker, func, queue, job, args, options){ | ||
| var self = this; | ||
|
|
||
| if(!options.retryLimit){ options.retryLimit = 1; } | ||
| if(!options.retryDelay){ options.retryDelay = (1000 * 5); } | ||
| if(!options.backoffStrategy){ options.backoffStrategy = null; } | ||
|
|
||
| self.name = 'retry'; | ||
| self.worker = worker; | ||
| self.queue = queue; | ||
| self.func = func; | ||
| self.job = job; | ||
| self.args = args; | ||
| self.options = options; | ||
|
|
||
| if(self.worker.queueObject){ | ||
| self.queueObject = self.worker.queueObject; | ||
| }else{ | ||
| self.queueObject = self.worker; | ||
| } | ||
| }; | ||
|
|
||
| //////////////////// | ||
| // PLUGIN METHODS // | ||
| //////////////////// | ||
|
|
||
| retry.prototype.argsKey = function(){ | ||
| var self = this; | ||
| // return crypto.createHash('sha1').update(self.args.join('-')).digest('hex'); | ||
| if(!self.args || self.args.length === 0){ return ''; } | ||
| return self.args.join('-'); | ||
| }; | ||
|
|
||
| retry.prototype.retryKey = function(){ | ||
| var self = this; | ||
| return self.queueObject.connection.key('resque-retry', self.func, self.argsKey()).replace(/\s/, ''); | ||
| }; | ||
|
|
||
| retry.prototype.failureKey = function(){ | ||
| var self = this; | ||
| return self.queueObject.connection.key('failure-resque-retry:' + self.func + ':' + self.argsKey()).replace(/\s/, ''); | ||
| }; | ||
|
|
||
| retry.prototype.maxDelay = function(){ | ||
| var self = this; | ||
| var maxDelay = self.options.retryDelay || 1; | ||
| if(Array.isArray(self.options.backoffStrategy)){ | ||
| self.options.backoffStrategy.forEach(function(d){ | ||
| if(d > maxDelay){ maxDelay = d; } | ||
| }); | ||
| } | ||
| return maxDelay; | ||
| }; | ||
|
|
||
| retry.prototype.redis = function(){ | ||
| var self = this; | ||
| return self.queueObject.connection.redis; | ||
| }; | ||
|
|
||
| retry.prototype.attemptUp = function(callback){ | ||
| var self = this; | ||
| var key = self.retryKey(); | ||
| self.redis().setnx(key, -1, function(error){ | ||
| if(error){ return callback(error); } | ||
| self.redis().incr(key, function(error, retryCount){ | ||
| if(error){ return callback(error); } | ||
| self.redis().expire(key, self.maxDelay(), function(error){ | ||
| if(error){ return callback(error); } | ||
| var remaning = self.options.retryLimit - retryCount - 1; | ||
| return callback(null, remaning); | ||
| }); | ||
| }); | ||
| }); | ||
| }; | ||
|
|
||
| retry.prototype.saveLastError = function(callback){ | ||
| var self = this; | ||
| var now = new Date(); | ||
| var failedAt = '' + | ||
| now.getFullYear() + '/' + | ||
| (('0' + (now.getMonth() + 1)).slice(-2)) + '/' + | ||
| (('0' + now.getDate()).slice(-2)) + ' ' + | ||
| (('0' + now.getHours()).slice(-2)) + ':' + | ||
| (('0' + now.getMinutes()).slice(-2)) + ':' + | ||
| (('0' + now.getSeconds()).slice(-2)) | ||
| ; | ||
|
|
||
| var data = { | ||
| failed_at : failedAt, | ||
| payload : self.args, | ||
| exception : String(self.worker.error), | ||
| error : String(self.worker.error), | ||
| backtrace : self.worker.error.stack.split(os.EOL) || [], | ||
| worker : self.func, | ||
| queue : self.queue | ||
| }; | ||
|
|
||
| self.redis().setex(self.failureKey(), self.maxDelay(), JSON.stringify(data), callback); | ||
| }; | ||
|
|
||
| retry.prototype.cleanup = function(callback){ | ||
| var self = this; | ||
| var key = self.retryKey(); | ||
| var failureKey = self.failureKey(); | ||
| self.redis().del(key, function(error){ | ||
| if(error){ return callback(error); } | ||
| self.redis().del(failureKey, function(error){ | ||
| if(error){ return callback(error); } | ||
| return callback(); | ||
| }); | ||
| }); | ||
| }; | ||
|
|
||
| retry.prototype.after_perform = function(callback){ | ||
| var self = this; | ||
|
|
||
| if(!self.worker.error){ | ||
| self.cleanup(callback); | ||
| } | ||
|
|
||
| self.attemptUp(function(error, remaning){ | ||
| if(error){ return callback(error); } | ||
| self.saveLastError(function(error){ | ||
| if(error){ return callback(error); } | ||
| if(remaning <= 0){ | ||
| self.cleanup(function(error){ | ||
| if(error){ return callback(error); } | ||
| return callback(self.worker.error, true); | ||
| }); | ||
| }else{ | ||
| var nextTryDelay = self.options.retryDelay; | ||
| if(Array.isArray(self.options.backoffStrategy)){ | ||
| var index = (self.options.retryLimit - remaning - 1); | ||
| if(index > (self.options.backoffStrategy.length - 1)){ | ||
| index = (self.options.backoffStrategy.length - 1); | ||
| } | ||
| nextTryDelay = self.options.backoffStrategy[index]; | ||
| } | ||
|
|
||
| self.queueObject.enqueueIn(nextTryDelay, self.queue, self.func, self.args, function(error){ | ||
| if(error){ return callback(error); } | ||
|
|
||
| self.worker.emit('reEnqueue', self.queue, self.job, { | ||
| delay: nextTryDelay, | ||
| remaningAttempts: remaning, | ||
| err: self.worker.error | ||
| }); | ||
|
|
||
| self.redis().decr(self.queueObject.connection.key('stat', 'processed')); | ||
| self.redis().decr(self.queueObject.connection.key('stat', 'processed', self.worker.name)); | ||
|
|
||
| self.redis().incr(self.queueObject.connection.key('stat', 'failed')); | ||
| self.redis().incr(self.queueObject.connection.key('stat', 'failed', self.worker.name)); | ||
|
|
||
| delete self.worker.error; | ||
| return callback(null, true); | ||
| }); | ||
| } | ||
| }); | ||
| }); | ||
| }; | ||
|
|
||
| exports.retry = retry; |
Oops, something went wrong.