Support for multiple reply parsers including hiredis.

Several parsing bugs fixed in JavaScript.
Some new config options that need to be better documented.

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+.
+This is a complete Redis client for node.js.  It supports all Redis commands, including MULTI, WATCH, and PUBLISH/SUBSCRIBE.
-It might not work on earlier versions of either, although it probably will.
-
-This client supports all Redis commands, including MULTI 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 
+`node_redis` was originally written to replace `node-redis-client` which hasn't been updated in a while, and no longer works
-distribution, and `bench.js` is as fast as `redis-benchmark` for some patterns and slower for others.  `node_redis` has many lovingly
+on recent versions of node.
-hand-crafted optimizations for speed.
 
 
 ## 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 
+Stream large set/get values into and out of Redis.  Otherwise the entire value must be in node's memory.
-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.
 
 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.
 

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.

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) {
+client.set("space chars", "space value");
-    console.log(reply.toString());
+
-    client.quit();
+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);

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
+// hiredis might not be installed
-// reply parsing code when its not available.
 try {
-    if (process.env["DISABLE_HIREDIS"])
+    require("./lib/parser/hiredis");
-        throw new Error(); // Fall back to the Javascript reply parsing code
+    parsers.push(require("./lib/parser/hiredis"));
-    reply_parser = require("./lib/parser/hiredis");
+} catch (err) {
-} catch(err) {
+    console.log("hiredis parser not installed.");
-    reply_parser = require("./lib/parser/javascript");
 }
 
-// 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;

lib/parser/hiredis.js

@@ -1,11 +1,15 @@
+/*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);
 }
@@ -13,26 +17,25 @@ function HiredisReplyParser(options) {
 util.inherits(HiredisReplyParser, events.EventEmitter);
 
 exports.Parser = HiredisReplyParser;
-exports.debug_mode = false;
-exports.type = "hiredis";
 
-HiredisReplyParser.prototype.reset = function() {
+HiredisReplyParser.prototype.reset = function () {
-    this.reader = new hiredis.Reader({ return_buffers: this.return_buffers });
+    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);
     }
-}
+};
-

lib/parser/javascript.js

@@ -1,10 +1,14 @@
+/*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);
 }
@@ -12,8 +16,6 @@ function RedisReplyParser(options) {
 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.parser_error(new Error("didn't see LF after NL reading multi bulk count"));
-                this.reset();
                 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.parser_error(new Error("didn't see LF after NL while reading bulk length"));
-                this.reset();
                 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.parser_error(new Error("saw " + incoming_buf[pos] + " when expecting final CR"));
-                this.reset();
                 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.parser_error(new Error("saw " + incoming_buf[pos] + " when expecting final LF"));
-                this.reset();
                 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)) {
+        if (!this.options.return_buffers && Buffer.isBuffer(reply)) {
-            this.add_multi_bulk_reply(reply.toString("utf8"));
+            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)) {
+        if (!this.options.return_buffers && Buffer.isBuffer(reply)) {
-            this.emit("reply", reply.toString("utf8"));
+            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;
     }
 };
-

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");
 }
-

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 client, connected = active_clients;
-        var 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", {
-            client.on("connect", function() {
+                parser: "hiredis",
-                /* Fire callback when all clients are connected */
+                return_buffers: false
-                if (++connected == num_clients)
-                    callback();
             });
+            client.on("connect", function() {
+                // 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 i = num_clients, 
-        var cmd = test.command[0];
+            cmd = test.command[0],
-        var args = test.command.slice(1);
+            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();
-

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": {

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 () {