Fix saving buffers with charsets other than utf-8 while using multi

This will also improve pipelinening for buffers and fixes the return value of Batch.exec

Fixes #913

changelog.md

@@ -5,6 +5,7 @@ Changelog
 
 Bugfixes
 
+-  Fixed saving buffers with charsets other than utf-8 while using multi ([@BridgeAR](https://github.com/BridgeAR))
 -  Fixed js parser handling big values very slow ([@BridgeAR](https://github.com/BridgeAR))
  -  The speed is up to ~500% faster than before but still up to ~50% slower than the hiredis parser.
 

index.js

@@ -731,12 +731,20 @@ RedisClient.prototype.send_command = function (command, args, callback) {
     for (i = 0; i < args.length; i += 1) {
         if (Buffer.isBuffer(args[i])) {
             buffer_args = true;
+            if (this.pipeline !== 0) {
+                this.pipeline += 2;
+                this.writeDefault = this.writeBuffers;
+            }
         } else if (typeof args[i] !== 'string') {
             args[i] = String(args[i]);
         // 30000 seemed to be a good value to switch to buffers after testing and checking the pros and cons
         } else if (args[i].length > 30000) {
             big_data = true;
             args[i] = new Buffer(args[i]);
+            if (this.pipeline !== 0) {
+                this.pipeline += 2;
+                this.writeDefault = this.writeBuffers;
+            }
         }
     }
     if (this.options.detect_buffers) {
@@ -764,7 +772,7 @@ RedisClient.prototype.send_command = function (command, args, callback) {
             this.should_buffer = true;
         }
         // Return false to signal buffering
-        return false;
+        return !this.should_buffer;
     }
 
     if (command === 'subscribe' || command === 'psubscribe' || command === 'unsubscribe' || command === 'punsubscribe') {
@@ -797,7 +805,7 @@ RedisClient.prototype.send_command = function (command, args, callback) {
 
         for (i = 0; i < args.length; i += 1) {
             arg = args[i];
-            if (!Buffer.isBuffer(arg)) {
+            if (typeof arg === 'string') {
                 this.write('$' + Buffer.byteLength(arg) + '\r\n' + arg + '\r\n');
             } else {
                 this.write('$' + arg.length + '\r\n');
@@ -810,6 +818,22 @@ RedisClient.prototype.send_command = function (command, args, callback) {
     return !this.should_buffer;
 };
 
+RedisClient.prototype.writeDefault = RedisClient.prototype.writeStrings = function (data) {
+    var command, str = '';
+    while (command = this.pipeline_queue.shift()) {
+        str += command;
+    }
+    this.should_buffer = !this.stream.write(str + data);
+};
+
+RedisClient.prototype.writeBuffers = function (data) {
+    var command;
+    while (command = this.pipeline_queue.shift()) {
+        this.stream.write(command);
+    }
+    this.should_buffer = !this.stream.write(data);
+};
+
 RedisClient.prototype.write = function (data) {
     if (this.pipeline === 0) {
         this.should_buffer = !this.stream.write(data);
@@ -818,11 +842,7 @@ RedisClient.prototype.write = function (data) {
 
     this.pipeline--;
     if (this.pipeline === 0) {
-        var command, str = '';
+        this.writeDefault(data);
-        while (command = this.pipeline_queue.shift()) {
-            str += command;
-        }
-        this.should_buffer = !this.stream.write(str + data);
         return;
     }
 
@@ -1121,10 +1141,12 @@ Multi.prototype.exec_transaction = function (callback) {
         this.send_command(command, args, index, cb);
     }
 
-    this._client.uncork();
+    this._client.send_command('exec', [], function(err, replies) {
-    return this._client.send_command('exec', [], function(err, replies) {
         self.execute_callback(err, replies);
     });
+    this._client.uncork();
+    this._client.writeDefault = this._client.writeStrings;
+    return !this._client.should_buffer;
 };
 
 Multi.prototype.execute_callback = function (err, replies) {
@@ -1231,7 +1253,8 @@ Multi.prototype.exec = Multi.prototype.EXEC = Multi.prototype.exec_batch = funct
         index++;
     }
     this._client.uncork();
-    return this._client.should_buffer;
+    this._client.writeDefault = this._client.writeStrings;
+    return !this._client.should_buffer;
 };
 
 var createClient = function (port_arg, host_arg, options) {

test/detect_buffers.spec.js

@@ -27,6 +27,10 @@ describe("detect_buffers", function () {
                 });
             });
 
+            afterEach(function () {
+                client.end();
+            });
+
             describe('get', function () {
                 describe('first argument is a string', function () {
                     it('returns a string', function (done) {

test/helper.js

@@ -156,7 +156,9 @@ module.exports = {
             i++;
             if (i === max) {
                 func();
+                return true;
             }
+            return false;
         };
     },
     killConnection: function (client) {

test/multi.spec.js

@@ -1,13 +1,74 @@
 'use strict';
 
 var assert = require('assert');
-var config = require("../lib/config");
+var config = require("./lib/config");
-var helper = require('../helper');
+var helper = require('./helper');
 var redis = config.redis;
+var zlib = require('zlib');
 var uuid = require('uuid');
+var client;
 
 describe("The 'multi' method", function () {
 
+    afterEach(function () {
+        client.end();
+    });
+
+    describe('regression test', function () {
+        it('saved buffers with charsets different than utf-8 (issue #913)', function (done) {
+            client = redis.createClient();
+
+            var end = helper.callFuncAfter(done, 100);
+
+            // Some random object created from http://beta.json-generator.com/
+            var test_obj = {
+                "_id": "5642c4c33d4667c4a1fefd99","index": 0, "guid": "5baf1f1c-7621-41e7-ae7a-f8c6f3199b0f", "isActive": true,
+                "balance": "$1,028.63", "picture": "http://placehold.it/32x32", "age": 31, "eyeColor": "green", "name": {"first": "Shana", "last": "Long"},
+                "company": "MANGLO", "email": "shana.long@manglo.us", "phone": "+1 (926) 405-3105", "address": "747 Dank Court, Norfolk, Ohio, 1112",
+                "about": "Eu pariatur in nisi occaecat enim qui consequat nostrud cupidatat id. " +
+                    "Commodo commodo dolore esse irure minim quis deserunt anim laborum aute deserunt et est. Quis nisi laborum deserunt nisi quis.",
+                "registered": "Friday, April 18, 2014 9:56 AM", "latitude": "74.566613", "longitude": "-11.660432", "tags": [7, "excepteur"],
+                "range": [0, 1, 2, 3, 4, 5, 6, 7, 8, 9], "friends": [3, {"id": 1, "name": "Schultz Dyer"}],
+                "greeting": "Hello, Shana! You have 5 unread messages.", "favoriteFruit": "strawberry"
+            };
+
+            function run () {
+                if (end() === true) {
+                    return;
+                }
+                // To demonstrate a big payload for hash set field values, let's create a big array
+                var test_arr = [];
+                for (var i = 0; i < 80; i++) {
+                    var new_obj = JSON.parse(JSON.stringify(test_obj));
+                    test_arr.push(new_obj);
+                }
+
+                var json = JSON.stringify(test_arr);
+                zlib.deflate(new Buffer(json), function (err, buffer) {
+                    if (err) {
+                        done(err);
+                        return;
+                    }
+
+                    var multi = client.multi();
+                    multi.del('SOME_KEY');
+
+                    for (i = 0; i < 100; i++) {
+                        multi.hset('SOME_KEY', 'SOME_FIELD' + i, buffer);
+                    }
+                    multi.exec(function (err, res) {
+                        if (err) {
+                            done(err);
+                            return;
+                        }
+                        run();
+                    });
+                });
+            }
+            run();
+        });
+    });
+
     helper.allTests(function(parser, ip, args) {
 
         describe("using " + parser + " and " + ip, function () {
@@ -19,14 +80,13 @@ describe("The 'multi' method", function () {
             });
 
             describe("when not connected", function () {
-                var client;
 
                 beforeEach(function (done) {
                     client = redis.createClient.apply(redis.createClient, args);
                     client.once("ready", function () {
                         client.quit();
                     });
-                    client.on('end', function () {
+                    client.once('end', function () {
                         return done();
                     });
                 });
@@ -37,7 +97,7 @@ describe("The 'multi' method", function () {
                         assert(err.message.match(/The connection has already been closed/));
                         done();
                     });
-                    assert.strictEqual(notBuffering, false);
+                    assert.strictEqual(notBuffering, true);
                 });
 
                 it("reports an error if promisified", function () {
@@ -48,17 +108,12 @@ describe("The 'multi' method", function () {
             });
 
             describe("when connected", function () {
-                var client;
 
                 beforeEach(function (done) {
                     client = redis.createClient.apply(redis.createClient, args);
                     client.once("connect", done);
                 });
 
-                afterEach(function () {
-                    client.end();
-                });
-
                 it("executes a pipelined multi properly in combination with the offline queue", function (done) {
                     var multi1 = client.multi();
                     multi1.set("m1", "123");
@@ -93,11 +148,6 @@ describe("The 'multi' method", function () {
             });
 
             describe("when connection is broken", function () {
-                var client;
-
-                afterEach(function () {
-                    client.end();
-                });
 
                 it("return an error even if connection is in broken mode if callback is present", function (done) {
                     client = redis.createClient({
@@ -137,7 +187,6 @@ describe("The 'multi' method", function () {
             });
 
             describe("when ready", function () {
-                var client;
 
                 beforeEach(function (done) {
                     client = redis.createClient.apply(redis.createClient, args);
@@ -148,10 +197,6 @@ describe("The 'multi' method", function () {
                     });
                 });
 
-                afterEach(function () {
-                    client.end();
-                });
-
                 it("returns an empty result array", function (done) {
                     var multi = client.multi();
                     var notBuffering = multi.exec(function (err, res) {