Add pubsub feature (#25)

Author: supercaracal

lib/redis_client/cluster.rb

@@ -65,6 +65,28 @@ def execute # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Met
       end
     end
 
+    class PubSub
+      def initialize(client)
+        @client = client
+        @pubsub = nil
+      end
+
+      def call(*command, **kwargs)
+        close
+        @pubsub = @client.send(:assign_node, *command).pubsub
+        @pubsub.call(*command, **kwargs)
+      end
+
+      def close
+        @pubsub&.close
+        @pubsub = nil
+      end
+
+      def next_event(timeout = nil)
+        @pubsub&.next_event(timeout)
+      end
+    end
+
     ZERO_CURSOR_FOR_SCAN = '0'
     CMD_SCAN = 'SCAN'
     CMD_SSCAN = 'SSCAN'
@@ -135,7 +157,7 @@ def pipelined
     end
 
     def pubsub
-      # TODO: impl
+      ::RedisClient::Cluster::PubSub.new(self)
     end
 
     def close

lib/redis_client/cluster/node.rb

@@ -129,11 +129,6 @@ def call_replica(method, *command, **kwargs, &block)
         end.values
       end
 
-      # TODO: impl
-      def process_all(commands, &block)
-        try_map { |_, client| client.process(commands, &block) }.values
-      end
-
       def scale_reading_clients
         clients = @clients.select do |node_key, _|
           replica_disabled? ? primary?(node_key) : replica?(node_key)

test/redis_client/test_cluster.rb

@@ -6,7 +6,7 @@
 
 class RedisClient
   class TestCluster
-    module Mixin
+    module Mixin # rubocop:disable Metrics/ModuleLength
       include TestingHelper
 
       def setup
@@ -116,6 +116,27 @@ def test_pipelined
         assert_equal(want, got)
       end
 
+      def test_pubsub
+        (0..9).each do |i|
+          pubsub = @client.pubsub
+          pubsub.call('SUBSCRIBE', "channel#{i}")
+          assert_equal(['subscribe', "channel#{i}", 1], pubsub.next_event(0.1))
+        end
+
+        sub = Fiber.new do |client|
+          channel = 'my-channel'
+          pubsub = client.pubsub
+          pubsub.call('SUBSCRIBE', channel)
+          assert_equal(['subscribe', channel, 1], pubsub.next_event(TEST_TIMEOUT_SEC))
+          Fiber.yield(channel)
+          Fiber.yield(pubsub.next_event(TEST_TIMEOUT_SEC))
+        end
+
+        channel = sub.resume(@client)
+        @client.call('PUBLISH', channel, 'hello world')
+        assert_equal(['message', channel, 'hello world'], sub.resume)
+      end
+
       def test_close
         assert_nil(@client.close)
       end