Raise errors from #pipelined instead of returning them (#297)

Author: KJTsanaktsidis

lib/redis_client/cluster/pipeline.rb

@@ -50,28 +50,31 @@ def get_block(inner_index)
       end
 
       ::RedisClient::ConnectionMixin.module_eval do
-        def call_pipelined_aware_of_redirection(commands, timeouts) # rubocop:disable Metrics/AbcSize
+        def call_pipelined_aware_of_redirection(commands, timeouts) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
           size = commands.size
           results = Array.new(commands.size)
           @pending_reads += size
           write_multi(commands)
 
           redirection_indices = nil
+          exception = nil
           size.times do |index|
             timeout = timeouts && timeouts[index]
             result = read(timeout)
             @pending_reads -= 1
-            if result.is_a?(CommandError)
+            if result.is_a?(::RedisClient::Error)
               result._set_command(commands[index])
-              if result.message.start_with?('MOVED', 'ASK')
+              if result.is_a?(::RedisClient::CommandError) && result.message.start_with?('MOVED', 'ASK')
                 redirection_indices ||= []
                 redirection_indices << index
+              else
+                exception ||= result
               end
             end
-
             results[index] = result
           end
 
+          raise exception if exception
           return results if redirection_indices.nil?
 
           err = ::RedisClient::Cluster::Pipeline::RedirectionNeeded.new
@@ -217,21 +220,15 @@ def handle_redirection(err, pipeline, inner_index)
 
         if err.message.start_with?('MOVED')
           node = @router.assign_redirection_node(err.message)
-          try_redirection(node, pipeline, inner_index)
+          redirect_command(node, pipeline, inner_index)
         elsif err.message.start_with?('ASK')
           node = @router.assign_asking_node(err.message)
-          try_asking(node) ? try_redirection(node, pipeline, inner_index) : err
+          try_asking(node) ? redirect_command(node, pipeline, inner_index) : err
         else
           err
         end
       end
 
-      def try_redirection(node, pipeline, inner_index)
-        redirect_command(node, pipeline, inner_index)
-      rescue StandardError => e
-        e
-      end
-
       def redirect_command(node, pipeline, inner_index)
         method = pipeline.get_callee_method(inner_index)
         command = pipeline.get_command(inner_index)

test/redis_client/test_cluster.rb

@@ -153,22 +153,16 @@ def test_pipelined
       end
 
       def test_pipelined_with_errors
-        got = @client.pipelined do |pipeline|
-          10.times do |i|
-            pipeline.call('SET', "string#{i}", i)
-            pipeline.call('SET', "string#{i}", i, 'too many args')
-            pipeline.call('SET', "string#{i}", i + 10)
+        assert_raises(RedisClient::Cluster::ErrorCollection) do
+          @client.pipelined do |pipeline|
+            10.times do |i|
+              pipeline.call('SET', "string#{i}", i)
+              pipeline.call('SET', "string#{i}", i, 'too many args')
+              pipeline.call('SET', "string#{i}", i + 10)
+            end
           end
         end
 
-        assert_equal(30, got.size)
-
-        10.times do |i|
-          assert_equal('OK', got[(3 * i) + 0])
-          assert_instance_of(::RedisClient::CommandError, got[(3 * i) + 1])
-          assert_equal('OK', got[(3 * i) + 2])
-        end
-
         wait_for_replication
 
         10.times { |i| assert_equal((i + 10).to_s, @client.call('GET', "string#{i}")) }