mongo-ruby-driver/lib/mongo/util/pool_manager.rb

module Mongo
  class PoolManager

    attr_reader :connection, :seeds, :arbiters, :primary, :secondaries,
      :primary_pool, :read_pool, :secondary_pools, :hosts

    def initialize(connection, seeds)
      @connection = connection
      @seeds = seeds
    end

    def connect
      initialize_data
      nodes = connect_to_members
      initialize_pools(nodes)
      update_seed_list(nodes)
    end

    # Ensure that the view of the replica set is current by
    # running the ismaster command and checking to see whether
    # we've connected to all known nodes. If not, automatically
    # connect to these unconnected nodes. This is handy when we've
    # connected to a replica set with no primary or when a secondary
    # node comes up after we've connected.
    #
    # If we're connected to nodes that are no longer part of the set,
    # remove these from our set of secondary pools.
    def update_required?(hosts)
      node = Thread.current[:refresher_node]
      if !node || !node.active?
        begin
          node = get_valid_seed_node
          Thread.current[:refresher_node] = node
        rescue ConnectionFailure
          warn "Could not refresh config because no valid seed node was unavailable."
        end
      end

      node_list = node.node_list

      unconnected_nodes = node_list - hosts
      removed_nodes = hosts - node_list

      if unconnected_nodes.empty? && removed_nodes.empty?
        return false
      else
        {:unconnected => unconnected_nodes, :removed => removed_nodes}
      end
    end

    def update(manager, node_struct)
      reference_manager_data(manager)
      unconnected_nodes = node_struct[:unconnected]
      removed_nodes = node_struct[:removed]

      if !removed_nodes.empty?
        removed_nodes.each do |node|
          if @primary_pool && @primary_pool.host_string == node
            @primary = nil
            @primary_pool.close
            @primary_pool = nil
          elsif rejected_pool = @secondary_pools.reject! {|pool| pool.host_string == node}
            @secondaries.reject! do |secondary|
              secondary.port == rejected_pool.port && secondary.host == rejected_pool.host
            end
          end
        end
      end

      if !unconnected_nodes.empty?
        nodes = []
        unconnected_nodes.each do |host_port|
          node = Mongo::Node.new(self.connection, host_port)
          if node.connect && node.set_config
            nodes << node
          end
        end

        if !nodes.empty?
          initialize_pools(nodes)
        end
      end
    end

    private

    # Note that @arbiters and @read_pool will be
    # assigned automatically.
    def reference_manager_data(manager)
      @primary = manager.primary
      @primary_pool = manager.primary_pool
      @secondaries = manager.secondaries
      @secondary_pools = manager.secondary_pools
      @read_pool = manager.read_pool
      @arbiters = manager.arbiters
      @hosts = manager.hosts
    end

    def initialize_data
      @primary = nil
      @primary_pool = nil
      @read_pool = nil
      @arbiters = []
      @secondaries = []
      @secondary_pools = []
      @hosts = []
    end

    def connected_nodes
      nodes = []
      if @primary_pool
        nodes << "#{@primary_pool.host}:#{@primary_pool.port}"
      end

      @secondary_pools.each do |pool|
        nodes << "#{pool.host}:#{pool.port}"
      end

      nodes
    end

    # Connect to each member of the replica set
    # as reported by the given seed node, and return
    # as a list of Mongo::Node objects.
    def connect_to_members
      nodes = []

      seed = get_valid_seed_node

      seed.node_list.each do |host|
        node = Mongo::Node.new(self.connection, host)
        if node.connect && node.set_config
          nodes << node
        end
      end

      if nodes.empty?
        raise ConnectionFailure, "Failed to connect to any given member."
      end

      nodes
    end

    # Initialize the connection pools for the primary and secondary nodes.
    def initialize_pools(nodes)
      nodes.each do |member|
        @hosts << member.host_string

        if member.primary?
          @primary = member.host_port
          @primary_pool = Pool.new(self.connection, member.host, member.port,
                                  :size => self.connection.pool_size,
                                  :timeout => self.connection.connect_timeout,
                                  :node => member)
        elsif member.secondary? && !@secondaries.include?(member.host_port)
          @secondaries << member.host_port
          @secondary_pools << Pool.new(self.connection, member.host, member.port,
                                       :size => self.connection.pool_size,
                                       :timeout => self.connection.connect_timeout,
                                       :node => member)
        end
      end

      @arbiters = nodes.first.arbiters
      choose_read_pool
    end

    # Pick a node from the set of possible secondaries.
    # If more than one node is available, use the ping
    # time to figure out which nodes to choose from.
    def choose_read_pool
      if @secondary_pools.empty?
        @read_pool = @primary_pool
      elsif @secondary_pools.size == 1
        @read_pool = @secondary_pools[0]
      else
        ping_ranges = Array.new(3) { |i| Array.new }
        @secondary_pools.each do |pool|
          case pool.ping_time
            when 0..150
              ping_ranges[0] << pool
            when 150..1000
              ping_ranges[1] << pool
            else
              ping_ranges[2] << pool
          end
        end

        for list in ping_ranges do
          break if !list.empty?
        end

        @read_pool = list[rand(list.length)]
      end
    end

    # Iterate through the list of provided seed
    # nodes until we've gotten a response from the
    # replica set we're trying to connect to.
    #
    # If we don't get a response, raise an exception.
    def get_valid_seed_node
      @seeds.each do |seed|
        node = Mongo::Node.new(self.connection, seed)
        if node.connect && node.set_config
          return node
        end
      end

      raise ConnectionFailure, "Cannot connect to a replica set using seeds " +
        "#{@seeds.map {|s| "#{s[0]}:#{s[1]}" }.join(', ')}"
    end

    def update_seed_list(nodes)
      @seeds = nodes.map { |n| n.host_port }
    end

  end
end