		"use strict";

		var fs = require( 'fs' );
		var dronosInstance = null;

		module.exports.init = function ( config ) {

		// should only run one dronos per process
		if ( dronosInstance ) {
		return dronosInstance;
		}

		config = {
		libDir: './lib',
		collectionPrefix: config.collectionPrefix \|\| null,
		ensureEntries: config.ensureEntries \|\| [],
		deleteEntries: config.deleteEntries \|\| [],
		mongoose: config.mongoose \|\| false
		};

		if ( !config.mongoose ) {
		return false;
		}

		var context = {
		options: config
		};

		if ( typeof context.options.collectionPrefix !== 'string' ) {
		context.options.collectionPrefix = "";
		}

		if ( !Array.isArray( context.options.ensureEntries ) ) {
		context.options.ensureEntries = [];
		}

		context.DronosModel = require( './models/Dronos.js' ).init( {
		mongoose: config.mongoose,
		prefix: context.options.collectionPrefix
		} );

		if ( !context.DronosModel ) {
		return false;
		}

		dronosInstance = startUp( context );

		return dronosInstance;

		};

		function startUp( context ) {

		var dronos = new Dronos( context );

		dronos.ensureEntries();
		dronos.deleteEntries();

		dronos.run();

		return dronos;

		}

		var Dronos = function ( context ) {

		this.context = context;

		};

		Dronos.prototype.deleteEntries = function () {
		var entries = this.context.options.deleteEntries;

		for ( var i = 0; i < entries.length; i++ ) {
		this._deleteEntry( entries[i] );
		}
		};

		Dronos.prototype.ensureEntries = function () {

		var entries = this.context.options.ensureEntries;

		for ( var i = 0; i < entries.length; i++ ) {
		this.ensureEntry( entries[i] );
		}

		};

		Dronos.prototype.ensureEntry = function ( entry ) {

		// using serialization to clone
		var template = {
		account: null,
		name: null,
		recurrencePattern: {
		minute: null,
		hour: null,
		dayOfMonth: null,
		month: null,
		dayOfWeek: null
		},
		lib: null,
		method: null,
		processingLimit: null
		};

		// create default by cloning template, then mixing in the entry
		entry = mixin( clone( template ), entry );

		if ( !this._verifyDataContainsRequiredFields( template, entry ) ) {
		return false;
		}

		this._ensureChronEntry( entry );

		return true;

		};

		Dronos.prototype._deleteEntry = function ( entry ) {

		entry = {
		account: entry.account \|\| null,
		name: entry.name \|\| null
		};

		if ( entry.account !== null && entry.name !== null ) {
		this.context.DronosModel.remove(
		{
		account: entry.account,
		name: entry.name
		}
		).exec();
		return true;
		}

		return false;

		};

		Dronos.prototype._ensureChronEntry = function ( entry ) {

		var self = this;

		this.context.DronosModel.findOneAndUpdate(
		{
		account: entry.account,
		name: entry.name
		},
		{
		account: entry.account,
		name: entry.name,
		lastUpdate: new Date(),
		repeat: true,
		recurrencePattern: {
		minute: entry.recurrencePattern.minute,
		hour: entry.recurrencePattern.hour,
		dayOfMonth: entry.recurrencePattern.dayOfMonth,
		month: entry.recurrencePattern.month,
		dayOfWeek: entry.recurrencePattern.dayOfWeek
		},
		lib: entry.lib,
		method: entry.method,
		processingLimit: entry.processingLimit,
		params: ( entry.params ? entry.params : null )
		},
		{
		new: true,
		upsert: true
		},
		function ( err, data ) {
		if ( data && data._id ) {

		var threshold = new Date();
		threshold.setSeconds( threshold.getSeconds() + 15 );

		self.context.DronosModel.update(
		{
		_id: data._id,
		$or: [
		{
		nextRun: {
		$gt: threshold
		}
		},
		{
		nextRun: {
		$exists: false
		}
		}
		]
		},
		{
		nextRun: _getNextRunTime( entry.recurrencePattern ),
		lastUpdate: new Date()
		}
		).exec();

		}
		}

		);

		};

		Dronos.prototype._verifyDataContainsRequiredFields = function ( template, data, path ) {

		if ( !path ) {
		path = "";
		}

		for ( var field in template ) {
		if ( template.hasOwnProperty( field ) ) {
		var val = template[field];

		// check for scalar in template
		if ( val === null ) {
		if ( typeof data[field] === 'undefined' \|\| data[field] === null ) {
		return false;
		}
		}
		else if ( Object.isObject( val ) ) {

		// data not an object, so don't bother recurse
		if ( !Object.isObject( data[field] ) ) {
		return false;
		}

		// recurse
		if ( !this._verifyDataContainsRequiredFields( val, data[field], path + field + "." ) ) {
		return false;
		}

		}
		else {
		return false;
		}
		}
		}

		return true;
		};

		Dronos.prototype.run = function () {

		//console.log( process.pid, 'Dronos: start running', new Date().toISOString() );

		var self = this;

		this.context.runningIds = [];

		this._resetBrokenEntries();

		// run entries, until no more are ready for the current minute, then sleep
		this._executeEntries( function () {

		// make sure run gets called again
		self._nextRunTimeout();

		} );

		};

		Dronos.prototype._resetBrokenEntries = function () {

		var threshold = new Date();

		// no process should execute for more than 15 seconds, so if it still has processing count above a 30 second threshold, clear stuff
		threshold.setSeconds( threshold.getSeconds() - 45 );

		this.context.DronosModel.update(
		{
		$or: [
		{
		nextRun: { $lt: threshold }, // if entry should have finished, but process count still high, assume node process failed to decrement
		processingCount: { $gt: 0 }
		},
		{
		lastUpdate: { $lt: threshold }, // if entry last update was more than 30 seconds ago, but processing count still up, decrement
		processingCount: { $gt: 0 }
		},
		{
		processingCount: { $lt: 0 } // processing count should never be below 0. can't have negative number of processing working
		}
		]
		},
		{
		$set: {
		processingCount: 0
		},
		lastUpdate: new Date()
		},
		{
		multi: true
		}
		).exec( function ( err, count ) {
		if ( count ) {
		console.log( process.pid, 'Dronos: reset broken entries', count );
		}
		} );

		};

		Dronos.prototype._executeEntries = function ( callback, time ) {

		var self = this;

		if ( !time ) {

		time = new Date();

		// we run right on the minute, and servers may be off in
		// time by a couple seconds, but less than 60 seconds. this
		// makes sure we aren't missing items for the current second
		// if this server's time is actually a little slow
		time.setSeconds( time.getSeconds() + 10 );

		}

		this.context.DronosModel.findOneAndUpdate(
		{
		_id: { $nin: this.context.runningIds },
		nextRun: { $lte: time }, // must be ready to run
		$or: [
		// must have less than max limit processes running, or unlimited processing limit
		{
		processingLimit: 0
		},
		{
		$where: 'this.processingCount < this.processingLimit'
		},
		{
		processingCount: {
		$exists: false
		}
		}
		]
		},
		{
		$inc: { processingCount: 1 },
		lastUpdate: new Date()
		},
		{
		sort: {
		nextRun: 1, // ones waiting longest get priority
		processingCount: 1 // if tied for longest waiting, one with fewest running gets priority
		},
		new: true
		},
		function ( err, data ) {

		// no more entries
		if ( !data ) {
		process.nextTick( callback );
		return;
		}

		data = data.toObject ? data.toObject() : {};

		var entry = {
		_id: data._id \|\| null,
		lib: data.lib \|\| null,
		method: data.method \|\| null,
		params: data.params \|\| null
		};

		// no entry, we are done
		if ( entry._id === null ) {
		process.nextTick( callback );
		}
		else {

		// make sure we don't run multiple instances of the same entry
		self.context.runningIds.push( entry._id );

		// start the entry running
		self._executeEntry( entry );

		// break up execution a little
		process.nextTick( function () {

		// if we just ran an entry, we always look for one more runable entry and forward the callback down
		self._executeEntries( callback, time );

		} );

		}

		}
		);

		};

		Dronos.prototype._executeEntry = function ( entry, callback ) {

		var lib = (process.cwd() + '/' + this.context.options.libDir + '/' + entry.lib).replace( /\/+/g, '/' );
		var method = entry.method;
		var params = entry.params;
		if ( !lib \|\| !method ) {
		return;
		}
		var self = this;
		var schedulerUtils = {
		nextRunTime: _getNextRunTime,
		incrementTime: _incrementTime
		};
		var doneCalled = false;
		var doneTimeout = null;

		var done = function () {

		if ( doneTimeout ) {
		clearTimeout( doneTimeout );
		}

		if ( doneCalled ) {
		console.error( process.pid, 'Dronos: done already called for execution of lib', entry.lib, method, new Date().toISOString() );
		return;
		}
		doneCalled = true;

		process.nextTick( function () {

		if ( entry._id ) {
		if ( entry.repeat ) {

		// build update object
		var update = {
		$inc: { processingCount: -1 },
		lastRun: new Date(),
		lastUpdate: new Date()
		};
		if ( entry.recurrencePattern ) {
		update.nextRun = _getNextRunTime( entry.recurrencePattern );
		}

		self.context.DronosModel.findByIdAndUpdate( entry._id, update, { new: true }, function () {
		if ( typeof callback === 'function' ) {
		process.nextTick( callback );
		}
		} );

		}
		else {

		self.context.DronosModel.findByIdAndRemove( entry._id, function () {
		if ( typeof callback === 'function' ) {
		process.nextTick( callback );
		}
		} );

		}
		}

		} );

		};

		fs.exists( lib, function ( exists ) {

		if ( !exists ) {
		console.error( 'missing lib', entry.lib );
		done();
		return;
		}

		var module = null;
		try {
		module = require( lib );
		}
		catch ( e ) {
		console.error( 'exception lib', entry.lib, e );
		done();
		return;
		}

		if ( !module \|\| typeof module[method] !== 'function' ) {
		console.error( 'missing function', entry.lib, method );
		done();
		return;
		}

		process.nextTick( function () {
		//console.log( process.pid, 'Dronos: executing', entry.lib + "." + method, new Date().toISOString() );
		module[method]( params, done, schedulerUtils );
		} );

		} );

		// triggered event code has 15 seconds to call done, before we consider it auto-done
		doneTimeout = setTimeout( function () {
		console.error( process.pid, 'Dronos: lib.method failed to call done in 15 seconds', entry.lib, method );
		done();
		}, 15000 );

		};

		Dronos.prototype._nextRunTimeout = function () {

		var now = new Date();
		var offset = clone( now );
		var self = this;
		var seconds = offset.getSeconds() + 1;
		var interval = 30; // number of seconds between cron interval check

		// calculate next interval in constant time
		var secondsOffset = interval - (seconds % interval);
		if ( secondsOffset >= interval ) {
		secondsOffset = 0;
		}

		offset.setSeconds( seconds + secondsOffset );
		offset.setMilliseconds( 0 );

		this.context.runTimeout = setTimeout( function () {
		self.context.runTimeout = null;
		self.run();
		}, offset.getTime() - now.getTime() );

		// console.log( process.pid, 'Dronos: done running', new Date().toISOString() );
		// console.log( process.pid, 'Dronos: nextRun', offset.toISOString() );
		};

		function _getNextRunTime( pattern ) {

		return _incrementTime( new Date(), pattern );

		}

		function _incrementTime( time, pattern ) {

		pattern = clone( pattern );

		_parseRecurrencePattern( pattern );

		if ( pattern.dayOfMonth.type !== 'all' && pattern.dayOfWeek.type !== 'all' ) {

		var dOM = _incrementTimeDayOfMonth( time, pattern );
		var dOW = _incrementTimeDayOfWeek( time, pattern );

		if ( dOM.getMilliseconds() < dOW.getMilliseconds() ) {
		return dOM;
		}
		else {
		return dOW;
		}

		}
		else {
		return _realIncrementTime( time, pattern );
		}

		}

		function _incrementTimeDayOfMonth( time, pattern ) {

		pattern = clone( pattern );
		pattern.dayOfWeek = { type: 'all' };

		return _realIncrementTime( time, pattern );

		}

		function _incrementTimeDayOfWeek( time, pattern ) {

		pattern = clone( pattern );
		pattern.dayOfMonth = { type: 'all' };

		return _realIncrementTime( time, pattern );

		}

		function _realIncrementTime( time, pattern ) {

		var newTime = clone( time );

		var secondsMilliSeconds = newTime.getSeconds() + newTime.getMilliseconds();

		// we always start at the top of the minute
		newTime.setSeconds( 0 );
		newTime.setMilliseconds( 0 );

		// round up to nearest minute
		if ( secondsMilliSeconds > 0 ) {
		newTime.setMinutes( newTime.getMinutes() + 1 );
		}

		_incrementField( newTime, 'FullYear', pattern.year );
		_incrementField( newTime, 'Month', pattern.month );

		// these are mutually exclusive at this level of processing
		if ( pattern.dayOfMonth.type !== 'all' ) {
		_incrementField( newTime, 'Date', pattern.dayOfMonth );
		}
		else if ( pattern.dayOfWeek.type !== 'all' ) {
		_incrementField( newTime, 'Day', pattern.dayOfWeek );
		}

		_incrementField( newTime, 'Hours', pattern.hour );
		_incrementField( newTime, 'Minutes', pattern.minute );

		return newTime;

		}

		function _incrementField( time, field, pattern ) {

		var currentValue = time['get' + field]();
		var newValue = _calculateNewValue( time, field, pattern );

		// if newValue is not an increment, nothing to do
		if ( newValue <= currentValue ) {
		return;
		}

		// clamp all units of time less than the current unit
		switch ( field ) {

		case 'FullYear':
		time.setMonth( 0 );

		/* falls through */
		case 'Month':
		time.setDate( 1 );

		/* falls through */
		case 'Day': // dayOfWeek
		case 'Date': // dayOfMonth
		time.setHours( 0 );

		/* falls through */
		case 'Hours':
		time.setMinutes( 0 );

		}

		// convert day of week value to day of month
		if ( field === 'Day' ) {
		field = 'Date';
		newValue = time.getDate() + _convertDayOfWeekOffsetToDateOffset( currentValue, newValue );
		}

		// apply increment to current field
		time['set' + field]( newValue );

		}

		function _convertDayOfWeekOffsetToDateOffset( currentValue, newValue ) {

		// no overflow, so its just the straight distance forward from currentValue to newValue
		if ( newValue >= currentValue ) {
		return newValue - currentValue;
		}

		// we overflowed, so calculate days until early day in next week
		return 7 - currentValue + newValue;

		}

		function _calculateNewValue( time, field, pattern ) {

		var type = pattern.type;
		var currentValue = time['get' + field]();

		var newValue = currentValue; // default is for no value to change, such as 'all' type

		// year doesn't overflow
		if ( field === 'FullYear' ) {
		return newValue;
		}

		// determine the minimum and maximum possible values for the current field
		var min = 0;
		var max = null;
		switch ( field ) {

		case 'Month':
		max = 11;
		break;

		case 'Day': // dayOfWeek
		max = 6;
		break;

		case 'Date': // dayOfMonth
		min = 1;
		max = _lastDayOfMonth( time );
		break;

		case 'Hours':
		max = 23;
		break;

		case 'Minutes':
		max = 59;
		break;

		}

		// months are stored in Date as ints 0-11, but for humans, we think of them as jan=1, mar=3, etc. so we convert to 1-12 scale for calculations
		if ( field === 'Month' ) {
		min++;
		max++;
		currentValue++;
		newValue++;
		}

		// handle last day of the time period for given field
		if ( type === 'last' ) {
		if ( field !== 'FullYear' ) { // year doesn't support last, unless you happen to know the year the earth will end.
		newValue = max;
		}
		}

		// handle literal values
		else if ( type === 'literals' ) {
		newValue = _nextLiteral( pattern.parts, currentValue );
		}

		// handle modulus
		else if ( type === 'modulus' && field !== 'FullYear' ) {
		newValue = _nextModulus( min, max, currentValue, pattern.mod );
		}

		newValue = _handleOverflow( min, max, currentValue, newValue );

		// undo month offsets used for calculations
		if ( field === 'Month' ) {
		newValue--;
		}

		return newValue;

		}

		function _handleOverflow( min, max, currentValue, newValue ) {

		// no overflow
		if ( newValue >= currentValue ) {
		return newValue;
		}

		// overflow, calculate offset
		return max + newValue - min + 1;

		}

		function _nextLiteral( literals, current ) {

		for ( var i = 0; i < literals.length; i++ ) {
		if ( literals[i] >= current ) {
		return literals[i];
		}
		}

		return literals[0];

		}

		function _nextModulus( start, end, current, mod ) {

		var originalCurrent = current;

		while ( current <= end ) {
		if ( current % mod === 0 ) {
		return current;
		}
		current++;
		}
		current = start;
		while ( current < originalCurrent ) {
		if ( current % mod === 0 ) {
		return current;
		}
		current++;
		}

		return null;

		}

		function _lastDayOfMonth( time ) {

		var value = null;
		switch ( time.getMonth() + 1 ) { // we don't need the +1, it just makes it easier to understand the cases, since we already think of jan = 1, mar = 3, etc.

		case 1:
		case 3:
		case 5:
		case 7:
		case 8:
		case 10:
		case 12:
		value = 31;
		break;

		case 4:
		case 6:
		case 9:
		case 11:
		value = 30;
		break;

		case 2:
		value = 28 + (time.getFullYear % 4 === 0 ? 1 : 0); // account for leap year
		break;

		}

		return value;

		}

		function _parseRecurrencePattern( recurrencePattern ) {

		var fields = ['minute', 'hour', 'dayOfMonth', 'dayOfWeek', 'month', 'year'];

		for ( var i = 0; i < fields.length; i++ ) {
		if ( recurrencePattern.hasOwnProperty( fields[i] ) ) {
		recurrencePattern[fields[i]] = _parseRecurrencePatternEntry( recurrencePattern[fields[i]] );
		}
		}

		}

		function _parseRecurrencePatternEntry( entry ) {

		// used throughout function
		var matches = null;

		// check for meta "last" indicator
		if ( entry.match( /^L$/ ) ) {
		return {
		type: 'last'
		};
		}

		// check for literal exact numbers
		if ( entry.match( /^[0-9,\-]+$/ ) ) {
		var split = entry.split( /,/ );
		var parts = [];

		// handle comma separation
		for ( var i = 0; i < split.length; i++ ) {

		// handle single value
		if ( split[i].match( /^[0-9]+$/ ) ) {
		parts.push( Math.floor( split[i] ) );
		}

		// handle range
		else {

		matches = split[i].match( /^([0-9]+)-([0-9]+)$/ );
		if ( matches && matches.length > 0 ) {
		var start = Math.min( matches[1], matches[2] );
		var end = Math.max( matches[1], matches[2] );
		for ( ; start <= end; start++ ) {
		parts.push( start );
		}

		}
		}

		}

		// see if there were valid combination of numbers and commas
		if ( parts.length > 0 ) {
		return {
		type: "literals",
		parts: parts
		};
		}

		}

		// check for modulus pattern
		matches = entry.match( /^\*\/([0-9]+)$/ );
		if ( matches && matches.length > 1 ) {

		return {
		type: "modulus",
		mod: Math.min( Math.floor( matches[1] ), 60 ) // modulus limited to 60 because of the units where modulus makes sense (minutes, hours, dayOfMonth), 60 is the highest modulus that doesn't clamp to 0.
		};

		}

		// default to * if nothing else matches
		return {
		type: "all"
		};

		}


		function isObject( thing ) {
		return typeof thing === 'object' && !Array.isArray( thing ) && thing !== null;
		}

		function clone( thing ) {

		var target = null;
		var type = typeof thing;

		if ( thing === undefined ) {
		target = undefined;
		}
		else if ( thing === null ) {
		target = null;
		}
		else if ( type === 'string' \|\| type === 'number' \|\| type === 'boolean' ) {
		target = thing;
		}
		else if ( thing instanceof Date ) {
		target = new Date( thing.toISOString() );
		}
		else if ( isObject( thing ) \|\| Array.isArray( thing ) ) {
		target = JSON.parse( JSON.stringify( thing ) ); // probably a slightly more efficient way to do thing, but thing is ok for now
		}
		else { // functions, etc. not clonable yet
		target = undefined;
		}

		return target;
		}

		function mixin() {

		var child = arguments[0];

		// Handle case when target is a string or something (possible in deep copy)
		if ( typeof child !== "object" && typeof child !== 'function' ) {
		child = {};
		}

		// handle arbitrary number of mixins. precedence is from last to first item passed in.
		for ( var i = 1; i < arguments.length; i++ ) {

		var parent = arguments[i];

		if ( !parent \|\| (!Array.isArray( parent ) && !isObject( parent ) ) ) {
		continue;
		}

		// Extend the base object
		for ( var name in parent ) {

		// don't copy parent stuffs
		if ( parent.hasOwnProperty( name ) ) {

		var target = child[ name ];
		var source = parent[ name ];

		// Prevent never-ending loop
		if ( child === source ) {
		continue;
		}

		// if target exists and is an array...
		if ( Array.isArray( target ) ) {

		if ( Array.isArray( source ) ) {

		// ...merge source array into target array
		for ( var j = 0; j < source.length; j++ ) {

		if ( typeof child[ name ][j] === 'object' ) {
		child[ name ][j] = mixin( child[ name ][j], source[j] );
		}
		else {
		child[ name ][j] = source[j];
		}

		}

		}

		}
		// if target is an object, try to mixin source
		else if ( isObject( target ) ) {

		mixin( target, source );

		}
		// otherwise, target becomes source
		else {

		child[ name ] = source;

		}
		}
		}

		}

		return child;
		}

		module.exports = require( './lib/dronos' );

