prepare to merge history keeper and rpc

pull/1/head
ansuz 5 years ago
parent b093d3f0d2
commit 80c012f34d

@ -0,0 +1,60 @@
/* jshint esversion: 6 */
const nThen = require("nthen");
const WebSocketServer = require('ws').Server;
const NetfluxSrv = require('chainpad-server/NetfluxWebsocketSrv');
module.exports.create = function (config) {
var historyKeeper;
var rpc;
const log = config.log;
const wsConfig = {
server: config.httpServer,
};
nThen(function (w) {
require('../storage/file').create(config, w(function (_store) {
config.store = _store;
}));
}).nThen(function (w) {
require("../storage/tasks").create(config, w(function (e, tasks) {
if (e) {
throw e;
}
config.tasks = tasks;
if (config.disableIntegratedTasks) { return; }
// XXX support stopping this interval
setInterval(function () {
tasks.runAll(function (err) {
if (err) {
// either TASK_CONCURRENCY or an error with tasks.list
// in either case it is already logged.
}
});
}, 1000 * 60 * 5); // run every five minutes
}));
}).nThen(function (w) {
require("./rpc").create(config, w(function (e, _rpc) {
if (e) {
w.abort();
throw e;
}
rpc = _rpc;
}));
}).nThen(function () {
var HK = require('./historyKeeper.js');
var hkConfig = {
tasks: config.tasks,
rpc: rpc,
store: config.store,
log: log,
};
// XXX historyKeeper exports a `setConfig` method
historyKeeper = HK.create(hkConfig);
}).nThen(function () {
var wsSrv = new WebSocketServer(wsConfig);
// XXX NetfluxSrv shares some internal functions with historyKeeper
// by passing them to setConfig
NetfluxSrv.run(wsSrv, config, historyKeeper);
});
};

