coolaj86
/
sqlite3-cluster.js
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Import subtree 'sqlite3-cluster'

This commit is contained in:
AJ ONeal 2015-07-24 14:04:22 -06:00
parent 83f2d1cae1 98a9f76d30
commit 1b63208266
11 changed files with 535 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
README.md Normal file
View File

@ -0,0 +1,35 @@
DRAFT
=====
This is just hypothetical while I build out the API
SQLite3 Server
=============
Node.js runs on a single core, which isn't very effective.
You can run multiple Node.js instances to take advantage of multiple cores,
but if you do that, you can't use SQLite in each process.
This module will either run client-server style in environments that benefit from it
(such as the Raspberry Pi 2 with 4 cores), or in-process for environments that don't
(such as the Raspberry Pi B and B+).
Usage
=====
```js
var sqlite = require('sqlite3-server');
var opts = {
key: '1892d335081d8d346e556c9c3c8ff2c3'
, bits: 128
, filename: path.join('/tmp/authn.sqlcipher')
, verbose: false
, port: 3232 // default random
, forceServer: true // default false
};
sqlite.create(opts).then(function (db) {
// EXACT same api as db
});
```

204
client.js Normal file
View File

@ -0,0 +1,204 @@
'use strict';
/*global Promise*/
// TODO iterate over the prototype
// translate request / response
var sqlite3real = require('sqlite3');
/*
function createConnection(opts) {
var server = ;
return server.create(opts).then(function () {
// created and listening
});
}
*/
function startServer(opts) {
return require('./server').create(opts).then(function (server) {
// this process doesn't need to connect to itself
// through a socket
return server.masterClient;
});
}
function getConnection(opts) {
return new Promise(function (resolve) {
setTimeout(function () {
var WebSocket = require('ws');
var ws = new WebSocket('ws+unix:' + opts.sock);
if (opts.serve) {
console.log("[EXPLICIT SERVER] #################################################");
return startServer(opts).then(function (client) {
// ws.masterClient = client;
resolve({ masterClient: client });
});
}
ws.on('error', function (err) {
console.error('[ERROR] ws connection failed, retrying');
console.error(err);
function retry() {
setTimeout(function () {
getConnection(opts).then(resolve);
}, 100 + (Math.random() * 250));
}
if (!opts.connect && ('ENOENT' === err.code || 'ECONNREFUSED' === err.code)) {
console.log('[NO SERVER] attempting to create a server #######################');
return startServer(opts).then(function (client) {
// ws.masterClient = client;
resolve({ masterClient: client });
}, function () {
retry();
});
}
retry();
});
ws.on('open', function () {
resolve(ws);
});
});
}, 100 + (Math.random() * 250));
}
function create(opts) {
if (!opts.sock) {
opts.sock = opts.filename + '.sock';
}
var promise;
var numcpus = require('os').cpus().length;
if (opts.standalone || (1 === numcpus && !opts.serve && !opts.connect)) {
return require('./wrapper').create(opts);
}
function retryServe() {
return startServer(opts).then(function (client) {
// ws.masterClient = client;
return { masterClient: client };
}, function () {
retryServe();
});
}
if (opts.serve) {
console.log('[EXPLICIT]');
promise = retryServe();
} else {
promise = getConnection(opts);
}
/*
if (opts.connect) {
}
*/
// TODO maybe use HTTP POST instead?
return promise.then(function (ws) {
if (ws.masterClient) {
console.log('[MASTER CLIENT] found');
return ws.masterClient;
}
var db = {};
var proto = sqlite3real.Database.prototype;
var messages = [];
function rpc(fname, args) {
var id;
var cb;
if ('function' === typeof args[args.length - 1]) {
id = Math.random();
cb = args.pop();
}
console.log('fname, args');
console.log(fname, args);
ws.send(JSON.stringify({
type: 'rpc'
, func: fname
, args: args
, filename: opts.filename
, id: id
}));
if (!cb) {
return;
}
function onMessage(data) {
var cmd;
try {
cmd = JSON.parse(data.toString('utf8'));
} catch(e) {
console.error('[ERROR] in client, from sql server parse json');
console.error(e);
console.error(data);
console.error();
//ws.send(JSON.stringify({ type: 'error', value: { message: e.message, code: "E_PARSE_JSON" } }));
return;
}
if (cmd.id !== id) {
return;
}
//console.log('onMessage data');
//console.log(cmd);
cb.apply(cmd.this, cmd.args);
if ('on' !== fname) {
var index = messages.indexOf(onMessage);
messages.splice(index, 1);
}
}
messages.push(onMessage);
}
db.sanitize = require('./wrapper').sanitize;
Object.keys(sqlite3real.Database.prototype).forEach(function (key) {
if ('function' === typeof proto[key]) {
db[key] = function () {
rpc(key, Array.prototype.slice.call(arguments));
};
}
});
ws.on('message', function (data) {
messages.forEach(function (fn) {
try {
fn(data);
} catch(e) {
// ignore
}
});
});
// serialize
// parallel
db.serialize = db.parallel = function () {
throw new Error('NOT IMPLEMENTED in SQLITE3-remote');
};
return db;
});
}
module.exports.sanitize = require('./wrapper').sanitize;
module.exports.create = create;

