public
Last active — forked from felixge/mysql.js

  • Download Gist
mysql.js
JavaScript
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146
var sys = require('sys');
var path = require('path');
 
var Db = exports.Db = function(options) {
this.options = {
host: 'localhost',
user: '',
pass: '',
database: '',
workers: 2
};
 
this.jobs = [];
this.workers = [];
 
process.mixin(this.options, options);
 
this.connect();
};
 
Db.prototype.connect = function() {
for (var i = 0; i < this.options.workers; i++) {
var options = {
host: this.options.host,
user: this.options.user,
pass: this.options.pass,
database: this.options.database
}
this.workers.push(new PhpWorker(options));
}
};
 
Db.prototype.work = function() {
for (var i = 0; i < this.workers.length; i++) {
var worker = this.workers[i];
if (worker.busy()) {
continue;
}
 
var job = this.jobs.shift(), self = this;
if (!job) {
break;
}
 
worker.send(job.params)
.addCallback(function(result) {
job.promise.emitSuccess(result);
setTimeout(function() {
self.work();
}, 1);
})
.addErrback(function(error) {
job.promise.emitError(error);
setTimeout(function() {
self.work();
}, 1)
});
break;
}
};
 
Db.prototype.query = function(query) {
var promise = new process.Promise();
 
var args = Array.prototype.slice.call(arguments);
 
this.jobs.push({
params: ['query'].concat(args),
promise: promise
});
this.work();
 
return promise;
};
 
var PhpWorker = function(options) {
this.options = {
host: 'localhost',
user: '',
pass: '',
database: '',
};
process.mixin(this.options, options);
 
this.job = null;
this.launch();
};
 
PhpWorker.prototype.relaunch = function() {
this.process.kill();
this.launch();
};
 
PhpWorker.prototype.launch = function() {
var cmd = [
'php',
path.join(path.dirname(__filename), 'worker.php'),
"'"+JSON.stringify(this.options)+"'"
].join(' ');
this.process = process.createChildProcess('/bin/sh', ['-c', cmd]);
 
var buffer = '', self = this;
this.process.addListener('output', function(chunk) {
buffer = buffer + (chunk || '');
while (buffer) {
var offset = buffer.indexOf("\n");
if (offset < 0) {
return;
}
 
var line = buffer.substr(0, offset);
buffer = buffer.substr(offset + 1);
 
self.onLine(line);
}
});
};
 
PhpWorker.prototype.onLine = function(line) {
try {
line = JSON.parse(line);
} catch (e) {
return puts('Could not parse line: '+line);
}
 
if (this.job) {
this.job.emitSuccess(line);
this.job = null;
} else {
puts('unexpected PhpWorker.onLine event, received: '+line);
}
};
 
PhpWorker.prototype.send = function(job) {
this.job = new process.Promise();
 
job = JSON.stringify(job);
puts('> '+job);
this.process.write(job+"\n", 'utf8');
 
return this.job;
};
 
PhpWorker.prototype.busy = function() {
return !!this.job;
};
test.js
JavaScript
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
process.mixin(require('sys'));
var mysql = require('../lib/mysql');
 
var db = new mysql.Db({
user: 'root',
pass: 'root',
database: 'mydb'
});
 
var query = db.query('SELECT %s FROM posts LIMIT 10', '*');
query.addCallback(function(results) {
p(results);
});
 
query.addErrback(function() {
p('error');
});
worker.php
PHP
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63
<?php
$options = json_decode(stripslashes($argv[1]), true);
 
function connect($options) {
static $retry = 0;
$db = @mysql_pconnect($options['host'], $options['user'], $options['pass']);
 
if (!is_resource($db)) {
return false;
}
 
if (!@mysql_select_db($options['database'], $db)) {
return false;
}
return $db;
}
 
$db = connect($options);
 
function send($line) {
$line = json_encode($line);
fwrite(STDOUT, $line."\n");
flush();
}
 
while($line = rtrim(fgets(STDIN))) {
$args = json_decode($line, true);
$cmd = array_shift($args);
 
while (true) {
$result = (is_resource($db))
? mysql_query('SELECT 1', $db)
: false;
 
if (is_resource($result)) {
break;
}
$db = connect($options);
sleep(1);
}
 
switch ($cmd) {
case 'query':
$template = array_shift($args);
foreach ($args as $i => $arg) {
$args[$i] = "'".mysql_real_escape_string($arg, $db)."'";
}
$query = vsprintf($template, $args);
$result = mysql_query($query, $db);
 
$rows = array();
while ($row = mysql_fetch_assoc($result)) {
$rows[] = $row;
}
mysql_free_result($result);
send($rows);
break;
default:
send(array('error' => 'unknown command: '.$cmd));
break;
}
}
?>

Please sign in to comment on this gist.

Something went wrong with that request. Please try again.