added the clock functionality

Author: iolufemi

package-lock.json

@@ -54,6 +54,7 @@
     "cacheman-redis": "^1.1.2",
     "cluster": "^0.7.7",
     "cors": "^2.8.3",
+    "cron": "^1.3.0",
     "crypto": "0.0.3",
     "debug": "^2.6.3",
     "express": "^4.15.2",

package.json

@@ -0,0 +1,47 @@
+"use strict";
+
+var db = require('../database').logMongo;
+
+var collection = 'Clock';
+
+var debug = require('debug')(collection);
+
+var schemaObject = {
+    job: {
+        type: 'String'
+    },
+    crontab: {
+        type: 'String'
+    },
+    name: {
+        type: 'String',
+        unique: true
+    },
+    enabled: {
+        type: 'Boolean',
+        default: true
+    },
+    arguments: {
+        type: db._mongoose.Schema.Types.Mixed
+    },
+    lastRunAt: {
+        type: db._mongoose.Schema.Types.Mixed
+    }
+};
+
+schemaObject.createdAt = {
+    type: 'Date',
+    default: Date.now
+};
+
+schemaObject.updatedAt = {
+    type: 'Date'
+    // default: Date.now
+};
+
+// Let us define our schema
+var Schema = new db._mongoose.Schema(schemaObject);
+
+var Model = db.model(collection, Schema);
+
+module.exports = Model;

services/queue/Model.js

@@ -1,12 +1,46 @@
-// {
-//     task
-//     crontab
-//     last runed at
-//     next run at
-//     name
-//     enabled
-//     arguments
-// }
+"use strict";
 
-// ToDo: work on a clock functionality so kue can support scheduled jobs
-// Todo: Use the cron package here https://www.npmjs.com/package/cron for timer
+var CronJob = require('cron').CronJob;
+var log = require('../logger');
+var Model = require('./Model');
+var queue = require('./');
+var _ = require('lodash');
+var config = require('../../config');
+
+log.info('Starting Queue Clock...');
+Model.find({enabled: true})
+.then(function(jobs){
+    _.forEach(jobs, function(job){
+        log.info('Initializing '+job.name+'...');
+        var cron = new CronJob({
+            cronTime: job.crontab,
+            onTick: function(){
+                // run this
+                // Check if job is enabled
+                Model.findOne({_id: job._id, enabled: true})
+                .then(function(resp){
+                    if(!resp){
+                        throw {notEnabled: true};
+                    }else{
+                        log.info('Pushing '+job.name+' to queue...');
+                        queue.create(job.job, job.arguments)
+                        .save();
+                    }
+                })
+                .catch(function(err){
+                    if(err.notEnabled){
+                        log.info(job.name+' is not enabled. Skipping...');
+                    }else{
+                        log.error('An error occured while running Job - '+job.name, err);
+                    }
+                });
+
+            },
+            start: true,
+            timeZone: config.clockTimezone
+        });
+    });
+})
+.catch(function(err){
+    log.error('An error occured while starting the queue clock: ', err);
+});

services/queue/clock.js

@@ -6,6 +6,7 @@ var queue = kue.createQueue({
   redis: config.redisURL
 });
 var log = require('../../services/logger');
+var Model = require('./Model');
 
 // Clean Up Completed Job
 queue
@@ -14,19 +15,19 @@ queue
 })
 .on('job complete', function(id, result){
     log.info('ID: ',id,' Result: ',result);
-  kue.Job.get(id, function(err, job){
-    if (err) {
-        return false;
-    }else{
-        job.remove(function(err){
-          if (err) {
-            throw new Error(err.message);
+    kue.Job.get(id, function(err, job){
+        if (err) {
+            return false;
         }else{
-            log.info('removed completed job #%d', job.id);
+            job.remove(function(err){
+              if (err) {
+                throw err;
+            }else{
+                log.info('removed completed job #%d', job.id);
+            }
+        });
         }
     });
-    }
-});
 });
 
 // Graceful Shutdown
@@ -57,3 +58,12 @@ queue.watchStuckJobs(1000);
 // Process Jobs Here
 module.exports = queue;
 module.exports.kue = kue;
+module.exports.addSchedule = function(crontab, name, job, data){
+    Model.create({crontab: crontab, name: name, job: job, arguments: data})
+    .then(function(){
+        // Silencio es dorado
+    })
+    .catch(function(err){
+        log.error('Error scheduling job - ', err);
+    });
+};