sqlite3-cluster.js/server.js

224 lines
6.0 KiB
JavaScript
Raw Normal View History

2015-07-22 00:58:34 +00:00
'use strict';
var PromiseA = require('bluebird').Promise;
2015-07-23 02:58:18 +00:00
var wsses = {};
function createApp(servers, options) {
var url = require('url');
var wss = servers.wss;
2015-11-10 12:16:50 +00:00
//var server = servers.server;
2015-07-24 22:11:10 +00:00
2015-11-10 12:16:50 +00:00
//var express = require('express');
//var app = express();
/*
function app(req, res) {
res.end('NOT IMPLEMENTED');
}
2015-11-10 12:16:50 +00:00
*/
2015-07-22 00:58:34 +00:00
wss.on('connection', function (ws) {
var location = url.parse(ws.upgradeReq.url, true);
// you might use location.query.access_token to authenticate or share sessions
// or ws.upgradeReq.headers.cookie (see http://stackoverflow.com/a/16395220/151312
if (!options.ipcKey) {
2015-11-10 12:16:50 +00:00
console.warn("[S] [SECURITY] please include { ipcKey: crypto.randomBytes(16).toString('base64') }"
+ " in your options and pass it from master to worker processes with worker.send()");
ws._authorized = true;
} else {
ws._authorized = (options.ipcKey === (location.query.ipcKey || location.query.ipc_key));
}
2015-07-22 00:58:34 +00:00
if (!ws._authorized) {
ws.send(JSON.stringify({ error: { message: "Unauthorized: ipc_key does not match", code: 'E_UNAUTHORIZED_IPCKEY' } }));
ws.close();
return;
}
2015-07-22 00:58:34 +00:00
2015-11-10 12:16:50 +00:00
if (!wss.__count) {
wss.__count = 0;
}
wss.__count += 1;
function decrWs() {
wss.__count -= 1;
if (!wss.__count) {
2015-11-10 12:16:50 +00:00
console.log('[S] client count is zero, but server will be left open');
/*
wss.close(function () {
console.log('wss.closed');
});
server.close(function () {
console.log('server closed, but will not exit due to bug? in wss.close()');
process.exit(0);
});
*/
}
2015-11-10 12:16:50 +00:00
}
ws.on('error', function (err) {
console.error('[S] [WebSocket error]');
console.error(err.stack);
decrWs();
});
2015-11-10 12:16:50 +00:00
ws.on('close', decrWs);
ws.on('message', function (buffer) {
var cmd;
var promise;
try {
cmd = JSON.parse(buffer.toString('utf8'));
} catch(e) {
2015-11-10 12:16:50 +00:00
console.error('[S] [ERROR] parse json');
console.error(e.stack || e);
console.error(buffer);
console.error();
ws.send(JSON.stringify({ type: 'error', value: { message: e.message, code: "E_PARSE_JSON" } }));
return;
}
2015-11-10 12:16:50 +00:00
//console.log('cmd');
//console.log(cmd);
// caching and create logic happens in the wrapper stored here below
2015-11-10 12:16:50 +00:00
promise = require('./wrapper').create(cmd && cmd.dbname && cmd || options).then(function (db) {
2015-07-22 00:58:34 +00:00
2015-07-23 02:58:18 +00:00
switch(cmd.type) {
case 'init':
2015-11-17 08:33:54 +00:00
//console.log('[S] init', cmd);
db[cmd.func].apply(db, cmd.args).then(function () {
var args = Array.prototype.slice.call(arguments);
var myself;
if (args[0] === db) {
args = [];
myself = true;
}
//console.log('[INIT HAPPENING]');
ws.send(JSON.stringify({
id: cmd.id
, self: myself
, args: args
//, this: this
}));
});
2015-07-23 02:58:18 +00:00
break;
2015-07-22 00:58:34 +00:00
2015-07-23 02:58:18 +00:00
case 'rpc':
2015-11-10 12:16:50 +00:00
if ('close' !== cmd.func && !db._initialized) {
//console.log('[RPC NOT HAPPENING]');
ws.send(JSON.stringify({
type: 'error'
, id: cmd.id
2015-11-10 12:16:50 +00:00
, args: [{ message: 'database has not been initialized', code: 'E_NO_INIT' }]
, error: { message: 'database has not been initialized', code: 'E_NO_INIT' }
}));
return;
}
cmd.args.push(function (err) {
2015-07-24 08:57:17 +00:00
var args = Array.prototype.slice.call(arguments);
var myself;
if (args[0] === db) {
args = [];
myself = true;
}
2015-07-24 08:57:17 +00:00
//console.log('[RPC HAPPENING]', args, cmd.id);
2015-07-24 08:57:17 +00:00
ws.send(JSON.stringify({
this: (!err && this !== global) ? this : {}
2015-07-24 08:57:17 +00:00
, args: args
, self: myself
2015-07-24 08:57:17 +00:00
, id: cmd.id
, error: err
2015-07-24 08:57:17 +00:00
}));
});
db[cmd.func].apply(db, cmd.args);
2015-07-23 02:58:18 +00:00
break;
2015-07-22 00:58:34 +00:00
2015-07-23 02:58:18 +00:00
default:
2015-07-24 22:11:10 +00:00
throw new Error('UNKNOWN TYPE');
//break;
2015-07-23 02:58:18 +00:00
}
2015-07-22 00:58:34 +00:00
2015-07-23 02:58:18 +00:00
});
2015-07-22 00:58:34 +00:00
});
});
2015-07-22 00:58:34 +00:00
2015-11-10 12:16:50 +00:00
// wsses[options.sock] = app;
return PromiseA.resolve();
2015-07-23 02:58:18 +00:00
}
2015-07-22 00:58:34 +00:00
2015-11-10 12:16:50 +00:00
function newSocket(options) {
if (wsses[options.sock]) {
return PromiseA.resolve(wsses[options.sock]);
}
2015-07-23 02:58:18 +00:00
2015-11-10 12:16:50 +00:00
wsses[options.sock] = new PromiseA(function (resolve) {
var fs = require('fs');
2015-07-23 02:58:18 +00:00
fs.unlink(options.sock, function () {
2015-11-10 12:16:50 +00:00
var server = require('http').createServer();
2015-07-23 02:58:18 +00:00
// ignore error when socket doesn't exist
2015-11-10 12:16:50 +00:00
server.listen(options.sock, function () {
resolve(server);
});
});
}).then(function (server) {
var WebSocketServer = require('ws').Server;
var servers = {
server: require('http').createServer()
, wss: new WebSocketServer({ server: server })
};
return createApp(servers, options).then(function (/*app*/) {
// server.on('request', app);
wsses[options.sock] = servers;
return wsses[options.sock];
});
});
return wsses[options.sock];
}
function createMasterClient(options) {
return require('./wrapper').create(options, null).then(function (db) {
return db;
});
}
function createServer(options) {
if (!options.sock) {
throw new Error("Please provide options.sock as the socket to serve from");
}
options.server = options.sock;
return newSocket(options).then(function () {
var result = {};
2015-07-23 02:58:18 +00:00
2015-11-10 12:16:50 +00:00
Object.keys(wsses[options.sock]).forEach(function (key) {
result[key] = wsses[options.sock][key];
2015-07-23 02:58:18 +00:00
});
2015-11-10 12:16:50 +00:00
});
}
2015-07-23 02:58:18 +00:00
2015-11-10 12:16:50 +00:00
function create(options) {
return createServer(options).then(function (result) {
if (!options.dbname) {
return result;
}
2015-07-24 22:11:10 +00:00
2015-11-10 12:16:50 +00:00
return createMasterClient(options).then(function (db) {
result.masterClient = db;
return result;
});
2015-07-22 00:58:34 +00:00
});
}
module.exports.create = create;
2015-11-10 12:16:50 +00:00
module.exports.createMasterClient = createMasterClient;
module.exports.createServer = createServer;