@ -1,6 +1,5 @@
/* jshint esversion: 6 */ /* jshint esversion: 6 */
/* global Buffer */ /* global Buffer */
;(function () { 'use strict';
const nThen = require('nthen'); const nThen = require('nthen');
const Nacl = require('tweetnacl/nacl-fast'); const Nacl = require('tweetnacl/nacl-fast');
@ -63,6 +62,8 @@ const isValidValidateKeyString = function (key) {
} }
}; };
var CHECKPOINT_PATTERN = /^cp\|(([A-Za-z0-9+\/=]+)\|)?/;
module.exports.create = function (cfg) { module.exports.create = function (cfg) {
const rpc = cfg.rpc; const rpc = cfg.rpc;
const tasks = cfg.tasks; const tasks = cfg.tasks;
@ -385,8 +386,6 @@ module.exports.create = function (cfg) {
return true; return true;
}; };
var CHECKPOINT_PATTERN = /^cp\|(([A-Za-z0-9+\/=]+)\|)?/;
/* onChannelMessage /* onChannelMessage
Determine what we should store when a message a broadcasted to a channel" Determine what we should store when a message a broadcasted to a channel"
@ -992,5 +991,3 @@ module.exports.create = function (cfg) {
onDirectMessage: onDirectMessage, onDirectMessage: onDirectMessage,
}; };
}; };
}());

@ -4,17 +4,12 @@
var Express = require('express'); var Express = require('express');
var Http = require('http'); var Http = require('http');
var Fs = require('fs'); var Fs = require('fs');
var WebSocketServer = require('ws').Server;
var NetfluxSrv = require('chainpad-server/NetfluxWebsocketSrv');
var Package = require('./package.json'); var Package = require('./package.json');
var Path = require("path"); var Path = require("path");
var nThen = require("nthen"); var nThen = require("nthen");
var config = require("./lib/load-config"); var config = require("./lib/load-config");
// support multiple storage back ends
var Storage = require('./storage/file');
var app = Express(); var app = Express();
// mode can be FRESH (default), DEV, or PACKAGE // mode can be FRESH (default), DEV, or PACKAGE
@ -69,11 +64,9 @@ var setHeaders = (function () {
if (Object.keys(headers).length) { if (Object.keys(headers).length) {
return function (req, res) { return function (req, res) {
const h = [ const h = [
/^\/pad(2)?\/inner\.html.*/, /^\/pad\/inner\.html.*/,
/^\/common\/onlyoffice\/.*\/index\.html.*/, /^\/common\/onlyoffice\/.*\/index\.html.*/,
/^\/sheet\/inner\.html.*/, /^\/(sheet|ooslide|oodoc)\/inner\.html.*/,
/^\/ooslide\/inner\.html.*/,
/^\/oodoc\/inner\.html.*/,
].some((regex) => { ].some((regex) => {
return regex.test(req.url) return regex.test(req.url)
}) ? padHeaders : headers; }) ? padHeaders : headers;
@ -117,11 +110,6 @@ app.use(function (req, res, next) {
app.use(Express.static(__dirname + '/www')); app.use(Express.static(__dirname + '/www'));
Fs.exists(__dirname + "/customize", function (e) {
if (e) { return; }
console.log("Cryptpad is customizable, see customize.dist/readme.md for details");
});
// FIXME I think this is a regression caused by a recent PR // FIXME I think this is a regression caused by a recent PR
// correct this hack without breaking the contributor's intended behaviour. // correct this hack without breaking the contributor's intended behaviour.
@ -207,80 +195,36 @@ app.use(function (req, res, next) {
var httpServer = Http.createServer(app); var httpServer = Http.createServer(app);
httpServer.listen(config.httpPort,config.httpAddress,function(){ nThen(function (w) {
var host = config.httpAddress; Fs.exists(__dirname + "/customize", w(function (e) {
var hostName = !host.indexOf(':') ? '[' + host + ']' : host; if (e) { return; }
console.log("Cryptpad is customizable, see customize.dist/readme.md for details");
var port = config.httpPort; }));
var ps = port === 80? '': ':' + port; }).nThen(function (w) {
httpServer.listen(config.httpPort,config.httpAddress,function(){
console.log('[%s] server available http://%s%s', new Date().toISOString(), hostName, ps); var host = config.httpAddress;
}); var hostName = !host.indexOf(':') ? '[' + host + ']' : host;
if (config.httpSafePort) {
Http.createServer(app).listen(config.httpSafePort, config.httpAddress);
}
var wsConfig = { server: httpServer };
var rpc; var port = config.httpPort;
var historyKeeper; var ps = port === 80? '': ':' + port;
var log; console.log('[%s] server available http://%s%s', new Date().toISOString(), hostName, ps);
});
// Initialize logging, the the store, then tasks, then rpc, then history keeper and then start the server if (config.httpSafePort) {
var nt = nThen(function (w) { Http.createServer(app).listen(config.httpSafePort, config.httpAddress, w());
// set up logger
var Logger = require("./lib/log");
//console.log("Loading logging module");
Logger.create(config, w(function (_log) {
log = config.log = _log;
}));
}).nThen(function (w) {
if (config.externalWebsocketURL) {
// if you plan to use an external websocket server
// then you don't need to load any API services other than the logger.
// Just abort.
w.abort();
return;
} }
Storage.create(config, w(function (_store) {
config.store = _store;
}));
}).nThen(function (w) {
var Tasks = require("./storage/tasks");
Tasks.create(config, w(function (e, tasks) {
if (e) {
throw e;
}
config.tasks = tasks;
if (config.disableIntegratedTasks) { return; }
setInterval(function () {
tasks.runAll(function (err) {
if (err) {
// either TASK_CONCURRENCY or an error with tasks.list
// in either case it is already logged.
}
});
}, 1000 * 60 * 5); // run every five minutes
}));
}).nThen(function (w) {
require("./lib/rpc").create(config, w(function (e, _rpc) {
if (e) {
w.abort();
throw e;
}
rpc = _rpc;
}));
}).nThen(function () { }).nThen(function () {
var HK = require('./lib/historyKeeper.js'); var wsConfig = { server: httpServer };
var hkConfig = {
tasks: config.tasks, // Initialize logging then start the API server
rpc: rpc, require("./lib/log").create(config, function (_log) {
store: config.store, config.log = _log;
log: log, config.httpServer = httpServer;
};
historyKeeper = HK.create(hkConfig); if (config.externalWebsocketURL) { return; }
}).nThen(function () { require("./lib/api").create(config);
var wsSrv = new WebSocketServer(wsConfig); });
NetfluxSrv.run(wsSrv, config, historyKeeper);
}); });

Loading…
Cancel
Save