Add some test cases (#21)

Author: supercaracal

lib/redis_client/cluster.rb

@@ -1,5 +1,6 @@
 # frozen_string_literal: true
 
+require 'redis_client'
 require 'redis_client/cluster/command'
 require 'redis_client/cluster/errors'
 require 'redis_client/cluster/key_slot_converter'
@@ -8,6 +9,62 @@
 
 class RedisClient
   class Cluster
+    class Pipeline
+      ReplySizeError = Class.new(::RedisClient::Error)
+
+      def initialize(client)
+        @client = client
+        @grouped = Hash.new([].freeze)
+        @replies = []
+        @size = 0
+      end
+
+      def call(*command, **kwargs)
+        node_key = @client.send(:find_node_key, command, primary_only: true)
+        @grouped[node_key] += [[@size, :call, command, kwargs]]
+        @size += 1
+      end
+
+      def call_once(*command, **kwargs)
+        node_key = @client.send(:find_node_key, command, primary_only: true)
+        @grouped[node_key] += [[@size, :call_once, command, kwargs]]
+        @size += 1
+      end
+
+      def blocking_call(timeout, *command, **kwargs)
+        node_key = @client.send(:find_node_key, command, primary_only: true)
+        @grouped[node_key] += [[@size, :blocking_call, timeout, command, kwargs]]
+        @size += 1
+      end
+
+      def empty?
+        @size.zero?
+      end
+
+      # TODO: use concurrency
+      def execute # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/MethodLength
+        @grouped.each do |node_key, rows|
+          node_key = node_key.nil? ? @client.instance_variable_get(:@node).primary_node_keys.sample : node_key
+          replies = @client.send(:find_node, node_key).pipelined do |pipeline|
+            rows.each do |row|
+              case row[1]
+              when :call then pipeline.call(*row[2], **row[3])
+              when :call_once then pipeline.call_once(*row[2], **row[3])
+              when :blocking_call then pipeline.blocking_call(row[2], *row[3], **row[4])
+              else raise NotImplementedError, row[1]
+              end
+            end
+          end
+
+          raise ReplySizeError, "commands: #{rows.size}, replies: #{replies.size}" if rows.size != replies.size
+
+          rows.each_with_index { |row, idx| @replies[row.first] = replies[idx] }
+        end
+
+        @replies
+      end
+    end
+
     ZERO_CURSOR_FOR_SCAN = '0'
 
     def initialize(config, pool: nil, **kwargs)
@@ -22,17 +79,17 @@ def inspect
       "#<#{self.class.name} #{@node.node_keys.join(', ')}>"
     end
 
-    def call(*command, **kwargs, &block)
-      send_command(:call, *command, **kwargs, &block)
+    def call(*command, **kwargs)
+      send_command(:call, *command, **kwargs)
     end
 
-    def call_once(*command, **kwargs, &block)
-      send_command(:call_once, *command, **kwargs, &block)
+    def call_once(*command, **kwargs)
+      send_command(:call_once, *command, **kwargs)
     end
 
-    def blocking_call(timeout, *command, **kwargs, &block)
+    def blocking_call(timeout, *command, **kwargs)
       node = assign_node(*command)
-      try_send(node, :blocking_call, timeout, *command, **kwargs, &block)
+      try_send(node, :blocking_call, timeout, *command, **kwargs)
     end
 
     def scan(*args, **kwargs, &block)
@@ -61,14 +118,22 @@ def zscan(key, *args, **kwargs, &block)
       try_send(node, :zscan, key, *args, **kwargs, &block)
     end
 
-    def pipelined
+    def mset
       # TODO: impl
     end
 
-    def multi
+    def mget
       # TODO: impl
     end
 
+    def pipelined
+      pipeline = ::RedisClient::Cluster::Pipeline.new(self)
+      yield pipeline
+      return [] if pipeline.empty? == 0
+
+      pipeline.execute
+    end
+
     def pubsub
       # TODO: impl
     end
@@ -82,16 +147,9 @@ def with(options = {})
     end
     alias then with
 
-    def id
-      @node.flat_map(&:id).sort.join(' ')
-    end
-
-    def connected?
-      @node.any?(&:connected?)
-    end
-
     def close
       @node.each(&:close)
+      nil
     end
 
     private

lib/redis_client/cluster/errors.rb

@@ -51,13 +51,5 @@ def initialize(command)
         super("Cluster client doesn't know which node the #{command} command should be sent to.")
       end
     end
-
-    # Raised when commands in pipelining include cross slot keys.
-    class CrossSlotPipeliningError < ::RedisClient::Error
-      def initialize(keys)
-        super("Cluster client couldn't send pipelining to single node. "\
-              "The commands include cross slot keys: #{ERR_ARG_NORMALIZATION.call(keys).join(',')}")
-      end
-    end
   end
 end

lib/redis_client/cluster/node.rb

@@ -95,6 +95,10 @@ def node_keys
         @clients.keys.sort
       end
 
+      def primary_node_keys
+        @clients.filter_map { |k, _| primary?(k) ? k : nil }.sort
+      end
+
       def find_by(node_key)
         @clients.fetch(node_key)
       rescue KeyError
@@ -133,7 +137,9 @@ def scale_reading_clients
           replica_disabled? ? primary?(node_key) : replica?(node_key)
         end
 
-        clients.values.sort_by { |client| "#{client.config.host}:#{client.config.port}" }
+        clients.values.sort_by do |client|
+          ::RedisClient::Cluster::NodeKey.build_from_host_port(client.config.host, client.config.port)
+        end
       end
 
       def slot_exists?(slot)

test/redis_client/cluster/test_errors.rb

@@ -72,18 +72,6 @@ def test_ambiguous_node_error
           assert_equal(e.message, c[:want], "Case: #{idx}")
         end
       end
-
-      def test_cross_slot_pipelining_error
-        [
-          { keys: %w[foo bar baz], want: 'keys: foo,bar,baz' },
-          { keys: '', want: 'keys: ' },
-          { keys: nil, want: 'keys: ' }
-        ].each_with_index do |c, idx|
-          raise ::RedisClient::Cluster::CrossSlotPipeliningError, c[:keys]
-        rescue StandardError => e
-          assert(e.message.end_with?(c[:want]), "Case: #{idx}")
-        end
-      end
     end
   end
 end

test/redis_client/cluster/test_node.rb

@@ -169,6 +169,12 @@ def test_node_keys
         end
       end
 
+      def test_primary_node_keys
+        want = @test_node_info.filter_map { |info| info[:role] == 'master' ? info[:node_key] : nil }.sort
+        got = @test_node.primary_node_keys
+        assert_equal(want, got)
+      end
+
       def test_find_by
         @test_node_info.each do |info|
           msg = "Case: primary only: #{info[:node_key]}"

test/redis_client/test_cluster.rb

@@ -12,16 +12,18 @@ module Mixin
       def setup
         @client = new_test_client
         @client.call('FLUSHDB')
-        testing_hook
+        wait_for_replication
       end
 
       def teardown
         @client.call('FLUSHDB')
-        testing_hook
+        wait_for_replication
         @client&.close
       end
 
-      def testing_hook; end
+      def wait_for_replication
+        @client.call('WAIT', '1', (TEST_TIMEOUT_SEC * 1000).to_i.to_s)
+      end
 
       def test_inspect
         assert_match(/^#<RedisClient::Cluster [0-9., :]*>$/, @client.inspect)
@@ -30,20 +32,23 @@ def test_inspect
       def test_call
         (0..9).each do |i|
           assert_equal('OK', @client.call('SET', "key#{i}", i), "Case: SET: key#{i}")
+          wait_for_replication
           assert_equal(i.to_s, @client.call('GET', "key#{i}"), "Case: GET: key#{i}")
         end
       end
 
       def test_call_once
         (0..9).each do |i|
           assert_equal('OK', @client.call_once('SET', "key#{i}", i), "Case: SET: key#{i}")
+          wait_for_replication
           assert_equal(i.to_s, @client.call_once('GET', "key#{i}"), "Case: GET: key#{i}")
         end
       end
 
       def test_blocking_call
         @client.call(*%w[RPUSH foo hello])
         @client.call(*%w[RPUSH foo world])
+        wait_for_replication
         client_side_timeout = 0.2
         server_side_timeout = 0.1
         assert_equal(%w[foo world], @client.blocking_call(client_side_timeout, 'BRPOP', 'foo', server_side_timeout), 'Case: 1st')
@@ -56,6 +61,7 @@ def test_scan
         assert_raises(ArgumentError) { @client.scan }
 
         (0..9).each { |i| @client.call('SET', "key#{i}", i) }
+        wait_for_replication
         want = (0..9).map { |i| "key#{i}" }
         got = []
         @client.scan('COUNT', '5') { |key| got << key }
@@ -65,6 +71,7 @@ def test_scan
       def test_sscan
         (0..9).each do |i|
           (0..9).each { |j| @client.call('SADD', "key#{i}", "member#{j}") }
+          wait_for_replication
           want = (0..9).map { |j| "member#{j}" }
           got = []
           @client.sscan("key#{i}", 'COUNT', '5') { |member| got << member }
@@ -75,6 +82,7 @@ def test_sscan
       def test_hscan
         (0..9).each do |i|
           (0..9).each { |j| @client.call('HSET', "key#{i}", "field#{j}", j) }
+          wait_for_replication
           want = (0..9).map { |j| "field#{j}" }
           got = []
           @client.hscan("key#{i}", 'COUNT', '5') { |field| got << field }
@@ -85,12 +93,30 @@ def test_hscan
       def test_zscan
         (0..9).each do |i|
           (0..9).each { |j| @client.call('ZADD', "key#{i}", j, "member#{j}") }
+          wait_for_replication
           want = (0..9).map { |j| "member#{j}" }
           got = []
           @client.zscan("key#{i}", 'COUNT', '5') { |member| got << member }
           assert_equal(want, got.sort)
         end
       end
+
+      def test_pipelined
+        want = (0..9).map { 'OK' } + (1..3).to_a + %w[PONG] + (0..9).map(&:to_s) + [%w[list 2]]
+        got = @client.pipelined do |pipeline|
+          (0..9).each { |i| pipeline.call('SET', "string#{i}", i) }
+          (0..2).each { |i| pipeline.call('RPUSH', 'list', i) }
+          pipeline.call_once('PING')
+          (0..9).each { |i| pipeline.call('GET', "string#{i}") }
+          pipeline.blocking_call(0.2, 'BRPOP', 'list', '0.1')
+        end
+
+        assert_equal(want, got)
+      end
+
+      def test_close
+        assert_nil(@client.close)
+      end
     end
 
     class PrimaryOnly < Minitest::Test
@@ -109,10 +135,6 @@ def new_test_client
         config = ::RedisClient::ClusterConfig.new(nodes: TEST_NODE_URIS, replica: true, **TEST_GENERIC_OPTIONS)
         ::RedisClient::Cluster.new(config)
       end
-
-      def testing_hook
-        @client.call('WAIT', '1', '300')
-      end
     end
 
     class Pooled < Minitest::Test