Keep compatibilities as possible with a standalone client for the transaction feature in a cluster client

Author: Taishi Kasuga

cluster/lib/redis/cluster.rb

@@ -97,7 +97,7 @@ def cluster(subcommand, *args)
     end
 
     def watch(*keys, &block)
-      synchronize { |c| c.call_v([:watch] + keys, &block) }
+      synchronize { |c| c.watch(*keys, &block) }
     end
 
     private

cluster/lib/redis/cluster/client.rb

@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 
 require 'redis-cluster-client'
+require 'redis/cluster/transaction_adapter'
 
 class Redis
   class Cluster
@@ -98,6 +99,22 @@ def multi(watch: nil, &block)
         handle_errors { super(watch: watch, &block) }
       end
 
+      def watch(*keys)
+        unless block_given?
+          raise Redis::Cluster::TransactionConsistencyError, 'A block is required if you use the cluster client.'
+        end
+
+        handle_errors do
+          RedisClient::Cluster::OptimisticLocking.new(@router).watch(keys) do |c, slot, asking|
+            transaction = Redis::Cluster::TransactionAdapter.new(
+              self, @router, @command_builder, node: c, slot: slot, asking: asking
+            )
+            yield transaction
+            transaction.execute
+          end
+        end
+      end
+
       private
 
       def handle_errors

cluster/lib/redis/cluster/transaction_adapter.rb

@@ -0,0 +1,49 @@
+# frozen_string_literal: true
+
+require 'redis-cluster-client'
+require 'redis_client/cluster/transaction'
+
+class Redis
+  class Cluster
+    class TransactionAdapter < RedisClient::Cluster::Transaction
+      def initialize(client, router, command_builder, node: nil, slot: nil, asking: false)
+        @client = client
+        super(router, command_builder, node: node, slot: slot, asking: asking)
+      end
+
+      def multi
+        yield self
+      end
+
+      def exec
+        # no need to do nothing
+      end
+
+      def discard
+        # no need to do nothing
+      end
+
+      def watch(*_)
+        # no need to do nothing
+      end
+
+      def unwatch
+        # no need to do nothing
+      end
+
+      private
+
+      def method_missing(name, *args, **kwargs, &block)
+        return call(name, *args, **kwargs, &block) if @client.respond_to?(name)
+
+        super
+      end
+
+      def respond_to_missing?(name, include_private = false)
+        return true if @client.respond_to?(name)
+
+        super
+      end
+    end
+  end
+end

cluster/test/client_transactions_test.rb

@@ -7,20 +7,20 @@ class TestClusterClientTransactions < Minitest::Test
   include Helper::Cluster
 
   def test_cluster_client_does_support_transaction_by_single_key
-    actual = redis.multi do |r|
-      r.set('counter', '0')
-      r.incr('counter')
-      r.incr('counter')
+    actual = redis.multi do |tx|
+      tx.set('counter', '0')
+      tx.incr('counter')
+      tx.incr('counter')
     end
 
     assert_equal(['OK', 1, 2], actual)
     assert_equal('2', redis.get('counter'))
   end
 
   def test_cluster_client_does_support_transaction_by_hashtag
-    actual = redis.multi do |r|
-      r.mset('{key}1', 1, '{key}2', 2)
-      r.mset('{key}3', 3, '{key}4', 4)
+    actual = redis.multi do |tx|
+      tx.mset('{key}1', 1, '{key}2', 2)
+      tx.mset('{key}3', 3, '{key}4', 4)
     end
 
     assert_equal(%w[OK OK], actual)
@@ -29,18 +29,18 @@ def test_cluster_client_does_support_transaction_by_hashtag
 
   def test_cluster_client_does_not_support_transaction_by_multiple_keys
     assert_raises(Redis::Cluster::TransactionConsistencyError) do
-      redis.multi do |r|
-        r.set('key1', 1)
-        r.set('key2', 2)
-        r.set('key3', 3)
-        r.set('key4', 4)
+      redis.multi do |tx|
+        tx.set('key1', 1)
+        tx.set('key2', 2)
+        tx.set('key3', 3)
+        tx.set('key4', 4)
       end
     end
 
     assert_raises(Redis::Cluster::TransactionConsistencyError) do
-      redis.multi do |r|
-        r.mset('key1', 1, 'key2', 2)
-        r.mset('key3', 3, 'key4', 4)
+      redis.multi do |tx|
+        tx.mset('key1', 1, 'key2', 2)
+        tx.mset('key3', 3, 'key4', 4)
       end
     end
 
@@ -63,10 +63,50 @@ def test_cluster_client_does_support_transaction_with_optimistic_locking
       another.resume
       v1 = redis.get('{key}1')
       v2 = redis.get('{key}2')
-      tx.call('SET', '{key}1', v2)
-      tx.call('SET', '{key}2', v1)
+      tx.set('{key}1', v2)
+      tx.set('{key}2', v1)
     end
 
     assert_equal %w[3 4], redis.mget('{key}1', '{key}2')
   end
+
+  def test_cluster_client_can_be_used_compatible_with_standalone_client
+    redis.set('{my}key', 'value')
+    redis.set('{my}counter', '0')
+    redis.watch('{my}key', '{my}counter') do |client|
+      if redis.get('{my}key') == 'value'
+        client.multi do |tx|
+          tx.set('{my}key', 'updated value')
+          tx.incr('{my}counter')
+        end
+      else
+        client.unwatch
+      end
+    end
+
+    assert_equal('updated value', redis.get('{my}key'))
+    assert_equal('1', redis.get('{my}counter'))
+
+    another = Fiber.new do
+      cli = build_another_client
+      cli.set('{my}key', 'another value')
+      cli.close
+      Fiber.yield
+    end
+
+    redis.watch('{my}key', '{my}counter') do |client|
+      another.resume
+      if redis.get('{my}key') == 'value'
+        client.multi do |tx|
+          tx.set('{my}key', 'latest value')
+          tx.incr('{my}counter')
+        end
+      else
+        client.unwatch
+      end
+    end
+
+    assert_equal('another value', redis.get('{my}key'))
+    assert_equal('1', redis.get('{my}counter'))
+  end
 end

cluster/test/commands_on_transactions_test.rb

@@ -42,23 +42,23 @@ def test_watch
 
     assert_raises(Redis::Cluster::TransactionConsistencyError) do
       redis.watch('key1', 'key2') do |tx|
-        tx.call('SET', 'key1', '1')
-        tx.call('SET', 'key2', '2')
+        tx.set('key1', '1')
+        tx.set('key2', '2')
       end
     end
 
     assert_raises(Redis::Cluster::TransactionConsistencyError) do
       redis.watch('{hey}1', '{hey}2') do |tx|
-        tx.call('SET', '{key}1', '1')
-        tx.call('SET', '{key}2', '2')
+        tx.set('{key}1', '1')
+        tx.set('{key}2', '2')
       end
     end
 
     assert_empty(redis.watch('{key}1', '{key}2') {})
 
     redis.watch('{key}1', '{key}2') do |tx|
-      tx.call('SET', '{key}1', '1')
-      tx.call('SET', '{key}2', '2')
+      tx.set('{key}1', '1')
+      tx.set('{key}2', '2')
     end
 
     assert_equal %w[1 2], redis.mget('{key}1', '{key}2')