feat(pipeline): add flag to disable raising exceptions

Author: philippeboyd

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 - Made various performance optimizations to the Ruby driver. See #184.
 - Always assume UTF-8 encoding instead of relying on `Encoding.default_external`.
+- Add `raise_exception` flag in `pipelined` allowing failed commands to be returned in the result array when set to `false`. See #187.
 
 # 0.21.1
 

README.md

@@ -332,6 +332,27 @@ end
 # => ["OK", 1]
 ```
 
+#### Exception management
+
+The `raise_exception` flag in the `#pipelined` method of `RedisClient` is a feature that modifies the pipeline execution
+behavior. When set to `false`, it doesn't raise an exception when a command error occurs. Instead, it allows the
+pipeline to execute all commands, and any failed command will be available in the returned array. (Defaults to `true`)
+
+```ruby
+results = redis.pipelined(raise_exception: false) do |pipeline|
+  pipeline.call("SET", "foo", "bar") # => nil
+  pipeline.call("DOESNOTEXIST", 12) # => raises RedisClient::CommandError
+  pipeline.call("INCR", "baz") # => nil
+end
+# results => ["OK", #<RedisClient::CommandError: ERR unknown command 'DOESNOTEXIST', with args beginning with: '12'>, 2]
+
+results.each do |result|
+  if result.is_a?(RedisClient::CommandError)
+    # Do something with the failed result
+  end
+end
+```
+
 ### Transactions
 
 You can use [`MULTI/EXEC` to run a number of commands in an atomic fashion](https://redis.io/topics/transactions).

lib/redis_client.rb

@@ -421,8 +421,8 @@ def disable_reconnection(&block)
     ensure_connected(retryable: false, &block)
   end
 
-  def pipelined
-    pipeline = Pipeline.new(@command_builder)
+  def pipelined(raise_exception: true)
+    pipeline = Pipeline.new(@command_builder, raise_exception: raise_exception)
     yield pipeline
 
     if pipeline._size == 0
@@ -431,7 +431,7 @@ def pipelined
       results = ensure_connected(retryable: pipeline._retryable?) do |connection|
         commands = pipeline._commands
         @middlewares.call_pipelined(commands, config) do
-          connection.call_pipelined(commands, pipeline._timeouts)
+          connection.call_pipelined(commands, pipeline._timeouts, pipeline._raise_exception)
         end
       end
 
@@ -579,6 +579,10 @@ def _timeouts
       nil
     end
 
+    def _raise_exception
+      true
+    end
+
     def _retryable?
       @retryable
     end
@@ -600,9 +604,10 @@ def _coerce!(results)
   end
 
   class Pipeline < Multi
-    def initialize(_command_builder)
-      super
+    def initialize(_command_builder, raise_exception: true)
+      super(_command_builder)
       @timeouts = nil
+      @raise_exception = raise_exception
     end
 
     def blocking_call(timeout, *command, **kwargs, &block)
@@ -627,6 +632,10 @@ def _timeouts
       @timeouts
     end
 
+    def _raise_exception
+      @raise_exception
+    end
+
     def _empty?
       @commands.empty?
     end

lib/redis_client/connection_mixin.rb

@@ -39,7 +39,7 @@ def call(command, timeout)
       end
     end
 
-    def call_pipelined(commands, timeouts)
+    def call_pipelined(commands, timeouts, raise_exception = true)
       exception = nil
 
       size = commands.size
@@ -67,7 +67,7 @@ def call_pipelined(commands, timeouts)
         results[index] = result
       end
 
-      if exception
+      if exception && raise_exception
         raise exception
       else
         results

lib/redis_client/decorator.rb

@@ -47,8 +47,8 @@ def with(*args)
       end
       ruby2_keywords :with if respond_to?(:ruby2_keywords, true)
 
-      def pipelined
-        @client.pipelined { |p| yield @_pipeline_class.new(p) }
+      def pipelined(**kwargs)
+        @client.pipelined(**kwargs) { |p| yield @_pipeline_class.new(p) }
       end
 
       def multi(**kwargs)

test/shared/redis_client_tests.rb

@@ -251,6 +251,33 @@ def test_pipelining_error
     assert_equal "42", @redis.call("GET", "foo")
   end
 
+  def test_pipelining_error_with_explicit_raising_exception
+    error = assert_raises RedisClient::CommandError do
+      @redis.pipelined(raise_exception: true) do |pipeline|
+        pipeline.call("DOESNOTEXIST", 12)
+        pipeline.call("SET", "foo", "42")
+      end
+    end
+
+    assert_equal ["DOESNOTEXIST", "12"], error.command
+
+    assert_equal "42", @redis.call("GET", "foo")
+  end
+
+  def test_pipelining_error_without_raising_exception
+    result = @redis.pipelined(raise_exception: false) do |pipeline|
+      pipeline.call("DOESNOTEXIST", 12)
+      pipeline.call("SET", "foo", "42")
+    end
+
+    assert result[0].is_a?(RedisClient::CommandError)
+    assert_equal ["DOESNOTEXIST", "12"], result[0].command
+
+    assert_equal "OK", result[1]
+
+    assert_equal "42", @redis.call("GET", "foo")
+  end
+
   def test_multi_error
     error = assert_raises RedisClient::CommandError do
       @redis.multi do |pipeline|