diff --git a/.gitignore b/.gitignore index fc1136152..741aedaf7 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ datastore +tasks www/bower_components/* node_modules /config.js diff --git a/server.js b/server.js index a51c94343..5a2cbfbb8 100644 --- a/server.js +++ b/server.js @@ -9,6 +9,7 @@ var WebSocketServer = require('ws').Server; var NetfluxSrv = require('./node_modules/chainpad-server/NetfluxWebsocketSrv'); var Package = require('./package.json'); var Path = require("path"); +var nThen = require("nthen"); var config; try { @@ -198,32 +199,37 @@ if (config.httpSafePort) { var wsConfig = { server: httpServer }; -var createSocketServer = function (err, rpc) { - if(!config.useExternalWebsocket) { - if (websocketPort !== config.httpPort) { - console.log("setting up a new websocket server"); - wsConfig = { port: websocketPort}; - } - var wsSrv = new WebSocketServer(wsConfig); - Storage.create(config, function (store) { - NetfluxSrv.run(store, wsSrv, config, rpc); - }); - } -}; +var rpc; -var loadRPC = function (cb) { - config.rpc = typeof(config.rpc) === 'undefined'? './rpc.js' : config.rpc; +var nt = nThen(function (w) { + if (!config.enableTaskScheduling) { return; } + var Tasks = require("./storage/tasks"); - if (typeof(config.rpc) === 'string') { - // load pin store... - var Rpc = require(config.rpc); - Rpc.create(config, function (e, rpc) { - if (e) { throw e; } - cb(void 0, rpc); - }); - } else { - cb(); + console.log("loading task scheduler"); + Tasks.create(config, w(function (e, tasks) { + config.tasks = tasks; + })); +}).nThen(function (w) { + config.rpc = typeof(config.rpc) === 'undefined'? './rpc.js' : config.rpc; + if (typeof(config.rpc) !== 'string') { return; } + // load pin store... + var Rpc = require(config.rpc); + Rpc.create(config, w(function (e, _rpc) { + if (e) { + w.abort(); + throw e; + } + rpc = _rpc; + })); +}).nThen(function () { + if(config.useExternalWebsocket) { return; } + if (websocketPort !== config.httpPort) { + console.log("setting up a new websocket server"); + wsConfig = { port: websocketPort}; } -}; + var wsSrv = new WebSocketServer(wsConfig); + Storage.create(config, function (store) { + NetfluxSrv.run(store, wsSrv, config, rpc); + }); +}); -loadRPC(createSocketServer); diff --git a/storage/tasks.js b/storage/tasks.js new file mode 100644 index 000000000..b25a3a637 --- /dev/null +++ b/storage/tasks.js @@ -0,0 +1,94 @@ +var Fs = require("fs"); +var Path = require("path"); +var nacl = require("tweetnacl"); +var nThen = require("nthen"); + +var Tasks = module.exports; + +var encode = function (time, command, args) { + if (typeof(time) !== 'number') { return null; } + if (typeof(command) !== 'string') { return null; } + if (!Array.isArray(args)) { return [time, command]; } + return [time, command].concat(args); +}; + +var randomId = function () { + var bytes = Array.prototype.slice.call(nacl.randomBytes(16)); + return bytes.map(function (b) { + var n = Number(b & 0xff).toString(16); + return n.length === 1? '0' + n: n; + }).join(''); +}; + +var mkPath = function (env, id) { + return Path.join(env.root, id.slice(0, 2), id) + '.ndjson'; +}; + +var getFreeId = function (env, cb, tries) { + if (tries > 5) { return void cb('ETOOMANYTRIES'); } + + // generate a unique id + var id = randomId(); + + // derive a path from that id + var path = mkPath(env, id); + + Fs.stat(path, function (err) { + if (err && err.code === "ENOENT") { + cb(void 0, id); + } else { + getFreeId(env, cb); + } + }); +}; + +var write = function (env, task, cb) { + var str = JSON.stringify(task) + '\n'; + var id = nacl.util.encodeBase64(nacl.hash(nacl.util.decodeUTF8(str))).replace(/\//g, '-'); + + var path = mkPath(env, id); + nThen(function (w) { + // check if the file already exists... + Fs.stat(path, w(function (err) { + if (err && err.code === 'ENOENT') { return; } + w.abort(); cb(); + })); + }).nThen(function (w) { + // create the parent directory if it does not exist + var dir = id.slice(0, 2); + var dirpath = Path.join(env.root, dir); + + Fs.mkdir(dirpath, 0x1ff, w(function (err) { + if (err && err.code !== 'EEXIST') { + return void cb(err); + } + })); + }).nThen(function (w) { + // write the file to the path + Fs.writeFile(mkPath(env, id), str, function (e) { + if (e) { return void cb(e); } + cb(); + }); + }); +}; + +Tasks.create = function (config, cb) { + var env = { + root: config.taskPath || './tasks', + }; + + // make sure the path exists... + Fs.mkdir(env.root, 0x1ff, function (err) { + if (err && err.code !== 'EEXIST') { + throw err; + } + cb(void 0, { + write: function (time, command, args, cb) { + var task = encode(time, command, args); + write(env, task, cb); + }, + }); + }); +}; + +