lib/redis_failover/client.rb in redis_failover-0.9.7.2 vs lib/redis_failover/client.rb in redis_failover-1.0.0

@@ -38,19 +38,21 @@
 
     # Creates a new failover redis client.
     #
     # @param [Hash] options the options used to initialize the client instance
     # @option options [String] :zkservers comma-separated ZooKeeper host:port
+    # @option options [String] :zk an existing ZK client connection instance
     # @option options [String] :znode_path znode path override for redis nodes
     # @option options [String] :password password for redis nodes
     # @option options [String] :db database to use for redis nodes
     # @option options [String] :namespace namespace for redis nodes
     # @option options [Logger] :logger logger override
     # @option options [Boolean] :retry_failure indicates if failures are retried
     # @option options [Integer] :max_retries max retries for a failure
     # @option options [Boolean] :safe_mode indicates if safe mode is used or not
     # @option options [Boolean] :master_only indicates if only redis master is used
+    # @note Use either :zkservers or :zk
     # @return [RedisFailover::Client]
     def initialize(options = {})
       Util.logger = options[:logger] if options[:logger]
       @master = nil
       @slaves = []
@@ -131,11 +133,11 @@
     #
     # @param [Hash] options the options used for manual failover
     # @option options [String] :host the host of the failover candidate
     # @option options [String] :port the port of the failover candidate
     def manual_failover(options = {})
-      ManualFailover.new(@zk, options).perform
+      ManualFailover.new(@zk, @root_znode, options).perform
       self
     end
 
     # Gracefully performs a shutdown of this client. This method is
     # mostly useful when the client is used in a forking environment.
@@ -175,17 +177,17 @@
 
     private
 
     # Sets up the underlying ZooKeeper connection.
     def setup_zk
-      @zk = ZK.new(@zkservers)
-      @zk.watcher.register(@znode) { |event| handle_zk_event(event) }
+      @zk = ZK.new(@zkservers) if @zkservers
+      @zk.register(redis_nodes_path) { |event| handle_zk_event(event) }
       if @safe_mode
         @zk.on_expired_session { purge_clients }
       end
-      @zk.on_connected { @zk.stat(@znode, :watch => true) }
-      @zk.stat(@znode, :watch => true)
+      @zk.on_connected { @zk.stat(redis_nodes_path, :watch => true) }
+      @zk.stat(redis_nodes_path, :watch => true)
       update_znode_timestamp
     end
 
     # Handles a ZK event.
     #
@@ -194,16 +196,16 @@
       update_znode_timestamp
       if event.node_created? || event.node_changed?
         build_clients
       elsif event.node_deleted?
         purge_clients
-        @zk.stat(@znode, :watch => true)
+        @zk.stat(redis_nodes_path, :watch => true)
       else
         logger.error("Unknown ZK node event: #{event.inspect}")
       end
     ensure
-      @zk.stat(@znode, :watch => true)
+      @zk.stat(redis_nodes_path, :watch => true)
     end
 
     # Determines if a method is a known redis operation.
     #
     # @param [Symbol] method the method to check
@@ -308,19 +310,23 @@
 
     # Fetches the known redis nodes from ZooKeeper.
     #
     # @return [Hash] the known master/slave redis servers
     def fetch_nodes
-      data = @zk.get(@znode, :watch => true).first
+      data = @zk.get(redis_nodes_path, :watch => true).first
       nodes = symbolize_keys(decode(data))
       logger.debug("Fetched nodes: #{nodes.inspect}")
 
       nodes
     rescue Zookeeper::Exceptions::InheritedConnectionError => ex
       logger.debug { "Caught #{ex.class} '#{ex.message}' - reopening ZK client" }
       @zk.reopen
       retry
+    rescue *ZK_ERRORS => ex
+      logger.warn { "Caught #{ex.class} '#{ex.message}' - retrying" }
+      sleep(RETRY_WAIT_TIME)
+      retry
     end
 
     # Builds new Redis clients for the specified nodes.
     #
     # @param [Array<String>] nodes the array of redis host:port pairs
@@ -473,17 +479,26 @@
 
     # Parses the configuration operations.
     #
     # @param [Hash] options the configuration options
     def parse_options(options)
-      @zkservers = options.fetch(:zkservers) { raise ArgumentError, ':zkservers required'}
-      @znode = options.fetch(:znode_path, Util::DEFAULT_ZNODE_PATH)
+      @zk, @zkservers = options.values_at(:zk, :zkservers)
+      if [@zk, @zkservers].all? || [@zk, @zkservers].none?
+        raise ArgumentError, 'must specify :zk or :zkservers'
+      end
+
+      @root_znode = options.fetch(:znode_path, Util::DEFAULT_ROOT_ZNODE_PATH)
       @namespace = options[:namespace]
       @password = options[:password]
       @db = options[:db]
       @retry = options.fetch(:retry_failure, true)
       @max_retries = @retry ? options.fetch(:max_retries, 3) : 0
       @safe_mode = options.fetch(:safe_mode, true)
       @master_only = options.fetch(:master_only, false)
+    end
+
+    # @return [String] the znode path for the master redis nodes config
+    def redis_nodes_path
+      "#{@root_znode}/nodes"
     end
   end
 end