Browse files

Support for multiple reply parsers including hiredis.

Several parsing bugs fixed in JavaScript.
Some new config options that need to be better documented.
  • Loading branch information...
1 parent 232f34a commit b9073645735cb3415f8056a42951ad22655832b1 @mranney mranney committed Dec 6, 2010
Showing with 170 additions and 150 deletions.
  1. +15 −20 README.md
  2. +5 −0 changelog.md
  3. +19 −5 examples/unix_socket.js
  4. +39 −66 index.js
  5. +15 −12 lib/parser/hiredis.js
  6. +32 −23 lib/parser/javascript.js
  7. +1 −2 lib/util.js
  8. +23 −15 multi_bench.js
  9. +3 −2 package.json
  10. +18 −5 test.js
View
35 README.md
@@ -1,22 +1,26 @@
redis - a node.js redis client
===========================
-This is a complete Redis client for node.js. It is designed for node 0.2.2+ and redis 2.0.1+.
-It might not work on earlier versions of either, although it probably will.
-
-This client supports all Redis commands, including MULTI and PUBLISH/SUBSCRIBE.
+This is a complete Redis client for node.js. It supports all Redis commands, including MULTI, WATCH, and PUBLISH/SUBSCRIBE.
Install with:
npm install redis
+
+By default, a pure JavaScript reply parser is used. This is clever and portable, but not as fast for large responses as `hiredis` from the
+Redis distribution. To use the `hiredis`, do:
+
+ npm install hiredis
+
+If `hiredis` is installed, `node_redis` will use it by default.
## Why?
-`node_redis` works in the latest versions of node, is published in `npm`, and is very fast, particularly for small responses.
+`node_redis` works in the latest versions of node, is published in `npm`, is used by many people, and is in production on a
+number of sites.
-`node_redis` is designed with performance in mind. The included `bench.js` runs similar tests to `redis-benchmark`, included with the Redis
-distribution, and `bench.js` is as fast as `redis-benchmark` for some patterns and slower for others. `node_redis` has many lovingly
-hand-crafted optimizations for speed.
+`node_redis` was originally written to replace `node-redis-client` which hasn't been updated in a while, and no longer works
+on recent versions of node.
## Usage
@@ -27,7 +31,7 @@ Simple example, included as `example.js`:
client = redis.createClient();
client.on("error", function (err) {
- console.log("Redis connection error to " + client.host + ":" + client.port + " - " + err);
+ console.log("Error " + err);
});
client.set("string key", "string val", redis.print);
@@ -392,22 +396,12 @@ Defaults to 1.7. The default initial connection retry is 250, so the second ret
## TODO
-Many common uses of Redis are fine with JavaScript Strings, and Strings are faster than Buffers. We should get a way to
-use Strings if binary-safety isn't a concern. Also, dealing with Buffer results is kind of annoying.
-
-Stream large set/get into and out of Redis.
+Stream large set/get values into and out of Redis. Otherwise the entire value must be in node's memory.
Performance can be better for very large values.
I think there are more performance improvements left in there for smaller values, especially for large lists of small values.
-## Also
-
-This library might still have some bugs in it, but it seems to be quite useful for a lot of people at this point.
-There are other Redis libraries available for node, and they might work better for you.
-
-Comments and patches welcome.
-
## Contributors
Some people have have added features and fixed bugs in `node_redis` other than me.
@@ -421,6 +415,7 @@ In order of first contribution, they are:
* [Hank Sims](http://github.com/hanksims)
* [Aivo Paas](http://github.com/aivopaas)
* [Paul Carey](https://github.com/paulcarey)
+* [Pieter Noordhuis](https://github.com/pietern)
Thanks.
View
5 changelog.md
@@ -1,6 +1,11 @@
Changelog
=========
+## v0.4.0 - December 5, 2010
+
+Support for multiple response parsers and hiredis.
+Return Strings instead of Buffers by default.
+
## v0.3.9 - November 30, 2010
Fix parser bug on failed EXECs.
View
24 examples/unix_socket.js
@@ -1,5 +1,6 @@
var redis = require("redis"),
- client = redis.createClient("/tmp/redis.sock");
+ client = redis.createClient("/tmp/redis.sock"),
+ profiler = require("v8-profiler");
client.on("connect", function () {
console.log("Got Unix socket connection.")
@@ -9,7 +10,20 @@ client.on("error", function (err) {
console.log(err.message);
});
-client.info(function (err, reply) {
- console.log(reply.toString());
- client.quit();
-});
+client.set("space chars", "space value");
+
+setInterval(function () {
+ client.get("space chars");
+}, 100);
+
+function done() {
+ client.info(function (err, reply) {
+ console.log(reply.toString());
+ client.quit();
+ });
+}
+
+setTimeout(function () {
+ console.log("Taking snapshot.");
+ var snap = profiler.takeSnapshot();
+}, 5000);
View
105 index.js
@@ -2,22 +2,23 @@
var net = require("net"),
util = require("./lib/util").util,
+ Queue = require("./lib/queue").Queue,
events = require("events"),
- reply_parser,
+ parsers = [],
default_port = 6379,
default_host = "127.0.0.1";
-// Try to use hiredis for reply parsing and fall back on the Javascript-based
-// reply parsing code when its not available.
+// hiredis might not be installed
try {
- if (process.env["DISABLE_HIREDIS"])
- throw new Error(); // Fall back to the Javascript reply parsing code
- reply_parser = require("./lib/parser/hiredis");
-} catch(err) {
- reply_parser = require("./lib/parser/javascript");
+ require("./lib/parser/hiredis");
+ parsers.push(require("./lib/parser/hiredis"));
+} catch (err) {
+ console.log("hiredis parser not installed.");
}
-// can can set this to true to enable for all connections
+parsers.push(require("./lib/parser/javascript"));
+
+// can set this to true to enable for all connections
exports.debug_mode = false;
function to_array(args) {
@@ -31,62 +32,12 @@ function to_array(args) {
return arr;
}
-// Queue class adapted from Tim Caswell's pattern library
-// http://github.com/creationix/pattern/blob/master/lib/pattern/queue.js
-var Queue = function () {
- this.tail = [];
- this.head = to_array(arguments);
- this.offset = 0;
-};
-
-Queue.prototype.shift = function () {
- if (this.offset === this.head.length) {
- var tmp = this.head;
- tmp.length = 0;
- this.head = this.tail;
- this.tail = tmp;
- this.offset = 0;
- if (this.head.length === 0) {
- return;
- }
- }
- return this.head[this.offset++]; // sorry, JSLint
-};
-
-Queue.prototype.push = function (item) {
- return this.tail.push(item);
-};
-
-Queue.prototype.forEach = function (fn, thisv) {
- var array = this.head.slice(this.offset), i, il;
-
- array.push.apply(array, this.tail);
-
- if (thisv) {
- for (i = 0, il = array.length; i < il; i += 1) {
- fn.call(thisv, array[i], i, array);
- }
- } else {
- for (i = 0, il = array.length; i < il; i += 1) {
- fn(array[i], i, array);
- }
- }
-
- return array;
-};
-
-Object.defineProperty(Queue.prototype, 'length', {
- get: function () {
- return this.head.length - this.offset + this.tail.length;
- }
-});
-
function RedisClient(stream, options) {
events.EventEmitter.call(this);
this.stream = stream;
- this.options = options;
-
+ this.options = options || {};
+
this.connected = false;
this.connections = 0;
this.attempts = 1;
@@ -98,7 +49,31 @@ function RedisClient(stream, options) {
this.subscriptions = false;
this.closing = false;
- var self = this;
+ var parser_module, self = this;
+
+ if (self.options.parser) {
+ if (! parsers.some(function (parser) {
+ if (parser.name === self.options.parser) {
+ parser_module = parser;
+ if (exports.debug_mode) {
+ console.log("Using parser module: " + parser_module.name);
+ }
+ return true;
+ }
+ })) {
+ throw new Error("Couldn't find named parser " + self.options.parser + " on this system");
+ }
+ } else {
+ if (exports.debug_mode) {
+ console.log("Using default parser module: " + parsers[0].name);
+ }
+ parser_module = parsers[0];
+ }
+
+ parser_module.debug_mode = exports.debug_mode;
+ this.reply_parser = new parser_module.Parser({
+ return_buffers: self.options.return_buffers || false
+ });
this.stream.on("connect", function () {
if (exports.debug_mode) {
@@ -109,8 +84,6 @@ function RedisClient(stream, options) {
self.command_queue = new Queue();
self.emitted_end = false;
- reply_parser.debug_mode = exports.debug_mode;
- self.reply_parser = new reply_parser.Parser({ return_buffers: false });
// "reply error" is an error sent back by redis
self.reply_parser.on("reply error", function (reply) {
self.return_error(reply);
@@ -379,7 +352,7 @@ RedisClient.prototype.send_command = function () {
});
// Always use "Multi bulk commands", but if passed any Buffer args, then do multiple writes, one for each arg
- // This means that using Buffers in commands is going to be slower, so use Strings if you don't need binary.
+ // This means that using Buffers in commands is going to be slower, so use Strings if you don't already have a Buffer.
command_str = "*" + elem_count + "\r\n$" + command.length + "\r\n" + command + "\r\n";
@@ -563,7 +536,7 @@ exports.createClient = function (port_arg, host_arg, options) {
red_client, net_client;
net_client = net.createConnection(port, host);
-
+
red_client = new RedisClient(net_client, options);
red_client.port = port;
View
27 lib/parser/hiredis.js
@@ -1,38 +1,41 @@
+/*global Buffer require exports console setTimeout */
+
var events = require("events"),
util = require("../util").util,
hiredis = require("hiredis");
+exports.debug_mode = false;
+exports.name = "hiredis";
+
function HiredisReplyParser(options) {
+ this.name = exports.name;
this.options = options || {};
- this.return_buffers = this.options.return_buffers;
- if (this.return_buffers == undefined) this.return_buffers = true;
this.reset();
events.EventEmitter.call(this);
}
util.inherits(HiredisReplyParser, events.EventEmitter);
exports.Parser = HiredisReplyParser;
-exports.debug_mode = false;
-exports.type = "hiredis";
-HiredisReplyParser.prototype.reset = function() {
- this.reader = new hiredis.Reader({ return_buffers: this.return_buffers });
-}
+HiredisReplyParser.prototype.reset = function () {
+ this.reader = new hiredis.Reader({
+ return_buffers: this.options.return_buffers || false
+ });
+};
-HiredisReplyParser.prototype.execute = function(data) {
+HiredisReplyParser.prototype.execute = function (data) {
var reply;
this.reader.feed(data);
try {
while ((reply = this.reader.get()) !== undefined) {
- if (reply && reply.constructor == Error) {
+ if (reply && reply.constructor === Error) {
this.emit("reply error", reply);
} else {
this.emit("reply", reply);
}
}
- } catch(err) {
+ } catch (err) {
this.emit("error", err);
}
-}
-
+};
View
55 lib/parser/javascript.js
@@ -1,19 +1,21 @@
+/*global Buffer require exports console setTimeout */
+
var events = require("events"),
util = require("../util").util;
+exports.debug_mode = false;
+exports.name = "javascript";
+
function RedisReplyParser(options) {
+ this.name = exports.name;
this.options = options || {};
- this.return_buffers = this.options.return_buffers;
- if (this.return_buffers == undefined) this.return_buffers = true;
this.reset();
events.EventEmitter.call(this);
}
util.inherits(RedisReplyParser, events.EventEmitter);
exports.Parser = RedisReplyParser;
-exports.debug_mode = false;
-exports.type = "javascript";
// Buffer.toString() is quite slow for small strings
function small_toString(buf) {
@@ -39,13 +41,18 @@ RedisReplyParser.prototype.reset = function () {
this.multi_bulk_nested_replies = null;
};
+RedisReplyParser.prototype.parser_error = function (message) {
+ this.emit("error", message);
+ this.reset();
+};
+
RedisReplyParser.prototype.execute = function (incoming_buf) {
var pos = 0, bd_tmp, bd_str, i, il;
//, state_times = {}, start_execute = new Date(), start_switch, end_switch, old_state;
//start_switch = new Date();
while (pos < incoming_buf.length) {
- // old_state = this.state;
+ // old_state = this.state;
// console.log("execute: " + this.state + ", " + pos + "/" + incoming_buf.length + ", " + String.fromCharCode(incoming_buf[pos]));
switch (this.state) {
@@ -137,11 +144,11 @@ RedisReplyParser.prototype.execute = function (incoming_buf) {
this.send_reply(null);
this.multi_bulk_length = 0;
} else if (this.multi_bulk_length === 0) {
+ this.multi_bulk_replies = null;
this.send_reply([]);
}
} else {
- this.emit("error", new Error("didn't see LF after NL reading multi bulk count"));
- this.reset();
+ this.parser_error(new Error("didn't see LF after NL reading multi bulk count"));
return;
}
pos += 1;
@@ -171,13 +178,11 @@ RedisReplyParser.prototype.execute = function (incoming_buf) {
console.log("Growing return_buffer from " + this.return_buffer.length + " to " + this.bulk_length);
}
this.return_buffer = new Buffer(this.bulk_length);
- // home the old one gets cleaned up somehow
}
this.return_buffer.end = 0;
}
} else {
- this.emit("error", new Error("didn't see LF after NL while reading bulk length"));
- this.reset();
+ this.parser_error(new Error("didn't see LF after NL while reading bulk length"));
return;
}
pos += 1;
@@ -186,10 +191,9 @@ RedisReplyParser.prototype.execute = function (incoming_buf) {
this.return_buffer[this.return_buffer.end] = incoming_buf[pos];
this.return_buffer.end += 1;
pos += 1;
- // TODO - should be faster to use Bufer.copy() here, especially if the response is large.
- // However, when the response is small, Buffer.copy() seems a lot slower. Computers are hard.
if (this.return_buffer.end === this.bulk_length) {
bd_tmp = new Buffer(this.bulk_length);
+ // When the response is small, Buffer.copy() is a lot slower.
if (this.bulk_length > 10) {
this.return_buffer.copy(bd_tmp, 0, 0, this.bulk_length);
} else {
@@ -206,8 +210,7 @@ RedisReplyParser.prototype.execute = function (incoming_buf) {
this.state = "final lf";
pos += 1;
} else {
- this.emit("error", new Error("saw " + incoming_buf[pos] + " when expecting final CR"));
- this.reset();
+ this.parser_error(new Error("saw " + incoming_buf[pos] + " when expecting final CR"));
return;
}
break;
@@ -216,13 +219,12 @@ RedisReplyParser.prototype.execute = function (incoming_buf) {
this.state = "type";
pos += 1;
} else {
- this.emit("error", new Error("saw " + incoming_buf[pos] + " when expecting final LF"));
- this.reset();
+ this.parser_error(new Error("saw " + incoming_buf[pos] + " when expecting final LF"));
return;
}
break;
default:
- throw new Error("invalid state " + this.state);
+ this.parser_error(new Error("invalid state " + this.state));
}
// end_switch = new Date();
// if (state_times[old_state] === undefined) {
@@ -248,14 +250,23 @@ RedisReplyParser.prototype.send_error = function (reply) {
RedisReplyParser.prototype.send_reply = function (reply) {
if (this.multi_bulk_length > 0 || this.multi_bulk_nested_length > 0) {
- if (!this.return_buffers && Buffer.isBuffer(reply)) {
- this.add_multi_bulk_reply(reply.toString("utf8"));
+ if (!this.options.return_buffers && Buffer.isBuffer(reply)) {
+ if (reply.end > 10) {
+ this.add_multi_bulk_reply(reply.toString());
+ } else {
+ this.add_multi_bulk_reply(small_toString(reply));
+ }
} else {
this.add_multi_bulk_reply(reply);
}
} else {
- if (!this.return_buffers && Buffer.isBuffer(reply)) {
- this.emit("reply", reply.toString("utf8"));
+ if (!this.options.return_buffers && Buffer.isBuffer(reply)) {
+ console.log("converting buffer to string of len " + reply.end + ", " + util.inspect(reply));
+ if (reply.length > 10) {
+ this.emit("reply", reply.toString());
+ } else {
+ this.emit("reply", small_toString(reply));
+ }
} else {
this.emit("reply", reply);
}
@@ -265,7 +276,6 @@ RedisReplyParser.prototype.send_reply = function (reply) {
RedisReplyParser.prototype.add_multi_bulk_reply = function (reply) {
if (this.multi_bulk_replies) {
this.multi_bulk_replies.push(reply);
- // use "less than" here because a nil mb reply claims "0 length", but we need 1 slot to hold it
if (this.multi_bulk_replies.length < this.multi_bulk_length) {
return;
}
@@ -288,4 +298,3 @@ RedisReplyParser.prototype.add_multi_bulk_reply = function (reply) {
this.multi_bulk_replies = null;
}
};
-
View
3 lib/util.js
@@ -1,7 +1,6 @@
if (process.versions.node.match(/^0.3/)) {
exports.util = require("util");
} else {
- /* This module is called "sys" in 0.2.x */
+ // This module is called "sys" in 0.2.x
exports.util = require("sys");
}
-
View
38 multi_bench.js
@@ -46,19 +46,26 @@ tests.push({
});
function create_clients(callback) {
- if (active_clients == num_clients) {
+ if (active_clients === num_clients) {
+ // common case is all clients are already created
+ console.log("create_clients: all clients already created " + num_clients);
callback();
} else {
- var client;
- var connected = active_clients;
+ var client, connected = active_clients;
while (active_clients < num_clients) {
- client = clients[active_clients++] = redis.createClient();
+ client = clients[active_clients++] = redis.createClient(6379, "127.0.0.1", {
+ parser: "hiredis",
+ return_buffers: false
+ });
client.on("connect", function() {
- /* Fire callback when all clients are connected */
- if (++connected == num_clients)
+ // Fire callback when all clients are connected
+ connected += 1;
+ if (connected === num_clients) {
callback();
+ }
});
+ // TODO - need to check for client disconnect
client.on("error", function (msg) {
console.log("Connect problem:" + msg.stack);
});
@@ -68,10 +75,10 @@ function create_clients(callback) {
function issue_request(client, test, cmd, args) {
var i = issued_requests++;
- latency[i] = new Date;
+ latency[i] = Date.now();
client[cmd](args, function() {
- latency[i] = (new Date) - latency[i];
+ latency[i] = Date.now() - latency[i];
if (issued_requests < num_requests) {
issue_request(client, test, cmd, args);
} else {
@@ -84,11 +91,11 @@ function issue_request(client, test, cmd, args) {
function test_run(test) {
create_clients(function() {
- var i = num_clients;
- var cmd = test.command[0];
- var args = test.command.slice(1);
+ var i = num_clients,
+ cmd = test.command[0],
+ args = test.command.slice(1);
- test_start = new Date;
+ test_start = Date.now();
issued_requests = 0;
while(i-- && issued_requests < num_requests) {
issue_request(clients[i], test, cmd, args);
@@ -98,7 +105,7 @@ function test_run(test) {
function test_complete(test) {
var min, max, sum, avg;
- var total_time = (new Date) - test_start;
+ var total_time = Date.now() - test_start;
var op_rate = (issued_requests / (total_time / 1000.0)).toFixed(2);
var i;
@@ -116,8 +123,9 @@ function test_complete(test) {
function next() {
var test = tests.shift();
- if (test) test_run(test);
+ if (test) {
+ test_run(test);
+ }
}
next();
-
View
5 package.json
@@ -1,5 +1,5 @@
{ "name" : "redis",
- "version" : "0.3.9",
+ "version" : "0.4.0",
"description" : "Redis client library",
"author": "Matt Ranney <mjr@ranney.com>",
"contributors": [
@@ -9,7 +9,8 @@
"Orion Henry",
"Hank Sims",
"Aivo Paas",
- "Paul Carey"
+ "Paul Carey",
+ "Pieter Noordhuis"
],
"main": "./index.js",
"scripts": {
View
23 test.js
@@ -1,6 +1,8 @@
/*global require console setTimeout process Buffer */
var redis = require("./index"),
- client = redis.createClient(),
+ client = redis.createClient(6379, "127.0.0.1", {
+ parser: "javascript"
+ }),
client2 = redis.createClient(),
client3 = redis.createClient(),
assert = require("assert"),
@@ -12,7 +14,7 @@ var redis = require("./index"),
server_info;
// Uncomment this to see the wire protocol and other debugging info
-//redis.debug_mode = true;
+redis.debug_mode = true;
function buffers_to_strings(arr) {
return arr.map(function (val) {
@@ -149,8 +151,15 @@ tests.MULTI_3 = function () {
client.sadd("some set", "mem 2");
client.sadd("some set", "mem 3");
client.sadd("some set", "mem 4");
+
+ // make sure empty mb reply works
+ client.del("some missing set");
+ client.smembers("some missing set", function (err, reply) {
+ // make sure empty mb reply works
+ assert.strictEqual(true, is_empty_array(reply), name);
+ });
- // test nested multi-bulk replies with nulls.
+ // test nested multi-bulk replies with empty mb elements.
client.multi([
["smembers", "some set"],
["del", "some set"],
@@ -479,7 +488,7 @@ tests.HGETALL = function () {
client.HGETALL(["hosts"], function (err, obj) {
assert.strictEqual(null, err, name + " result sent back unexpected error: " + err);
assert.strictEqual(3, Object.keys(obj).length, name);
- assert.ok(Buffer.isBuffer(obj.mjr), name);
+// assert.ok(Buffer.isBuffer(obj.mjr), name);
assert.strictEqual("1", obj.mjr.toString(), name);
assert.strictEqual("23", obj.another.toString(), name);
assert.strictEqual("1234", obj.home.toString(), name);
@@ -1003,6 +1012,8 @@ function run_next_test() {
}
}
+console.log("Using reply parser " + client.reply_parser.name);
+
client.on("connect", function () {
// Fetch and stash info results in case anybody needs info on the server we are using.
client.info(function (err, reply) {
@@ -1018,10 +1029,12 @@ client.on("connect", function () {
obj.versions.push(+num);
});
server_info = obj;
+ console.log("Connected to " + client.host + ":" + client.port + ", Redis server version " + obj.redis_version + "\n");
+
+ run_next_test();
});
connected = true;
- run_next_test();
});
client.on('end', function () {

0 comments on commit b907364

Please sign in to comment.