New Case Study:See how Anthropic automated 95% of dependency reviews with Socket.Learn More →
Socket
Sign inDemoInstall
Socket
Sign inDemoInstall

ponos

Package Overview
Dependencies
Maintainers
3
Versions
46
Alerts
File Explorer

Advanced tools

License
Socket logo

Install Socket

Detect and block malicious and high-risk dependencies

Install

ponos - npm Package Compare versions

Comparing version 2.0.0 to 3.0.0-rc1

examples/basic-worker.js
examples/topic-exchanges-on-the-fly.js
lib/index.js
lib/rabbitmq.js

7

CHANGELOG.md
## Change Log
### v2.0.0 (2016/03/03 22:16 +00:00)
- [#30](https://github.com/Runnable/ponos/pull/30) package.json engine (@bkendall)
- [#29](https://github.com/Runnable/ponos/pull/29) auto deploy to NPM (@bkendall)
- [#28](https://github.com/Runnable/ponos/pull/28) ES6 (@bkendall)
- [#26](https://github.com/Runnable/ponos/pull/26) remove npm publish (@bkendall)
- [#27](https://github.com/Runnable/ponos/pull/27) fix allowed failures entry (@bkendall)
### v1.3.0 (2016/01/21 21:32 +00:00)

@@ -4,0 +11,0 @@ - [#25](https://github.com/Runnable/ponos/pull/25) Add monitor-dog (@podviaznikov)

26

lib/logger.js

@@ -1,9 +0,11 @@

'use strict'
'use strict';
const bunyan = require('bunyan')
const bunyan = require('bunyan');
/**
* Bunyan logger for ponos.
*
* @private
* @author Bryan Kendall
* @module ponos:logger
* @module ponos/lib/logger
*/

@@ -14,6 +16,6 @@ module.exports = bunyan.createLogger({

serializers: bunyan.stdSerializers
})
});
// Expose get streams for unit testing
module.exports._getStreams = _getStreams
module.exports._getStreams = _getStreams;

@@ -25,9 +27,7 @@ /**

*/
function _getStreams () {
return [
{
level: process.env.LOG_LEVEL,
stream: process.stdout
}
]
}
function _getStreams() {
return [{
level: process.env.LOG_LEVEL,
stream: process.stdout
}];
}

390

lib/server.js

@@ -1,18 +0,21 @@

'use strict'
const clone = require('101/clone')
const defaults = require('101/defaults')
const ErrorCat = require('error-cat')
const hermes = require('runnable-hermes')
const isFunction = require('101/is-function')
const isObject = require('101/is-object')
const isString = require('101/is-string')
const pick = require('101/pick')
const Promise = require('bluebird')
/* global ErrorCat */
'use strict';
const logger = require('./logger')
const Worker = require('./worker')
const assign = require('101/assign');
const clone = require('101/clone');
const defaults = require('101/defaults');
const errorCat = require('error-cat');
const Immutable = require('immutable');
const isFunction = require('101/is-function');
const isObject = require('101/is-object');
const pick = require('101/pick');
const Promise = require('bluebird');
const logger = require('./logger');
const RabbitMQ = require('./rabbitmq');
const Worker = require('./worker');
/**
* Ponos worker server class. Given a queue adapter the worker server will
* Ponos server class. Given a queue adapter the worker server will
* connect to RabbitMQ, subscribe to the given queues, and begin spawning

@@ -25,186 +28,151 @@ * workers for incoming jobs.

*
* @class
* @module ponos:server
* @author Bryan Kendall
* @author Ryan Sandor Richards
* @param {object} opts Options for the server.
* @param {Array} opts.queues An array of queue names to which the server should
* subscribe.
* @param {runnable-hermes~Hermes} [opts.hermes] A hermes client.
* @param {string} [opts.hostname] Hostname for RabbitMQ.
* @param {string|number} [opts.port] Port for RabbitMQ.
* @param {string} [opts.username] Username for RabbitMQ.
* @param {string} [opts.password] Username for Password.
* @param {bunyan} [opts.log] A bunyan logger to use for the server.
* @param {Object} opts Options for the server.
* @param {ErrorCat} [opts.errorCat] An error cat instance to use for the
* server.
* @param {Object<String, Function>} [opts.events] Mapping of event (fanout)
* exchanges which to subscribe and handlers.
* @param {bunyan} [opts.log] A bunyan logger to use for the server.
* @param {String} [opts.name=ponos] A name to namespace the created exchange queues.
* @param {Object} [opts.rabbitmq] RabbitMQ connection options.
* @param {String} [opts.rabbitmq.hostname=localhost] Hostname for RabbitMQ. Can
* be set with environment variable RABBITMQ_HOSTNAME.
* @param {Number} [opts.rabbitmq.port=5672] Port for RabbitMQ. Can be set with
* environment variable RABBITMQ_PORT.
* @param {String} [opts.rabbitmq.username] Username for RabbitMQ. Can be set
* with environment variable RABBITMQ_USERNAME.
* @param {String} [opts.rabbitmq.password] Username for Password. Can be set
* with environment variable RABBITMQ_PASSWORD.
* @param {Object<String, Function>} [opts.tasks] Mapping of queues to subscribe
* directly with handlers.
*/
class Server {
constructor (opts) {
this._tasks = {}
this._workerOptions = {}
this.opts = Object.assign({}, opts)
this.log = this.opts.log || logger.child({ module: 'ponos:server' })
this.errorCat = this.opts.errorCat || new ErrorCat()
constructor(opts) {
this._opts = assign({}, opts);
this.log = this._opts.log || logger.child({ module: 'ponos:server' });
this._workerOptions = {};
if (this.opts.hermes) {
this.hermes = this.opts.hermes
} else {
if (!Array.isArray(this.opts.queues)) {
throw new Error('ponos.server: missing required `queues` option.')
}
if (!this.opts.queues.every(isString)) {
throw new Error(
'ponos.server: each element of `queues` must be a string.'
)
}
// Defaults for the hermes client's rabbitmq connection
defaults(this.opts, {
hostname: process.env.RABBITMQ_HOSTNAME || 'localhost',
port: process.env.RABBITMQ_PORT || 5672,
username: process.env.RABBITMQ_USERNAME || 'guest',
password: process.env.RABBITMQ_PASSWORD || 'guest',
name: 'ponos'
})
this.hermes = hermes.hermesSingletonFactory(this.opts)
this._tasks = new Immutable.Map();
if (this._opts.tasks) {
this.setAllTasks(this._opts.tasks);
}
this.hermes = Promise.promisifyAll(this.hermes)
}
/**
* Helper function that subscribes to a given queue name.
* @private
* @param {string} queueName Name of the queue.
*/
_subscribe (queueName) {
this.log.trace('subscribing to ' + queueName)
if (!this._tasks[queueName]) {
this.log.warn({ queueName: queueName }, 'handler not defined')
return
this._events = new Immutable.Map();
if (this._opts.events) {
this.setAllEvents(this._opts.events);
}
this.hermes.subscribe(queueName, (job, done) => {
this._runWorker(queueName, job, done)
})
}
/**
* Helper function to subscribe to all queues.
* @private
* @return {promise} Resolved when queues are all subscribed.
*/
_subscribeAll () {
this.log.trace('_subscribeAll')
return Promise.resolve(this.hermes.getQueues()).bind(this)
.map(this._subscribe)
}
this.errorCat = this._opts.errorCat || errorCat;
/**
* Helper function that unsubscribes to a given queue name.
* @private
* @param {string} queueName Name of the queue.
* @return {promise} A promise that resolves post worker creation
* and subscription.
*/
_unsubscribe (queueName) {
this.log.trace('unsubscribing from ' + queueName)
const handler = null // null will remove all handlers
return this.hermes.unsubscribeAsync(queueName, handler)
// add the name to RabbitMQ options
const rabbitmqOpts = defaults(this._opts.rabbitmq || {}, { name: this._opts.name });
this._rabbitmq = new RabbitMQ(rabbitmqOpts);
}
/**
* Helper function to unsubscribe from all queues.
* @private
* @return {promise} Resolved when queues are all subscribed.
* Start consuming from the subscribed queues. This is called by `.start`.
* This can be called after the server has been started to start consuming
* from additional queues.
*
* @return {Promise} Promise resolved when consuming has started.
*/
_unsubscribeAll () {
this.log.trace('_unsubscribeAll')
return Promise.resolve(this.hermes.getQueues()).bind(this)
.map(this._unsubscribe)
consume() {
return this._rabbitmq.consume();
}
/**
* Runs a worker for the given queue name, job, and acknowledgement callback.
* @private
* @param {string} queueName Name of the queue.
* @param {object} job Job for the worker to perform.
* @param {function} done RabbitMQ acknowledgement callback.
* Starts the worker server, connects to RabbitMQ, subscribes and consumes
* from all the provided queues and exchanges (tasks and events).
*
* @return {Promise} Promise that resolves once the server is listening.
*/
_runWorker (queueName, job, done) {
this.log.trace({ queue: queueName, job: job }, '_runWorker')
let opts = clone(this._workerOptions[queueName])
defaults(opts, {
queue: queueName,
job: job,
task: this._tasks[queueName],
done: done,
log: this.log,
errorCat: this.errorCat
})
Worker.create(opts)
start() {
this.log.trace('starting');
return this._rabbitmq.connect().then(() => {
return this._subscribeAll();
}).then(() => {
return this.consume();
}).then(() => {
this.log.trace('started');
}).catch(err => {
this.errorCat.report(err);
throw err;
});
}
/**
* Starts the worker server and listens for jobs coming from all queues.
* @return {promise} A promise that resolves when the server is listening.
* Stops the worker server, unsubscribing and disconnecting from RabbitMQ.
*
* @return {Promise} A promise that resolves when the server is stopped.
*/
start () {
this.log.trace('starting')
return this.hermes.connectAsync().bind(this)
.then(this._subscribeAll)
.then(() => {
this.log.trace('started')
})
.catch((err) => {
this.errorCat.report(err)
throw err
})
stop() {
this.log.trace('stopping');
return this._rabbitmq.unsubscribe().then(() => {
return this._rabbitmq.disconnect();
}).then(() => {
this.log.trace('stopped');
}).catch(err => {
this.errorCat.report(err);
throw err;
});
}
/**
* Stops the worker server.
* @return {promise} A promise that resolves when the server is stopped.
* Takes a map of queues and task handlers and sets them all.
*
* @param {Object<String, Function>} map A map of queue names and task
* handlers.
* @param {String} map.key Queue name.
* @param {Object} map.value Object with a handler and additional options for
* the worker (must have a `.task` handler function)
* @param {Function} map.value Handler function to take a job.
* @returns {Server} The server.
*/
stop () {
this.log.trace('stopping')
return this._unsubscribeAll().bind(this)
.then(() => {
return this.hermes.closeAsync()
})
.then(() => {
this.log.trace('stopped')
})
.catch((err) => {
this.errorCat.report(err)
throw err
})
setAllTasks(map) {
if (!isObject(map)) {
throw new Error('ponos.server: setAllTasks must be called with an object');
}
Object.keys(map).forEach(key => {
const value = map[key];
if (isObject(value)) {
if (!isFunction(value.task)) {
this.log.warn({ key: key }, 'no task function defined for key');
return;
}
this.setTask(key, value.task, value);
} else {
this.setTask(key, map[key]);
}
});
return this;
}
/**
* Takes a map of queues and task handlers and sets them all.
* @param {object} map A map of queue names to task handlers.
* @param {string} map.key Queue name.
* @param {function} map.value Function to take a job or an object with a task
* and additional options for the worker.
* @returns {ponos.Server} The server.
* Takes a map of event exchanges and handlers and subscribes to them all.
*
* @param {Object<String, Function>} map A map of exchanges and task handlers.
* @param {String} map.key Exchange name.
* @param {Object} map.value Object with handler and additional options for
* the worker (must have a `.task` handler function)
* @param {Function} map.value Handler function to take a job.
* @returns {Server} The server.
*/
setAllTasks (map) {
setAllEvents(map) {
if (!isObject(map)) {
throw new Error('ponos.server: setAllTasks must be called with an object')
throw new Error('ponos.server: setAllEvents must be called with an object');
}
Object.keys(map).forEach((key) => {
const value = map[key]
Object.keys(map).forEach(key => {
const value = map[key];
if (isObject(value)) {
if (!isFunction(value.task)) {
this.log.warn({ key: key }, 'no task function defined for key')
return
this.log.warn({ key: key }, 'no task function defined for key');
return;
}
this.setTask(key, value.task, value)
this.setEvent(key, value.task, value);
} else {
this.setTask(key, map[key])
this.setEvent(key, map[key]);
}
})
return this
});
return this;
}

@@ -214,20 +182,100 @@

* Assigns a task to a queue.
* @param {string} queueName Queue name.
* @param {function} task Function to take a job and return a promise.
* @param {object} opts Options for the worker that performs the task.
* @returns {ponos.Server} The server.
*
* @param {String} queueName Queue name.
* @param {Function} task Function to take a job and return a promise.
* @param {Object} [opts] Options for the worker that performs the task.
* @returns {Server} The server.
*/
setTask (queueName, task, opts) {
this.log.trace({ queue: queueName }, 'setting task for queue')
setTask(queueName, task, opts) {
this.log.trace({
queue: queueName,
method: 'setTask'
}, 'setting task for queue');
if (!isFunction(task)) {
throw new Error('ponos.server: setTask task handler must be a function')
throw new Error('ponos.server: setTask task handler must be a function');
}
this._tasks[queueName] = task
this._workerOptions[queueName] = isObject(opts)
? pick(opts, 'msTimeout')
: {}
return this
this._tasks = this._tasks.set(queueName, task);
this._workerOptions[queueName] = opts && isObject(opts) ? pick(opts, 'msTimeout') : {};
return this;
}
/**
* Assigns a task to an exchange.
*
* @param {String} exchangeName Exchange name.
* @param {Function} task Function to take a job and return a promise.
* @param {Object} [opts] Options for the worker that performs the task.
* @returns {Server} The server.
*/
setEvent(exchangeName, task, opts) {
this.log.trace({
exchange: exchangeName,
method: 'setEvent'
}, 'setting task for queue');
if (!isFunction(task)) {
throw new Error('ponos.server: setEvent task handler must be a function');
}
this._events = this._events.set(exchangeName, task);
this._workerOptions[exchangeName] = opts && isObject(opts) ? pick(opts, 'msTimeout') : {};
return this;
}
// Private Methods
/**
* Helper function to subscribe to all queues.
*
* @private
* @return {Promise} Promise that resolves when queues are all subscribed.
*/
_subscribeAll() {
this.log.trace('_subscribeAll');
const tasks = this._tasks;
const events = this._events;
return Promise.map(tasks.keySeq(), queue => {
return this._rabbitmq.subscribeToQueue(queue, (job, done) => {
this._runWorker(queue, tasks.get(queue), job, done);
});
}).then(() => {
return Promise.map(events.keySeq(), exchange => {
return this._rabbitmq.subscribeToFanoutExchange(exchange, (job, done) => {
this._runWorker(exchange, events.get(exchange), job, done);
});
});
});
}
/**
* Runs a worker for the given queue name, job, and acknowledgement callback.
*
* @private
* @param {String} queueName Name of the queue.
* @param {Function} handler Handler to perform the work.
* @param {Object} job Job for the worker to perform.
* @param {Function} done RabbitMQ acknowledgement callback.
*/
_runWorker(queueName, handler, job, done) {
this.log.trace({
queue: queueName,
job: job,
method: '_runWorker'
}, 'running worker');
const opts = clone(this._workerOptions[queueName]);
defaults(opts, {
queue: queueName,
job: job,
task: handler,
done: done,
log: this.log,
errorCat: this.errorCat
});
Worker.create(opts);
}
}
module.exports = Server
/**
* Server class.
* @module ponos/lib/server
* @see Server
*/
module.exports = Server;

362

lib/worker.js

@@ -1,61 +0,58 @@

'use strict'
const defaults = require('101/defaults')
const ErrorCat = require('error-cat')
const exists = require('101/exists')
const isNumber = require('101/is-number')
const isObject = require('101/is-object')
const merge = require('101/put')
const monitor = require('monitor-dog')
const pick = require('101/pick')
const Promise = require('bluebird')
/* global ErrorCat WorkerError DDTimer */
'use strict';
const TimeoutError = Promise.TimeoutError
const defaults = require('101/defaults');
const errorCat = require('error-cat');
const exists = require('101/exists');
const isNumber = require('101/is-number');
const isObject = require('101/is-object');
const merge = require('101/put');
const monitor = require('monitor-dog');
const pick = require('101/pick');
const Promise = require('bluebird');
const WorkerStopError = require('error-cat/errors/worker-stop-error');
const logger = require('./logger')
const TaskFatalError = require('./errors/task-fatal-error')
const logger = require('./logger');
const TimeoutError = Promise.TimeoutError;
/**
* Worker class: performs tasks for jobs on a given queue.
* @class
* Performs tasks for jobs on a given queue.
*
* @author Bryan Kendall
* @author Ryan Sandor Richards
* @module ponos:worker
* @param {object} opts Options for the worker.
* @param {string} opts.queue Name of the queue for the job the worker
* is processing.
* @param {function} opts.task A function to handle the tasks.
* @param {object} opts.job Data for the job to process.
* @param {function} opts.done Callback to execute when the job has successfully
* @param {Object} opts Options for the worker.
* @param {Function} opts.done Callback to execute when the job has successfully
* been completed.
* @param {boolean} [opts.runNow] Whether or not to run the job immediately,
* defaults to `true`.
* @param {Object} opts.job Data for the job to process.
* @param {String} opts.queue Name of the queue for the job the worker is
* processing.
* @param {Function} opts.task A function to handle the tasks.
* @param {ErrorCat} [opts.errorCat] An error-cat instance to use for the
* worker.
* @param {bunyan} [opts.log] The bunyan logger to use when logging messages
* from the worker.
* @param {ErrorCat} [opts.errorCat] An error-cat instance to use for the
* worker.
* @param {number} [opts.msTimeout] A specific millisecond timeout for this
* worker.
* @param {boolean} [opts.runNow] Whether or not to run the job immediately,
* defaults to `true`.
*/
class Worker {
constructor (opts) {
constructor(opts) {
// managed required fields
const fields = [
'done',
'job',
'queue',
'task'
]
const fields = ['done', 'job', 'queue', 'task'];
fields.forEach(function (f) {
if (!exists(opts[f])) {
throw new Error(f + ' is required for a Worker')
throw new Error(f + ' is required for a Worker');
}
})
});
// manage field defaults
fields.push('errorCat', 'log', 'msTimeout', 'runNow')
opts = pick(opts, fields)
fields.push('errorCat', 'log', 'msTimeout', 'runNow', 'server');
opts = pick(opts, fields);
defaults(opts, {
// default non-required user options
errorCat: new ErrorCat(),
errorCat: errorCat,
log: logger.child({ module: 'ponos:worker' }),

@@ -67,20 +64,20 @@ runNow: true,

retryDelay: process.env.WORKER_MIN_RETRY_DELAY || 1
})
});
// put all opts on this
Object.assign(this, opts)
this.log.info({ queue: this.queue, job: this.job }, 'Worker created')
Object.assign(this, opts);
this.log.info({ queue: this.queue, job: this.job }, 'Worker created');
// Ensure that the `msTimeout` option is valid
this.msTimeout = parseInt(this.msTimeout, 10)
this.msTimeout = parseInt(this.msTimeout, 10);
if (!isNumber(this.msTimeout)) {
throw new Error('Provided `msTimeout` is not an integer')
throw new Error('Provided `msTimeout` is not an integer');
}
if (this.msTimeout < 0) {
throw new Error('Provided `msTimeout` is negative')
throw new Error('Provided `msTimeout` is negative');
}
if (this.runNow) {
this.run()
this.run();
}

@@ -90,163 +87,176 @@ }

/**
* Factory method for creating new workers. This method exists to make it easier
* to unit test other modules that need to instantiate new workers.
* Factory method for creating new workers. This method exists to make it
* easier to unit test other modules that need to instantiate new workers.
*
* @see Worker
* @param {object} opts Worker options.
* @param {Object} opts Options for the Worker.
* @returns {Worker} New Worker.
*/
static create (opts) {
return new Worker(opts)
static create(opts) {
return new Worker(opts);
}
/**
* Runs the worker. If the task for the job fails, then this method will retry
* the task (with an exponential backoff) as set by the environment.
*
* @returns {Promise} Promise that is resolved once the task succeeds or
* fails.
*/
run() {
const log = this.log.child({
method: 'run',
queue: this.queue,
job: this.job
});
this._incMonitor('ponos');
const timer = this._createTimer();
return Promise.resolve().bind(this).then(function runTheTask() {
const attemptData = {
attempt: this.attempt++,
timeout: this.msTimeout
};
log.info(attemptData, 'running task');
let taskPromise = Promise.resolve().bind(this).then(() => {
return Promise.resolve().bind(this).then(() => {
return this.task(this.job, this.server);
});
});
if (this.msTimeout) {
taskPromise = taskPromise.timeout(this.msTimeout);
}
return taskPromise;
}).then(function successDone(result) {
log.info({ result: result }, 'Task complete');
this._incMonitor('ponos.finish', { result: 'success' });
return this.done();
})
// if the type is TimeoutError, we will log and retry
.catch(TimeoutError, function timeoutErrRetry(err) {
log.warn({ err: err }, 'Task timed out');
this._incMonitor('ponos.finish', { result: 'timeout-error' });
// by throwing this type of error, we will retry :)
throw err;
}).catch(function decorateError(err) {
if (!isObject(err.data)) {
err.data = {};
}
if (!err.data.queue) {
err.data.queue = this.queue;
}
if (!err.data.job) {
err.data.job = this.job;
}
throw err;
})
// if it's a WorkerStopError, we can't accomplish the task
.catch(WorkerStopError, function knownErrDone(err) {
log.error({ err: err }, 'Worker task fatally errored');
this._incMonitor('ponos.finish', { result: 'fatal-error' });
this._reportError(err);
// If we encounter a fatal error we should no longer try to schedule
// the job.
return this.done();
}).catch(function unknownErrRetry(err) {
const attemptData = {
err: err,
nextAttemptDelay: this.retryDelay
};
log.warn(attemptData, 'Task failed, retrying');
this._incMonitor('ponos.finish', { result: 'task-error' });
this._reportError(err);
// Try again after a delay
return Promise.delay(this.retryDelay).bind(this).then(function retryRun() {
// Exponentially increase the retry delay
if (this.retryDelay < process.env.WORKER_MAX_RETRY_DELAY) {
this.retryDelay *= 2;
}
return this.run();
});
}).finally(function stopTimer() {
if (timer) {
timer.stop();
}
});
}
// Private Methods
/**
* Helper function for reporting errors to rollbar via error-cat.
*
* @private
* @param {error} err Error to report.
* @param {Error} err Error to report.
*/
_reportError (err) {
err.data = isObject(err.data) ? err.data : {}
err.data.queue = this.queue
err.data.job = this.job
this.errorCat.report(err)
_reportError(err) {
this.errorCat.report(err);
}
/**
* Helper function for creating monitor-dog events tags
* `queue` is the only mandatory tag.
* Few tags would be created depending on the queue name
* If queueName use `.` as delimiter e.x. `10.0.0.20.api.github.push` then
* following tags would be created:
* - token0: 'push'
* - token1: 'github.push'
* - token2: 'api.github.push'
* - token3: '10.0.0.20.api.github.push'
* Helper function for creating monitor-dog events tags. `queue` is the only
* mandatory tag. Few tags will be created depending on the queue name. If
* queueName use `.` as delimiter e.x. `10.0.0.20.api.github.push` then the
* following tags will be created:
* {
* token0: 'push'
* token1: 'github.push'
* token2: 'api.github.push'
* token3: '10.0.0.20.api.github.push'
* }
*
* @private
* @returns {Object} tags as Object {queue: 'docker.event.publish'}
* @returns {Object} tags as Object { queue: 'docker.event.publish' }.
*/
_eventTags () {
const tokens = this.queue.split('.').reverse()
let lastToken = ''
let tags = tokens.reduce(function (acc, currentValue, currentIndex) {
const key = 'token' + currentIndex
const newToken = currentIndex === 0
? currentValue
: currentValue + '.' + lastToken
acc[key] = newToken
lastToken = newToken
return acc
}, {})
tags.queue = this.queue
return tags
_eventTags() {
const tokens = this.queue.split('.').reverse();
let lastToken = '';
let tags = tokens.reduce((acc, currentValue, currentIndex) => {
const key = 'token' + currentIndex;
const newToken = currentIndex === 0 ? currentValue : currentValue + '.' + lastToken;
acc[key] = newToken;
lastToken = newToken;
return acc;
}, {});
tags.queue = this.queue;
return tags;
}
/**
* Helper function calling `monitor.increment`.
* Monitor wouldn't be called if `process.env.WORKER_MONITOR_DISABLED` set.
* @param {string} eventName name to be reported into the datadog
* @param {object} [extraTags] extra tags to be send with the event
* Helper function calling `monitor.increment`. Monitor won't be called if
* `WORKER_MONITOR_DISABLED` is set.
*
* @private
* @param {String} eventName Name to be reported into the datadog.
* @param {Object} [extraTags] Extra tags to be send with the event.
*/
_incMonitor (eventName, extraTags) {
_incMonitor(eventName, extraTags) {
if (process.env.WORKER_MONITOR_DISABLED) {
return
return;
}
let tags = this._eventTags()
let tags = this._eventTags();
if (extraTags) {
tags = merge(tags, extraTags)
tags = merge(tags, extraTags);
}
monitor.increment(eventName, tags)
monitor.increment(eventName, tags);
}
/**
* Helper function calling `monitor.timer`.
* Timer wouldn't be created if `process.env.WORKER_MONITOR_DISABLED` set
* @return {object} new timer
* Helper function calling `monitor.timer`. Timer won't be created if
* `WORKER_MONITOR_DISABLED` is set.
*
* @return {Object} New timer.
* @private
*/
_createTimer () {
const tags = this._eventTags()
return !process.env.WORKER_MONITOR_DISABLED
? monitor.timer('ponos.timer', true, tags)
: null
_createTimer() {
const tags = this._eventTags();
return !process.env.WORKER_MONITOR_DISABLED ? monitor.timer('ponos.timer', true, tags) : null;
}
/**
* Runs the worker. If the task for the job fails, then this method will retry
* the task (with an exponential backoff) a number of times defined by the
* environment of the process.
* @returns {promise} Promise resolved once the task succeeds or fails.
*/
run () {
const log = this.log.child({
method: 'run',
queue: this.queue,
job: this.job
})
this._incMonitor('ponos')
const timer = this._createTimer()
return Promise.resolve().bind(this)
.then(function runTheTask () {
const attemptData = {
attempt: this.attempt++,
timeout: this.msTimeout
}
log.info(attemptData, 'running task')
let taskPromise = Promise.resolve().bind(this)
.then(() => {
return Promise.resolve().bind(this)
.then(() => { return this.task(this.job) })
})
if (this.msTimeout) {
taskPromise = taskPromise.timeout(this.msTimeout)
}
return taskPromise
})
.then(function successDone (result) {
log.info({ result: result }, 'Task complete')
this._incMonitor('ponos.finish', { result: 'success' })
return this.done()
})
// if the type is TimeoutError, we will log and retry
.catch(TimeoutError, function timeoutErrRetry (err) {
log.warn({ err: err }, 'Task timed out')
this._incMonitor('ponos.finish', { result: 'timeout-error' })
// by throwing this type of error, we will retry :)
throw err
})
// if it's a known type of error, we can't accomplish the task
.catch(TaskFatalError, function knownErrDone (err) {
log.error({ err: err }, 'Worker task fatally errored')
this._incMonitor('ponos.finish', { result: 'fatal-error' })
this._reportError(err)
// If we encounter a fatal error we should no longer try to schedule
// the job.
return this.done()
})
.catch(function unknownErrRetry (err) {
const attemptData = {
err: err,
nextAttemptDelay: this.retryDelay
}
log.warn(attemptData, 'Task failed, retrying')
this._incMonitor('ponos.finish', { result: 'task-error' })
this._reportError(err)
// Try again after a delay
return Promise.delay(this.retryDelay).bind(this)
.then(function retryRun () {
// Exponentially increase the retry delay
if (this.retryDelay < process.env.WORKER_MAX_RETRY_DELAY) {
this.retryDelay *= 2
}
return this.run()
})
})
.finally(function stopTimer () {
if (timer) {
timer.stop()
}
})
}
}
module.exports = Worker
/**
* Worker class.
* @module ponos/lib/worker
* @see Worker
*/
module.exports = Worker;

37

package.json
{
"name": "ponos",
"version": "2.0.0",
"version": "3.0.0-rc1",
"description": "An opinionated queue based worker server for node.",
"main": "index.js",
"main": "lib/index.js",
"engines": {
"node": ">=4 <5"
"node": ">=4"
},
"scripts": {
"build": "babel --out-dir lib src",
"build:clean": "rm -rf lib",
"changelog": "github-changes -o Runnable -r ponos -a --only-pulls --use-commit-body --order-semver",

@@ -14,6 +16,9 @@ "coverage": "istanbul cover ./node_modules/.bin/_mocha -- $npm_package_options_mocha test/unit && npm run coverage-check",

"coveralls": "cat ./coverage/lcov.info | coveralls",
"docs": "jsdoc --recurse --readme ./README.md lib/",
"docs": "npm run build && jsdoc --recurse --readme ./README.md lib/",
"format": "standard --format",
"functional": "mocha $npm_package_options_mocha test/functional",
"lint": "standard",
"lint": "npm run lint:format && npm run lint:type",
"lint:format": "standard --verbose",
"lint:type": "flow --timeout 30",
"prepublish": "not-in-install && npm run build || in-install",
"test": "npm run lint && npm run unit && npm run functional",

@@ -27,3 +32,3 @@ "unit": "mocha $npm_package_options_mocha test/unit"

"options": {
"mocha": "--require resources/mocha-bootstrap --recursive --reporter spec --bail --timeout 5000"
"mocha": "--require resources/mocha-bootstrap --recursive --reporter spec --bail --timeout 5000 --compilers js:babel-register"
},

@@ -47,2 +52,3 @@ "keywords": [

"standard": {
"parser": "babel-eslint",
"globals": [

@@ -59,20 +65,29 @@ "describe",

"101": "^1.1.1",
"amqplib": "^0.4.1",
"bluebird": "^3.0.5",
"bunyan": "^1.5.1",
"error-cat": "^1.3.4",
"es6-error": "^2.0.2",
"error-cat": "^2.0.4",
"immutable": "^3.8.1",
"monitor-dog": "1.5.0",
"runnable-hermes": "^6.4.0"
"uuid": "^2.0.2"
},
"devDependencies": {
"babel-cli": "^6.8.0",
"babel-eslint": "^6.0.4",
"babel-plugin-transform-class-properties": "^6.8.0",
"babel-plugin-transform-flow-strip-types": "^6.8.0",
"babel-register": "^6.8.0",
"chai": "^3.3.0",
"chai-as-promised": "^5.1.0",
"coveralls": "^2.11.4",
"flow-bin": "^0.24.2",
"github-changes": "^1.0.0",
"istanbul": "^0.4.0",
"in-publish": "^2.0.0",
"istanbul": "^1.0.0-alpha.2",
"jsdoc": "^3.4.0",
"mocha": "^2.3.3",
"sinon": "^1.17.0",
"standard": "^6.0.7"
"sinon-as-promised": "^4.0.0",
"standard": "^7.0.1"
}
}

145

README.md

@@ -7,2 +7,3 @@ # Ponos

[![devdependencies]](https://david-dm.org/Runnable/ponos#info=devDependencies)
[![codeclimate]](https://codeclimate.com/github/Runnable/ponos)

@@ -15,15 +16,14 @@ Documentation is available at [runnable.github.io/ponos][documentation]

options | environment | default
----------------|---------------------|--------------
`opts.hostname` | `RABBITMQ_HOSTNAME` | `'localhost'`
`opts.port` | `RABBITMQ_PORT` | `'5672'`
`opts.username` | `RABBITMQ_USERNAME` | `'guest'`
`opts.password` | `RABBITMQ_PASSWORD` | `'guest'`
`opts.log` | *N/A* | Basic [bunyan](https://github.com/trentm/node-bunyan) instance with `stdout` stream (for logging)
`opts.errorCat` | *N/A* | Basic [error-cat](https://github.com/runnable/error-cat) instance (for rollbar error reporting)
options | environment | default
-------------------------|---------------------|--------------
`opts.rabbitmq.hostname` | `RABBITMQ_HOSTNAME` | `'localhost'`
`opts.rabbitmq.port` | `RABBITMQ_PORT` | `'5672'`
`opts.rabbitmq.username` | `RABBITMQ_USERNAME` | _none_
`opts.rabbitmq.password` | `RABBITMQ_PASSWORD` | _none_
`opts.log` | _N/A_ | Basic [bunyan](https://github.com/trentm/node-bunyan) instance with `stdout` stream (for logging)
`opts.errorCat` | _N/A_ | Basic [error-cat](https://github.com/runnable/error-cat) instance (for rollbar error reporting)
Other options for Ponos are as follows:
Environment | default | description
environment variable | default | description
-------------------------|---------|------------

@@ -34,3 +34,2 @@ `WORKER_MAX_RETRY_DELAY` | `0` | The maximum time, in milliseconds, that the worker will wait before retrying a task. The timeout will exponentially increase from `MIN_RETRY_DELAY` to `MAX_RETRY_DELAY` if the latter is set higher than the former. If this value is not set, the worker will not exponentially back off.

## Usage

@@ -49,5 +48,5 @@

return Promise.resolve()
.then(function () {
return doSomeWork(job);
});
.then(() => {
return doSomeWork(job)
})
}

@@ -58,50 +57,71 @@ ```

## Tasks vs. Events
Ponos provides (currently) two paradigms for doing work. First is subscribing directly to queues in RabbitMQ using the `tasks` parameter in the constructor. The other is the ability to subscribe to a fanout exchange using the `events` parameter, which can provide for a much more useful utilization of RabbitMQ's structure.
```javascript
const ponos = require('ponos')
const server = new ponos.Server({
tasks: {
'a-queue': (job) => { return Promise.resolve(job) }
},
events: {
'an-exchange': (job) => { return Promise.resolve(job) }
}
})
```
### Worker Errors
Ponos's worker is designed to retry any error that is not specifically a fatal error. A fatal error is defined with the `TaskFatalError` class. If a worker rejects with a `TaskFatalError`, the worker will automatically assume the job can _never_ be completed and will acknowledge the job.
Ponos's worker is designed to retry any error that is not specifically a fatal error. Ponos has been designed to work well with our error library [`error-cat`](https://github.com/Runnable/error-cat).
As an example, a `TaskFatalError` can be used to fail a task given an invalid job:
A fatal error is defined with the `WorkerStopError` class from `error-cat`. If a worker rejects with a `WorkerStopError`, the worker will automatically assume the job can _never_ be completed and will acknowledge the job.
As an example, a `WorkerStopError` can be used to fail a task given an invalid job:
```javascript
var TaskFatalError = require('ponos').TaskFatalError;
const WorkerStopError = require('error-cat/errors/worker-stop-error')
function myWorker (job) {
return Promise.resolve()
.then(function () {
if (!job.host) { throw new TaskFatalError('host is required'); }
.then(() => {
if (!job.host) {
throw new WorkerStopError('host is required', {}, 'my.queue', job)
}
})
.then(function () {
return doSomethingWithHost(job);
.then(() => {
return doSomethingWithHost(job)
})
.catch(function (err) {
myErrorReporter(err);
throw err;
});
}
```
This worker will reject the promise with a `TaskFatalError`. Ponos will log the error itself, acknowledge the job to remove it from the queue, and continue with other jobs. It is up to the user to additionally catch the error and log it to any other external service.
This worker will reject the promise with a `WorkerStopError`. Ponos will log the error itself, acknowledge the job to remove it from the queue, and continue with other jobs. You may catch and re-throw the error if you wish to do additional logging or reporting.
Finally, as was mentioned before, Ponos will retry any other errors. Ponos provides a `TaskError` class you may use, or you may throw normal `Error`s. If you do, the worker will catch these and retry according to the server's configuration (retry delay, back-off, max delay, etc.).
Finally, as was mentioned before, Ponos will retry any other errors. `error-cat` provides a `WorkerError` class you may use, or you may throw normal `Error`s. If you do, the worker will catch these and retry according to the server's configuration (retry delay, back-off, max delay, etc.).
```javascript
var TaskError = require('ponos').TaskError;
var TaskFatalError = require('ponos').TaskFatalError;
const WorkerError = require('error-cat/errors/worker-error')
const WorkerStopError = require('error-cat/errors/worker-stop-error')
function myWorker (job) {
return Promise.resolve()
.then(function () {
return externalService.something(job);
.then(() => {
return externalService.something(job)
})
// Note: w/o this catch, the error would simply propagate to the worker and
// be handled.
.catch(function (err) {
logErrorToService(err);
.catch((err) => {
logErrorToService(err)
// If the error is 'recoverable' (e.g., network fluke, temporary outage),
// we want to be able to retry.
if (err.isRecoverable) {
throw new Error('hit a momentary issue. try again.');
throw new Error('hit a momentary issue. try again.')
} else {
// maybe we know we can't recover from this
throw new TaskFatalError('cannot recover. acknowledge and remove job');
throw new WorkerStopError(
'cannot recover. acknowledge and remove job',
{},
'this.queue',
job
)
}
});
})
}

@@ -111,8 +131,7 @@ ```

## Worker Options
Currently workers can be defined with a `msTimeout` option. This value defaults to
`process.env.WORKER_TIMEOUT || 0`. One can set a specific millisecond timeout for
a worker like so:
Currently workers can be defined with a `msTimeout` option. This value defaults to `process.env.WORKER_TIMEOUT || 0`. One can set a specific millisecond timeout for a worker like so:
```js
server.setTask('my-queue', workerFunction, { msTimeout: 1234 });
server.setTask('my-queue', workerFunction, { msTimeout: 1234 })
```

@@ -131,36 +150,37 @@

}
});
})
```
These options are also available for `setEvent` and `setAllEvents`.
## Full Example
```javascript
var ponos = require('ponos');
const ponos = require('ponos')
var tasks = {
'queue-1': function (job) { return Promise.resolve(job); },
'queue-2': function (job) { return Promise.resolve(job); }
};
const tasks = {
'queue-1': (job) => { return Promise.resolve(job) },
'queue-2': (job) => { return Promise.resolve(job) }
}
const events = {
'exchange-1': (job) => { return Promise.resolve(job) }
}
// Create the server
var server = new ponos.Server({
queues: Object.keys(tasks);
});
events: events,
tasks: tasks
})
// Set tasks for workers handling jobs on each queue
server.setAllTasks(tasks);
// If tasks were not provided in the constructor, set tasks for workers handling
// jobs on each queue
server.setAllTasks(tasks)
// Similarly, you can set events.
server.setAllEvents(events)
// Start the server!
server.start()
.then(function () { console.log('Server started!'); })
.catch(function (err) { console.error('Server failed', err); });
// Or, start using your own hermes client
var hermes = require('runnable-hermes');
var server = new ponos.Server({ hermes: hermes.hermesSingletonFactory({...}) });
// You can also nicely chain the promises!
server.start()
.then(function () { /*...*/ })
.catch(function (err) { /*...*/ });
.then(() => { console.log('Server started!') })
.catch((err) => { console.error('Server failed', err) })
```

@@ -172,3 +192,3 @@

[travis]: https://img.shields.io/travis/Runnable/ponos.svg?style=flat-square "Build Status"
[travis]: https://img.shields.io/travis/Runnable/ponos/master.svg?style=flat-square "Build Status"
[coveralls]: https://img.shields.io/coveralls/Runnable/ponos/master.svg?style=flat-square "Coverage Status"

@@ -178,1 +198,2 @@ [dependencies]: https://img.shields.io/david/Runnable/ponos.svg?style=flat-square "Dependency Status"

[documentation]: https://runnable.github.io/ponos "Ponos Documentation"
[codeclimate]: https://img.shields.io/codeclimate/github/Runnable/ponos.svg?style=flat-square "Code Climate"
index.js
lib/errors/task-error.js
lib/errors/task-fatal-error.js
resources/generate-docs.sh
.npmignore

Sorry, the diff of this file is not supported yet

SocketSocket SOC 2 Logo

Product

  • Package Alerts
  • Integrations
  • Docs
  • Pricing
  • FAQ
  • Roadmap
  • Changelog

About

Packages

npm

Go

Maven

PyPI

Rubygems

Stay in touch

Get open source security insights delivered straight into your inbox.

  • Terms
  • Privacy
  • Security

Made with ⚡️ by Socket Inc