21
cluster.js Normal file
View File

@ -0,0 +1,21 @@
'use strict';
var sqlite3 = require('./index');
function create(opts) {
var cluster = require('cluster');
var numCores = require('os').cpus().length;
if (!opts.serve && ('boolean' !== typeof opts.serve)) {
opts.serve = (numCores > 1) && cluster.isMaster;
}
if (!opts.connect && ('boolean' !== typeof opts.connect)) {
opts.connect = (numCores > 1) && cluster.isWorker;
}
return sqlite3.create(opts);
}
module.exports.sanitize = sqlite3.sanitize;
module.exports.create = create;

3
index.js Normal file
View File

@ -0,0 +1,3 @@
'use strict';
module.exports = require('./client');

11
install-sqlcipher.bash Normal file
View File

@ -0,0 +1,11 @@
#brew options sqlcipher
#brew install sqlcipher --with-fts
echo STOP
echo You must manually install sqlcipher
exit 1
export LDFLAGS="-L`brew --prefix`/opt/sqlcipher/lib"
export CPPFLAGS="-I`brew --prefix`/opt/sqlcipher/include"
npm install sqlite3 --build-from-source --sqlite_libname=sqlcipher --sqlite=`brew --prefix`
node -e 'require("sqlite3")'

15
serve.js Normal file
View File

@ -0,0 +1,15 @@
'use strict';
var server = require('http').createServer();
var WebSocketServer = require('ws').Server;
var wss = new WebSocketServer({ server: server });
var port = process.env.PORT || process.argv[0] || 4080;
var app = require('./sqlite-server');
server.listen(port, function () {
});
app.create({ server: server, wss: wss }).then(function (app) {
server.on('request', app);
});

102
server.js Normal file
View File

@ -0,0 +1,102 @@
'use strict';
/*global Promise*/
var wsses = {};
function createApp(server, options) {
console.log('Create App');
if (wsses[options.filename]) {
return Promise.resolve(wsses[options.filename]);
}
return require('./wrapper').create(options).then(function (db) {
var url = require('url');
var express = require('express');
var app = express();
var wss = server.wss;
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
ws.__session_id = location.query.session_id || Math.random();
ws.on('message', function (buffer) {
console.log('[SERVER MESSAGE]', buffer);
var cmd;
try {
cmd = JSON.parse(buffer.toString('utf8'));
} catch(e) {
console.error('[ERROR] parse json');
console.error(e);
console.error(buffer);
console.error();
ws.send(JSON.stringify({ type: 'error', value: { message: e.message, code: "E_PARSE_JSON" } }));
return;
}
switch(cmd.type) {
case 'init':
break;
case 'rpc':
cmd.args.push(function () {
var args = Array.prototype.slice.call(arguments);
ws.send(JSON.stringify({
this: this
, args: args
, id: cmd.id
}));
});
db[cmd.func].apply(db, cmd.args);
break;
default:
break;
}
});
ws.send(JSON.stringify({ type: 'session', value: ws.__session_id }));
});
app.masterClient = db;
wsses[options.filename] = app;
return app;
});
}
function create(options) {
console.log('Create Server');
return new Promise(function (resolve) {
var server = require('http').createServer();
var WebSocketServer = require('ws').Server;
var wss = new WebSocketServer({ server: server });
//var port = process.env.PORT || process.argv[0] || 4080;
console.log('options.sock');
console.log(options.sock);
var fs = require('fs');
fs.unlink(options.sock, function () {
// ignore error when socket doesn't exist
server.listen(options.sock, function () {
console.log('Listening');
});
});
createApp({ server: server, wss: wss }, options).then(function (app) {
server.on('request', app);
resolve({ masterClient: app.masterClient });
});
});
}
module.exports.create = create;

