Passing a stream as second parameter in the constructor

To support private streams this is implemented.
This needs some monkey patching to work, so it is not officially
supported and might be removed at any time!

Fixes #950
Closes #951

Author: Ruben Bridgewater

index.js

@@ -34,13 +34,20 @@ function handle_detect_buffers_reply (reply, command, buffer_args) {
 
 exports.debug_mode = /\bredis\b/i.test(process.env.NODE_DEBUG);
 
-function RedisClient (options) {
+// Attention: The second parameter might be removed at will and is not officially supported.
+// Do not rely on this
+function RedisClient (options, stream) {
     // Copy the options so they are not mutated
     options = utils.clone(options);
     EventEmitter.call(this);
     var cnx_options = {};
     var self = this;
-    if (options.path) {
+    if (stream) {
+        // The stream from the outside is used so no connection from this side is triggered but from the server this client should talk to
+        // Reconnect etc won't work with this. This requires monkey patching to work, so it is not officially supported
+        options.stream = stream;
+        this.address = '"Private stream"';
+    } else if (options.path) {
         cnx_options.path = options.path;
         this.address = options.path;
     } else {
@@ -174,17 +181,25 @@ RedisClient.connection_id = 0;
 RedisClient.prototype.create_stream = function () {
     var self = this;
 
-    // On a reconnect destroy the former stream and retry
-    if (this.stream) {
-        this.stream.removeAllListeners();
-        this.stream.destroy();
-    }
-
-    /* istanbul ignore if: travis does not work with stunnel atm. Therefor the tls tests are skipped on travis */
-    if (this.options.tls) {
-        this.stream = tls.connect(this.connection_options);
+    if (this.options.stream) {
+        // Only add the listeners once in case of a reconnect try (that won't work)
+        if (this.stream) {
+            return;
+        }
+        this.stream = this.options.stream;
     } else {
-        this.stream = net.createConnection(this.connection_options);
+        // On a reconnect destroy the former stream and retry
+        if (this.stream) {
+            this.stream.removeAllListeners();
+            this.stream.destroy();
+        }
+
+        /* istanbul ignore if: travis does not work with stunnel atm. Therefor the tls tests are skipped on travis */
+        if (this.options.tls) {
+            this.stream = tls.connect(this.connection_options);
+        } else {
+            this.stream = net.createConnection(this.connection_options);
+        }
     }
 
     if (this.options.connect_timeout) {

test/connection.spec.js

@@ -5,20 +5,38 @@ var config = require("./lib/config");
 var helper = require('./helper');
 var redis = config.redis;
 var intercept = require('intercept-stdout');
+var net = require('net');
+var client;
 
 describe("connection tests", function () {
-    helper.allTests(function(parser, ip, args) {
 
-        describe("using " + parser + " and " + ip, function () {
+    beforeEach(function () {
+        client = null;
+    });
+    afterEach(function () {
+        client.end(true);
+    });
+
+    it('unofficially support for a private stream', function () {
+        // While using a private stream, reconnection and other features are not going to work properly.
+        // Besides that some functions also have to be monkey patched to be safe from errors in this case.
+        // Therefor this is not officially supported!
+        var socket = new net.Socket();
+        client = new redis.RedisClient({
+            prefix: 'test'
+        }, socket);
+        assert.strictEqual(client.stream, socket);
+        assert.strictEqual(client.stream.listeners('error').length, 1);
+        assert.strictEqual(client.address, '"Private stream"');
+        // Pretent a reconnect event
+        client.create_stream();
+        assert.strictEqual(client.stream, socket);
+        assert.strictEqual(client.stream.listeners('error').length, 1);
+    });
 
-            var client;
+    helper.allTests(function(parser, ip, args) {
 
-            beforeEach(function () {
-                client = null;
-            });
-            afterEach(function () {
-                client.end(true);
-            });
+        describe("using " + parser + " and " + ip, function () {
 
             describe("on lost connection", function () {
                 it("emit an error after max retry attempts and do not try to reconnect afterwards", function (done) {