fix: delay transformation of reply data to prevent errors while converting (#190)

Author: supercaracal

lib/redis_client/cluster/router.rb

@@ -14,6 +14,7 @@ class Cluster
     class Router
       ZERO_CURSOR_FOR_SCAN = '0'
       METHODS_FOR_BLOCKING_CMD = %i[blocking_call_v blocking_call].freeze
+      TSF = ->(b, s) { b.nil? ? s : b.call(s) }.curry
 
       attr_reader :node
 
@@ -30,25 +31,25 @@ def initialize(config, pool: nil, **kwargs)
       def send_command(method, command, *args, &block) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
         cmd = ::RedisClient::Cluster::NormalizedCmdName.instance.get_by_command(command)
         case cmd
-        when 'acl', 'auth', 'bgrewriteaof', 'bgsave', 'quit', 'save'
-          @node.call_all(method, command, args, &block).first
-        when 'flushall', 'flushdb'
-          @node.call_primaries(method, command, args, &block).first
-        when 'ping'     then @node.send_ping(method, command, args, &block).first
+        when 'ping'     then @node.send_ping(method, command, args).first.then(&TSF.call(block))
         when 'wait'     then send_wait_command(method, command, args, &block)
-        when 'keys'     then @node.call_replicas(method, command, args, &block).flatten.sort_by(&:to_s)
-        when 'dbsize'   then @node.call_replicas(method, command, args, &block).select { |e| e.is_a?(Integer) }.sum
+        when 'keys'     then @node.call_replicas(method, command, args).flatten.sort_by(&:to_s).then(&TSF.call(block))
+        when 'dbsize'   then @node.call_replicas(method, command, args).select { |e| e.is_a?(Integer) }.sum.then(&TSF.call(block))
         when 'scan'     then scan(command, seed: 1)
-        when 'lastsave' then @node.call_all(method, command, args, &block).sort_by(&:to_i)
+        when 'lastsave' then @node.call_all(method, command, args).sort_by(&:to_i).then(&TSF.call(block))
         when 'role'     then @node.call_all(method, command, args, &block)
         when 'config'   then send_config_command(method, command, args, &block)
         when 'client'   then send_client_command(method, command, args, &block)
         when 'cluster'  then send_cluster_command(method, command, args, &block)
-        when 'readonly', 'readwrite', 'shutdown'
-          raise ::RedisClient::Cluster::OrchestrationCommandNotSupported, cmd
         when 'memory'   then send_memory_command(method, command, args, &block)
         when 'script'   then send_script_command(method, command, args, &block)
         when 'pubsub'   then send_pubsub_command(method, command, args, &block)
+        when 'acl', 'auth', 'bgrewriteaof', 'bgsave', 'quit', 'save'
+          @node.call_all(method, command, args).first.then(&TSF.call(block))
+        when 'flushall', 'flushdb'
+          @node.call_primaries(method, command, args).first.then(&TSF.call(block))
+        when 'readonly', 'readwrite', 'shutdown'
+          raise ::RedisClient::Cluster::OrchestrationCommandNotSupported, cmd
         when 'discard', 'exec', 'multi', 'unwatch'
           raise ::RedisClient::Cluster::AmbiguousNodeError, cmd
         else
@@ -208,7 +209,7 @@ def assign_asking_node(err_msg)
       private
 
       def send_wait_command(method, command, args, retry_count: 3, &block) # rubocop:disable Metrics/AbcSize
-        @node.call_primaries(method, command, args, &block).select { |r| r.is_a?(Integer) }.sum
+        @node.call_primaries(method, command, args).select { |r| r.is_a?(Integer) }.sum.then(&TSF.call(block))
       rescue ::RedisClient::Cluster::ErrorCollection => e
         raise if e.errors.any?(::RedisClient::CircuitBreaker::OpenCircuitError)
         raise if retry_count <= 0
@@ -224,24 +225,24 @@ def send_wait_command(method, command, args, retry_count: 3, &block) # rubocop:d
       def send_config_command(method, command, args, &block)
         case ::RedisClient::Cluster::NormalizedCmdName.instance.get_by_subcommand(command)
         when 'resetstat', 'rewrite', 'set'
-          @node.call_all(method, command, args, &block).first
+          @node.call_all(method, command, args).first.then(&TSF.call(block))
         else assign_node(command).public_send(method, *args, command, &block)
         end
       end
 
       def send_memory_command(method, command, args, &block)
         case ::RedisClient::Cluster::NormalizedCmdName.instance.get_by_subcommand(command)
         when 'stats' then @node.call_all(method, command, args, &block)
-        when 'purge' then @node.call_all(method, command, args, &block).first
+        when 'purge' then @node.call_all(method, command, args).first.then(&TSF.call(block))
         else assign_node(command).public_send(method, *args, command, &block)
         end
       end
 
       def send_client_command(method, command, args, &block)
         case ::RedisClient::Cluster::NormalizedCmdName.instance.get_by_subcommand(command)
-        when 'list' then @node.call_all(method, command, args, &block).flatten
+        when 'list' then @node.call_all(method, command, args).flatten.then(&TSF.call(block))
         when 'pause', 'reply', 'setname'
-          @node.call_all(method, command, args, &block).first
+          @node.call_all(method, command, args).first.then(&TSF.call(block))
         else assign_node(command).public_send(method, *args, command, &block)
         end
       end
@@ -251,7 +252,7 @@ def send_cluster_command(method, command, args, &block)
         when 'addslots', 'delslots', 'failover', 'forget', 'meet', 'replicate',
              'reset', 'set-config-epoch', 'setslot'
           raise ::RedisClient::Cluster::OrchestrationCommandNotSupported, ['cluster', subcommand]
-        when 'saveconfig' then @node.call_all(method, command, args, &block).first
+        when 'saveconfig' then @node.call_all(method, command, args).first.then(&TSF.call(block))
         when 'getkeysinslot'
           raise ArgumentError, command.join(' ') if command.size != 4
 
@@ -260,25 +261,25 @@ def send_cluster_command(method, command, args, &block)
         end
       end
 
-      def send_script_command(method, command, args, &block)
+      def send_script_command(method, command, args, &block) # rubocop:disable Metrics/AbcSize
         case ::RedisClient::Cluster::NormalizedCmdName.instance.get_by_subcommand(command)
         when 'debug', 'kill'
-          @node.call_all(method, command, args, &block).first
+          @node.call_all(method, command, args).first.then(&TSF.call(block))
         when 'flush', 'load'
-          @node.call_primaries(method, command, args, &block).first
+          @node.call_primaries(method, command, args).first.then(&TSF.call(block))
         when 'exists'
-          @node.call_all(method, command, args, &block).transpose.map { |arr| arr.any?(&:zero?) ? 0 : 1 }
+          @node.call_all(method, command, args).transpose.map { |arr| arr.any?(&:zero?) ? 0 : 1 }.then(&TSF.call(block))
         else assign_node(command).public_send(method, *args, command, &block)
         end
       end
 
       def send_pubsub_command(method, command, args, &block) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
         case ::RedisClient::Cluster::NormalizedCmdName.instance.get_by_subcommand(command)
-        when 'channels' then @node.call_all(method, command, args, &block).flatten.uniq.sort_by(&:to_s)
+        when 'channels' then @node.call_all(method, command, args).flatten.uniq.sort_by(&:to_s).then(&TSF.call(block))
         when 'numsub'
-          @node.call_all(method, command, args, &block).reject(&:empty?).map { |e| Hash[*e] }
-               .reduce({}) { |a, e| a.merge(e) { |_, v1, v2| v1 + v2 } }
-        when 'numpat' then @node.call_all(method, command, args, &block).select { |e| e.is_a?(Integer) }.sum
+          @node.call_all(method, command, args).reject(&:empty?).map { |e| Hash[*e] }
+               .reduce({}) { |a, e| a.merge(e) { |_, v1, v2| v1 + v2 } }.then(&TSF.call(block))
+        when 'numpat' then @node.call_all(method, command, args).select { |e| e.is_a?(Integer) }.sum.then(&TSF.call(block))
         else assign_node(command).public_send(method, *args, command, &block)
         end
       end

test/redis_client/test_cluster.rb

@@ -254,6 +254,16 @@ def test_dedicated_commands # rubocop:disable Metrics/CyclomaticComplexity, Metr
           { command: %w[SCRIPT FLUSH], want: 'OK' },
           { command: %w[SCRIPT EXISTS b5bb9d8014a0f9b1d61e21e796d78dccdf1352f23cd32812f4850b878ae4944c], want: [0] },
           { command: %w[SCRIPT EXISTS 5b9fb3410653a731f8ddfeff39a0c061 31b6de18e43fe980ed07d8b0f5a8cabe], want: [0, 0] },
+          {
+            command: %w[SCRIPT EXISTS b5bb9d8014a0f9b1d61e21e796d78dccdf1352f23cd32812f4850b878ae4944c],
+            blk: ->(reply) { reply.map { |r| !r.zero? } },
+            want: [false]
+          },
+          {
+            command: %w[SCRIPT EXISTS 5b9fb3410653a731f8ddfeff39a0c061 31b6de18e43fe980ed07d8b0f5a8cabe],
+            blk: ->(reply) { reply.map { |r| !r.zero? } },
+            want: [false, false]
+          },
           { command: %w[PUBSUB CHANNELS test-channel*], want: [] },
           { command: %w[PUBSUB NUMSUB test-channel], want: { 'test-channel' => 0 } },
           { command: %w[PUBSUB NUMPAT], want: 0 },
@@ -264,7 +274,7 @@ def test_dedicated_commands # rubocop:disable Metrics/CyclomaticComplexity, Metr
           next if c.key?(:supported_redis_version) && TEST_REDIS_MAJOR_VERSION < c[:supported_redis_version]
 
           msg = "Case: #{c[:command].join(' ')}"
-          got = -> { @client.call(*c[:command]) }
+          got = -> { @client.call_v(c[:command], &c[:blk]) }
           if c.key?(:error)
             assert_raises(c[:error], msg, &got)
           elsif c.key?(:is_a)