add enable_persistent_queue option

README.md

@@ -202,6 +202,7 @@ connection to the redis server, commands are added to a queue and are executed
 once the connection has been established. Setting `enable_offline_queue` to
 `false` will disable this feature and the callback will be execute immediately
 with an error, or an error will be thrown if no callback is specified.
+* `persist_offline_queue`: Defaults to `false`. By default the offline queue will be flushed after one retry attempt, although this behavior may be changed in the future. If `persist_offline_queue` is set to `true`, the offline queue will never be flushed outside of sending it. This includes when a connection is closed by the user. However, outstanding commands are still flushed, for instance, when a connection breaks after writing to the output stream, so client code should not rely on the assumption that offline queue items are always submitted when the connection is restored.
 * `retry_max_delay`: defaults to `null`. By default every time the client tries to connect and fails time before
 reconnection (delay) almost doubles. This delay normally grows infinitely, but setting `retry_max_delay` limits delay
 to maximum value, provided in milliseconds.

index.js

@@ -55,6 +55,11 @@ function RedisClient(stream, options) {
     if (typeof this.options.enable_offline_queue === "boolean") {
         this.enable_offline_queue = this.options.enable_offline_queue;
     }
+
+    this.persist_offline_queue = false;
+    if(typeof this.options.persist_offline_queue === "boolean") {
+        this.persist_offline_queue = this.options.persist_offline_queue;
+    }
     this.retry_max_delay = null;
     if (options.retry_max_delay !== undefined && !isNaN(options.retry_max_delay) && options.retry_max_delay > 0) {
         this.retry_max_delay = options.retry_max_delay;
@@ -115,20 +120,22 @@ RedisClient.prototype.initialize_retry_vars = function () {
 // flush offline_queue and command_queue, erroring any items with a callback first
 RedisClient.prototype.flush_and_error = function (message) {
     var command_obj;
-    while (this.offline_queue.length > 0) {
-        command_obj = this.offline_queue.shift();
-        if (typeof command_obj.callback === "function") {
-            try {
-                command_obj.callback(message);
-            } catch (callback_err) {
-                process.nextTick(function () {
-                    throw callback_err;
-                });
+
+    if (!this.persist_offline_queue) {
+        while (this.offline_queue.length > 0) {
+            command_obj = this.offline_queue.shift();
+            if (typeof command_obj.callback === "function") {
+                try {
+                    command_obj.callback(message);
+                } catch (callback_err) {
+                    process.nextTick(function () {
+                        throw callback_err;
+                    });
+                }
             }
         }
+        this.offline_queue = new Queue();
     }
-    this.offline_queue = new Queue();
-
     while (this.command_queue.length > 0) {
         command_obj = this.command_queue.shift();
         if (typeof command_obj.callback === "function") {