Refactor Sentinel support to make Redis::Client less aware of Sentinel.

Author: djanowski

examples/consistency.rb

@@ -107,8 +107,8 @@ def test
     end
 end
 
-Sentinels = [{:host => "127.0.0.1", :port => 26380},
-             {:host => "127.0.0.1", :port => 26381}]
-r = Redis.new(:url => "sentinel://mymaster", :sentinels => Sentinels, :role => :master)
+Sentinels = [{:host => "127.0.0.1", :port => 26379},
+             {:host => "127.0.0.1", :port => 26380}]
+r = Redis.new(:url => "redis://master1", :sentinels => Sentinels, :role => :master)
 tester = ConsistencyTester.new(r)
 tester.test

lib/redis/client.rb

@@ -75,6 +75,12 @@ def initialize(options = {})
       @logger = @options[:logger]
       @connection = nil
       @command_map = {}
+
+      if options.include?(:sentinels)
+        @connector = Connector::Sentinel.new(@options)
+      else
+        @connector = Connector.new(@options)
+      end
     end
 
     def connect
@@ -85,25 +91,7 @@ def connect
         establish_connection
         call [:auth, password] if password
         call [:select, db] if db != 0
-        if @options[:sentinels]
-          # Check the instance is really of the role we are looking for.
-          # We can't assume the command is supported since it was
-          # introduced recently and this client should work with old stuff.
-          role = nil
-          begin
-            role = call [:role]
-          rescue
-            # Assume the test is passed if we can't get a reply from ROLE...
-            role = [@options[:role].to_s]
-          end
-
-          # Raise an error on role mismatch. TODO: we could do better, wait
-          # some time and connect again...
-          if role[0] != @options[:role].to_s
-            disconnect
-            raise ConnectionError, "Instance role mismatch, try again."
-          end
-        end
+        @connector.check(self)
       end
 
       self
@@ -319,60 +307,8 @@ def logging(commands)
       end
     end
 
-    def set_addr_via_sentinel
-      responder = nil # The Sentinel that was able to reply
-
-      # Try one Sentinel after the other, using the list provided
-      # by the user.
-      @options[:sentinels].each{|sentinel|
-        begin
-          if !sentinel[:link]
-            sentinel[:link] = Redis.new(:host => sentinel[:host],
-                                        :port => sentinel[:port],
-                                        :timeout => 0.300)
-          end
-          if @options[:role] == :master
-            reply = sentinel[:link].client.call(["sentinel","get-master-addr-by-name",@options[:mastername]])
-            next if !reply
-            # Got it, set :host and :port
-            @options[:host] = reply[0]
-            @options[:port] = reply[1]
-            responder = sentinel
-            break
-          elsif @options[:role] == :slave
-            reply = sentinel[:link].client.call(["sentinel","slaves",@options[:mastername]])
-            slaves = []
-            reply.each{|slave|
-                slaves << Hash[*slave]
-            }
-            random_slave = slaves[rand(slaves.length)]
-            @options[:host] = random_slave['ip']
-            @options[:port] = random_slave['port']
-            responder = sentinel
-            break
-          else
-            raise ArgumentError, "Unknown instance role #{@options[:role]}"
-          end
-        rescue
-            next; # Try the next one on error
-        end
-      }
-
-      if responder
-        # If we were able to obtain the address, make sure to put the Sentinel
-        # that was able to reply as the first in the list.
-        @options[:sentinels].delete(responder)
-        @options[:sentinels].unshift(responder)
-      else
-        raise ConnectionError, "Unable to fetch #{@options[:role]} via Sentinel."
-      end
-    end
-
     def establish_connection
-      if @options[:sentinels]
-        set_addr_via_sentinel
-      end
-      @connection = @options[:driver].connect(@options.dup)
+      @connection = @options[:driver].connect(@connector.resolve.dup)
     rescue TimeoutError
       raise CannotConnectError, "Timed out connecting to Redis on #{location}"
     rescue Errno::ECONNREFUSED
@@ -444,12 +380,7 @@ def _parse_options(options)
           defaults[:port]     = uri.port if uri.port
           defaults[:password] = CGI.unescape(uri.password) if uri.password
           defaults[:db]       = uri.path[1..-1].to_i if uri.path
-        elsif uri.scheme == "sentinel"
-          defaults[:scheme]   = uri.scheme
-          defaults[:mastername] = uri.host
           defaults[:role] = :master
-          defaults[:password] = CGI.unescape(uri.password) if uri.password
-          defaults[:db]       = uri.path[1..-1].to_i if uri.path
         else
           raise ArgumentError, "invalid uri scheme '#{uri.scheme}'"
         end
@@ -465,13 +396,6 @@ def _parse_options(options)
         options[:scheme] = "unix"
         options.delete(:host)
         options.delete(:port)
-      elsif options[:mastername]
-        # Sentinel
-        options.delete(:host)
-        options.delete(:port)
-        if options[:sentinels].nil?
-          raise ArgumentError, "list of Sentinels required"
-        end
       else
         # TCP socket
         options[:host] = options[:host].to_s
@@ -519,5 +443,94 @@ def _parse_driver(driver)
 
       driver
     end
+
+    class Connector
+      def initialize(options)
+        @options = options
+      end
+
+      def resolve
+        @options
+      end
+
+      def check(client)
+      end
+
+      class Sentinel < Connector
+        def initialize(options)
+          super(options)
+
+          @sentinels = options.fetch(:sentinels).dup
+          @role = options[:role].to_s
+          @master = options[:host]
+        end
+
+        def check(client)
+          # Check the instance is really of the role we are looking for.
+          # We can't assume the command is supported since it was introduced
+          # recently and this client should work with old stuff.
+          begin
+            role = client.call([:role])[0]
+          rescue
+            # Assume the test is passed if we can't get a reply from ROLE...
+            role = @role
+          end
+
+          if role != @role
+            disconnect
+            raise ConnectionError, "Instance role mismatch. Expected #{@role}, got #{role}."
+          end
+        end
+
+        def resolve
+          result = case @role
+                   when "master"
+                     resolve_master
+                   when "slave"
+                     resolve_slave
+                   else
+                     raise ArgumentError, "Unknown instance role #{@role}"
+                   end
+
+          result || (raise ConnectionError, "Unable to fetch #{@role} via Sentinel.")
+        end
+
+        def sentinel_detect
+          @sentinels.each do |sentinel|
+            client = Client.new(:host => sentinel[:host], :port => sentinel[:port], :timeout => 0.3)
+
+            begin
+              if result = yield(client)
+                # This sentinel responded. Make sure we ask it first next time.
+                @sentinels.delete(sentinel)
+                @sentinels.unshift(sentinel)
+
+                return result
+              end
+            ensure
+              client.disconnect
+            end
+          end
+        end
+
+        def resolve_master
+          sentinel_detect do |client|
+            if reply = client.call(["sentinel", "get-master-addr-by-name", @master])
+              {host: reply[0], port: reply[1]}
+            end
+          end
+        end
+
+        def resolve_slave
+          sentinel_detect do |client|
+            if reply = client.call(["sentinel", "slaves", @master])
+              slave = Hash[*reply.sample]
+
+              {host: slave.fetch("ip"), port: slave.fetch("port")}
+            end
+          end
+        end
+      end
+    end
   end
 end