send_command_buf and b_ prefixed commands

Author: vitaliylag

README.md

@@ -109,6 +109,12 @@ client.get("missingkey", function(err, reply) {
 });
 ```
 
+If you want receive reply as Buffers instead of Strings just add "b_" prefix before command name:
+
+```js
+client.b_get("some key", function(err, buf) {});
+```
+
 For a list of Redis commands, see [Redis Command Reference](http://redis.io/commands)
 
 Minimal parsing is done on the replies. Commands that return a integer return JavaScript Numbers, arrays return JavaScript Array. `HGETALL` returns an Object keyed by the hash keys. All strings will either be returned as string or as buffer depending on your setting.
@@ -666,6 +672,10 @@ The command_name has to be lower case.
 
 All commands are sent as multi-bulk commands. `args` can either be an Array of arguments, or omitted / set to undefined.
 
+## client.send_command_buf(command_name[, [args][, callback]])
+
+The same as client.send_command but sending reply as Buffers instead of Strings.
+
 ## client.connected
 
 Boolean tracking the state of the connection to the Redis server.

index.js

@@ -27,19 +27,6 @@ if (typeof EventEmitter !== 'function') {
 
 function noop () {}
 
-function handle_detect_buffers_reply (reply, command, buffer_args) {
-    if (buffer_args === false || this.message_buffers) {
-        // If detect_buffers option was specified, then the reply from the parser will be a buffer.
-        // If this command did not use Buffer arguments, then convert the reply to Strings here.
-        reply = utils.reply_to_strings(reply);
-    }
-
-    if (command === 'hgetall') {
-        reply = utils.reply_to_object(reply);
-    }
-    return reply;
-}
-
 exports.debug_mode = /\bredis\b/i.test(process.env.NODE_DEBUG);
 
 // Attention: The second parameter might be removed at will and is not officially supported.
@@ -112,10 +99,6 @@ function RedisClient (options, stream) {
         self.warn('WARNING: You activated return_buffers and detect_buffers at the same time. The return value is always going to be a buffer.');
         options.detect_buffers = false;
     }
-    if (options.detect_buffers) {
-        // We only need to look at the arguments if we do not know what we have to return
-        this.handle_reply = handle_detect_buffers_reply;
-    }
     this.should_buffer = false;
     this.max_attempts = options.max_attempts | 0;
     if ('max_attempts' in options) {
@@ -144,18 +127,19 @@ function RedisClient (options, stream) {
     this.pub_sub_mode = 0;
     this.subscription_set = {};
     this.monitoring = false;
-    this.message_buffers = false;
+    this.message_buffers = false; // Do we have subscribes on message_buffer event
     this.closing = false;
     this.server_info = {};
     this.auth_pass = options.auth_pass || options.password;
     this.selected_db = options.db; // Save the selected db here, used when reconnecting
     this.old_state = null;
     this.fire_strings = true; // Determine if strings or buffers should be written to the stream
+    this.cur_command_ret_buf = 0;
     this.pipeline = false;
     this.sub_commands_left = 0;
     this.times_connected = 0;
     this.options = options;
-    this.buffers = options.return_buffers || options.detect_buffers;
+    this.using_buffer_parser = options.return_buffers || options.detect_buffers;
     this.reply = 'ON'; // Returning replies is the default
     // Init parser
     this.reply_parser = create_parser(this, options);
@@ -172,17 +156,23 @@ function RedisClient (options, stream) {
                 'The drain event listener is deprecated and will be removed in v.3.0.0.\n' +
                 'If you want to keep on listening to this event please listen to the stream drain event directly.'
             );
-        } else if (event === 'message_buffer' || event === 'pmessage_buffer' || event === 'messageBuffer' || event === 'pmessageBuffer' && !this.buffers) {
+        } else if (event === 'message_buffer' || event === 'pmessage_buffer' || event === 'messageBuffer' || event === 'pmessageBuffer') {
             this.message_buffers = true;
-            this.handle_reply = handle_detect_buffers_reply;
-            this.reply_parser = create_parser(this);
+            if (!this.using_buffer_parser) {
+                this.switchToBufferParser();
+            }
         }
     });
 }
 util.inherits(RedisClient, EventEmitter);
 
 RedisClient.connection_id = 0;
 
+RedisClient.prototype.switchToBufferParser = function() {
+    this.using_buffer_parser = true;
+    this.reply_parser = create_parser(this);
+}
+
 function create_parser (self) {
     return Parser({
         returnReply: function (data) {
@@ -208,7 +198,7 @@ function create_parser (self) {
             self.emit('error', err);
             self.create_stream();
         },
-        returnBuffers: self.buffers || self.message_buffers,
+        returnBuffers: self.using_buffer_parser,
         name: self.options.parser,
         stringNumbers: self.options.string_numbers
     });
@@ -304,7 +294,15 @@ RedisClient.prototype.create_stream = function () {
     }
 };
 
-RedisClient.prototype.handle_reply = function (reply, command) {
+RedisClient.prototype.handle_reply = function (reply, command, buffer_reply) {
+    if (!buffer_reply && this.using_buffer_parser) {
+        // Reply from parser will be Buffer if:
+        // 1) return_buffers option set to true
+        // 2) or detect_buffers option set to true and command used Buffer arguments
+        // 3) or buffer_reply argument was set to true when calling internal_send_command
+        reply = utils.reply_to_strings(reply);
+    }
+
     if (command === 'hgetall') {
         reply = utils.reply_to_object(reply);
     }
@@ -531,7 +529,7 @@ RedisClient.prototype.ready_check = function () {
 RedisClient.prototype.send_offline_queue = function () {
     for (var command_obj = this.offline_queue.shift(); command_obj; command_obj = this.offline_queue.shift()) {
         debug('Sending offline command: ' + command_obj.command);
-        this.internal_send_command(command_obj.command, command_obj.args, command_obj.callback, command_obj.call_on_write);
+        this.internal_send_command(command_obj.command, command_obj.args, command_obj.callback, command_obj.call_on_write, command_obj.buffer_reply);
     }
     this.drain();
 };
@@ -716,7 +714,7 @@ function normal_reply (self, reply) {
     var command_obj = self.command_queue.shift();
     if (typeof command_obj.callback === 'function') {
         if (command_obj.command !== 'exec') {
-            reply = self.handle_reply(reply, command_obj.command, command_obj.buffer_args);
+            reply = self.handle_reply(reply, command_obj.command, command_obj.buffer_reply);
         }
         command_obj.callback(null, reply);
     } else {
@@ -728,7 +726,7 @@ function subscribe_unsubscribe (self, reply, type) {
     // Subscribe commands take an optional callback and also emit an event, but only the _last_ response is included in the callback
     // The pub sub commands return each argument in a separate return value and have to be handled that way
     var command_obj = self.command_queue.get(0);
-    var buffer = self.options.return_buffers || self.options.detect_buffers && command_obj.buffer_args;
+    var buffer = command_obj.buffer_reply;
     var channel = (buffer || reply[1] === null) ? reply[1] : reply[1].toString();
     var count = +reply[2]; // Return the channel counter as number no matter if `string_numbers` is activated or not
     debug(type, channel);
@@ -803,7 +801,7 @@ RedisClient.prototype.return_reply = function (reply) {
     // the average performance of all other commands in case of no monitor mode
     if (this.monitoring) {
         var replyStr;
-        if (this.buffers && Buffer.isBuffer(reply)) {
+        if (this.using_buffer_parser && Buffer.isBuffer(reply)) {
             replyStr = reply.toString();
         } else {
             replyStr = reply;
@@ -864,9 +862,17 @@ function handle_offline_command (self, command_obj) {
     self.should_buffer = true;
 }
 
+RedisClient.prototype.internal_send_command_buf = function (command, args, callback, call_on_write) {
+    return this.internal_send_command(command, args, callback, call_on_write, true);
+};
+
 // Do not call internal_send_command directly, if you are not absolutly certain it handles everything properly
 // e.g. monitor / info does not work with internal_send_command only
-RedisClient.prototype.internal_send_command = function (command, args, callback, call_on_write) {
+RedisClient.prototype.internal_send_command = function (command, args, callback, call_on_write, buffer_reply) {
+    if (this.cur_command_ret_buf) {
+        buffer_reply = true; //needed for send_command_buf and individual commands
+    }
+
     var arg, prefix_keys, command_obj;
     var i = 0;
     var command_str = '';
@@ -881,10 +887,14 @@ RedisClient.prototype.internal_send_command = function (command, args, callback,
 
     if (this.ready === false || this.stream.writable === false) {
         // Handle offline commands right away
-        handle_offline_command(this, new OfflineCommand(command, args, callback, call_on_write));
+        handle_offline_command(this, new OfflineCommand(command, args, callback, call_on_write, buffer_reply));
         return false; // Indicate buffering
     }
 
+    if (buffer_reply && !this.using_buffer_parser) {
+        this.switchToBufferParser();
+    }
+
     for (i = 0; i < len; i += 1) {
         if (typeof args[i] === 'string') {
             // 30000 seemed to be a good value to switch to buffers after testing and checking the pros and cons
@@ -929,7 +939,7 @@ RedisClient.prototype.internal_send_command = function (command, args, callback,
         }
     }
     // Pass the original args to make sure in error cases the original arguments are returned
-    command_obj = new Command(command, args, buffer_args, callback);
+    command_obj = new Command(command, args, this.options.return_buffers || buffer_args && this.options.detect_buffers || buffer_reply, callback);
 
     if (this.options.prefix) {
         prefix_keys = commands.getKeyIndexes(command, args_copy);

lib/command.js

@@ -2,18 +2,19 @@
 
 // This Command constructor is ever so slightly faster than using an object literal, but more importantly, using
 // a named constructor helps it show up meaningfully in the V8 CPU profiler and in heap snapshots.
-function Command (command, args, buffer_args, callback) {
+function Command (command, args, buffer_reply, callback) {
     this.command = command;
     this.args = args;
-    this.buffer_args = buffer_args;
+    this.buffer_reply = buffer_reply;
     this.callback = callback;
 }
 
-function OfflineCommand (command, args, callback, call_on_write) {
+function OfflineCommand (command, args, callback, call_on_write, buffer_reply) {
     this.command = command;
     this.args = args;
     this.callback = callback;
     this.call_on_write = call_on_write;
+    this.buffer_reply = buffer_reply;
 }
 
 module.exports = {