package.json

		{
		"name": "dronos",
		"description": "Dronos is a distributed scheduling system (with patterns similar to Linux's cron system), using MongoDB to coordinate running tasks (drons) across multiple nodes.",
		"version": "0.0.1",
		"version": "0.0.3",
		"author": "Anthony Hildoer <anthony@bluerival.com>",
		@@ -11,5 +11,10 @@ "repository": {
		"dependencies": {
		"async": "0.9.0",
		"doublescore": "0.0.3",
		"later": "1.1.6",
		"moment": "2.8.3"
		},
		"devDependencies": {
		"mocha": "~1"
		"mocha": "2.0.1",
		"mongoose": "3.8.18"
		},
		@@ -24,5 +29,5 @@ "keywords": [
		"engines": {
		"node": ">=0.6.0"
		"node": ">=0.10.0"
		},
		"license": "MIT"
		}

README.md

		@@ -19,1 +19,26 @@ Dronos
		couple weeks, and everything should be ready for production (beta at least).


		License
		========

		(The MIT License)

		Copyright (c) 2013 BlueRival Software <anthony@bluerival.com>

		Permission is hereby granted, free of charge, to any person obtaining a copy of
		this software and associated documentation files (the 'Software'), to deal in
		the Software without restriction, including without limitation the rights to
		use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
		the Software, and to permit persons to whom the Software is furnished to do so,
		subject to the following conditions:

		The above copyright notice and this permission notice shall be included in all
		copies or substantial portions of the Software.

		THE SOFTWARE IS PROVIDED 'AS IS', WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
		IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
		FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
		COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
		IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
		CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

models/Dronos.js

.npmignore

Sorry, the diff of this file is not supported yet

dronos - npm Package Compare versions

New alerts

Fixed alerts

Improved metrics

Worsened metrics

Dependency changes