16
standalone.js Normal file
View File

@ -0,0 +1,16 @@
'use strict';
var sqlite3 = require('./index');
function create(opts) {
opts.standalone = true;
// TODO if cluster *is* used issue a warning?
// I suppose the user could be issuing a different filename for each
// ... but then they have no need to use this module, right?
return sqlite3.create(opts);
}
module.exports.sanitize = sqlite3.sanitize;
module.exports.create = create;

38
test-cluster.js Normal file
View File

@ -0,0 +1,38 @@
'use strict';
var cluster = require('cluster');
var numCores = require('os').cpus().length;
var i;
function run() {
var sqlite3 = require('./cluster');
sqlite3.create({
key: '00000000000000000000000000000000'
, bits: 128
, filename: '/tmp/test.cluster.sqlcipher'
, verbose: null
, standalone: null
, serve: null
, connect: null
}).then(function (client) {
client.run("SELECT 1", [], function (err) {
if (err) {
console.error('[ERROR]', cluster.isMaster && '0' || cluster.worker.id);
console.error(err);
return;
}
console.log('[this]', cluster.isMaster && '0' || cluster.worker.id);
console.log(this);
});
});
}
if (cluster.isMaster) {
for (i = 1; i <= numCores; i += 1) {
cluster.fork();
}
}
run();

28
test-standalone.js Normal file
View File

@ -0,0 +1,28 @@
'use strict';
function run() {
var sqlite3 = require('./standalone');
sqlite3.create({
key: '00000000000000000000000000000000'
, bits: 128
, filename: '/tmp/test.cluster.sqlcipher'
, verbose: null
, standalone: true
, serve: null
, connect: null
}).then(function (client) {
client.run("SELECT 1", [], function (err) {
if (err) {
console.error('[ERROR] standalone');
console.error(err);
return;
}
console.log('[this] standalone');
console.log(this);
});
});
}
run();

62
wrapper.js Normal file
View File

@ -0,0 +1,62 @@
'use strict';
/*global Promise*/
var sqlite3 = require('sqlite3');
var dbs = {};
function sanitize(str) {
return String(str).replace("'", "''");
}
function create(opts) {
var db;
if (!opts) {
opts = {};
}
if (opts.verbose) {
sqlite3.verbose();
}
if (!dbs[opts.filename] || dbs[opts.filename].__key !== opts.key) {
dbs[opts.filename] = new sqlite3.Database(opts.filename);
}
db = dbs[opts.filename];
db.sanitize = sanitize;
db.__key = opts.key;
return new Promise(function (resolve, reject) {
db.serialize(function() {
var setup = [];
if (opts.key) {
// TODO test key length
if (!opts.bits) {
opts.bits = 128;
}
// TODO db.run(sql, function () { resolve() });
setup.push(new Promise(function (resolve, reject) {
db.run("PRAGMA KEY = \"x'" + sanitize(opts.key) + "'\"", [], function (err) {
if (err) { reject(err); return; }
resolve(this);
});
}));
setup.push(new Promise(function (resolve, reject) {
db.run("PRAGMA CIPHER = 'aes-" + sanitize(opts.bits) + "-cbc'", [], function (err) {
if (err) { reject(err); return; }
resolve(this);
});
}));
}
Promise.all(setup).then(function () { resolve(db); }, reject);
});
});
}
module.exports.sanitize = sanitize;
module.exports.Database = sqlite3.Database;
module.exports.create = create;