# encoding: UTF-8 # -- # Copyright (C) 2008-2011 10gen Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # ++ require 'sync' module Mongo # Instantiates and manages connections to a MongoDB replica set. class ReplSetConnection < Connection attr_reader :nodes, :secondaries, :arbiters, :secondary_pools, :replica_set_name, :read_pool, :seeds, :tags_to_pools, :refresh_interval # Create a connection to a MongoDB replica set. # # Once connected to a replica set, you can find out which nodes are primary, secondary, and # arbiters with the corresponding accessors: Connection#primary, Connection#secondaries, and # Connection#arbiters. This is useful if your application needs to connect manually to nodes other # than the primary. # # @param [Array] args A list of host-port pairs to be used as seed nodes followed by a # hash containing any options. See the examples below for exactly how to use the constructor. # # @option options [String] :rs_name (nil) The name of the replica set to connect to. You # can use this option to verify that you're connecting to the right replica set. # @option options [Boolean, Hash] :safe (false) Set the default safe-mode options # propogated to DB objects instantiated off of this Connection. This # default can be overridden upon instantiation of any DB by explicity setting a :safe value # on initialization. # @option options [Boolean] :read_secondary(false) If true, a random secondary node will be chosen, # and all reads will be directed to that node. # @option options [Logger, #debug] :logger (nil) Logger instance to receive driver operation log. # @option options [Integer] :pool_size (1) The maximum number of socket connections allowed per # connection pool. Note: this setting is relevant only for multi-threaded applications. # @option options [Float] :pool_timeout (5.0) When all of the connections a pool are checked out, # this is the number of seconds to wait for a new connection to be released before throwing an exception. # Note: this setting is relevant only for multi-threaded applications. # @option opts [Float] :op_timeout (nil) The number of seconds to wait for a read operation to time out. # Disabled by default. # @option opts [Float] :connect_timeout (nil) The number of seconds to wait before timing out a # connection attempt. # @option opts [Boolean] :ssl (false) If true, create the connection to the server using SSL. # @option opts [Boolean] :auto_refresh (false) Set this to true to enable a background thread that # periodically updates the state of the connection. If, for example, you initially connect while a secondary # is down, :auto_refresh will reconnect to that secondary behind the scenes to # prevent you from having to reconnect manually. # @option opts [Integer] :refresh_interval (90) If :auto_refresh is enabled, this is the number of seconds # that the background thread will sleep between calls to check the replica set's state. # # @example Connect to a replica set and provide two seed nodes. Note that the number of seed nodes does # not have to be equal to the number of replica set members. The purpose of seed nodes is to permit # the driver to find at least one replica set member even if a member is down. # ReplSetConnection.new(['localhost', 30000], ['localhost', 30001]) # # @example Connect to a replica set providing two seed nodes and ensuring a connection to the replica set named 'prod': # ReplSetConnection.new(['localhost', 30000], ['localhost', 30001], :rs_name => 'prod') # # @example Connect to a replica set providing two seed nodes and allowing reads from a secondary node: # ReplSetConnection.new(['localhost', 30000], ['localhost', 30001], :read_secondary => true) # # @see http://api.mongodb.org/ruby/current/file.REPLICA_SETS.html Replica sets in Ruby # # @raise [ReplicaSetConnectionError] This is raised if a replica set name is specified and the # driver fails to connect to a replica set with that name. def initialize(*args) extend Sync_m if args.last.is_a?(Hash) opts = args.pop else opts = {} end unless args.length > 0 raise MongoArgumentError, "A ReplSetConnection requires at least one seed node." end # The list of seed nodes @seeds = args # TODO: get rid of this @nodes = @seeds.dup # The members of the replica set, stored as instances of Mongo::Node. @members = [] # Connection pool for primary node @primary = nil @primary_pool = nil # Connection pools for each secondary node @secondaries = [] @secondary_pools = [] # The secondary pool to which we'll be sending reads. # This may be identical to the primary pool. @read_pool = nil # A list of arbiter addresses (for client information only) @arbiters = [] # Refresh @auto_refresh = opts.fetch(:auto_refresh, false) @refresh_interval = opts[:refresh_interval] || 90 # Are we allowing reads from secondaries? if opts[:read_secondary] warn ":read_secondary options has now been deprecated and will " + "be removed in driver v2.0. Use the :read option instead." @read_secondary = opts.fetch(:read_secondary, false) @read = :secondary else @read = opts.fetch(:read, :primary) end @connected = false # Store the refresher thread @refresh_thread = nil # Maps @sockets_to_pools = {} @tags_to_pools = {} # Replica set name if opts[:rs_name] warn ":rs_name option has been deprecated and will be removed in 2.0. " + "Please use :name instead." @replica_set_name = opts[:rs_name] else @replica_set_name = opts[:name] end setup(opts) end def inspect "" end # Initiate a connection to the replica set. def connect sync_synchronize(:EX) do return if @connected manager = PoolManager.new(self, @seeds) manager.connect update_config(manager) initiate_auto_refresh if @primary.nil? #TODO: in v2.0, we'll let this be optional and do a lazy connect. raise ConnectionFailure, "Failed to connect to primary node." else @connected = true end end end # Note: this method must be called from within # an exclusive lock. def update_config(manager) @arbiters = manager.arbiters.nil? ? [] : manager.arbiters.dup @primary = manager.primary.nil? ? nil : manager.primary.dup @secondaries = manager.secondaries.dup @hosts = manager.hosts.dup @primary_pool = manager.primary_pool @read_pool = manager.read_pool @secondary_pools = manager.secondary_pools @tags_to_pools = manager.tags_to_pools @seeds = manager.seeds @manager = manager @nodes = manager.nodes @max_bson_size = manager.max_bson_size end # If ismaster doesn't match our current view # then create a new PoolManager, passing in our # existing view. It should be able to do the diff. # Then take out the connection lock and replace # our current values. def refresh return if !connected? if !Thread.current[:background] Thread.current[:background] = PoolManager.new(self, @seeds) end background_manager = Thread.current[:background] if background_manager.update_required?(@hosts) sync_synchronize(:EX) do background_manager.connect update_config(background_manager) end end end def connected? sync_synchronize(:SH) do @connected end end # @deprecated def connecting? false end # The replica set primary's host name. # # @return [String] def host super end # The replica set primary's port. # # @return [Integer] def port super end def nodes warn "DEPRECATED" @seeds end # Determine whether we're reading from a primary node. If false, # this connection connects to a secondary node and @read_secondaries is true. # # @return [Boolean] def read_primary? sync_synchronize(:SH) do @read_pool == @primary_pool end end alias :primary? :read_primary? def read_preference @read end # Close the connection to the database. # TODO: we should get an exclusive lock here. def close @connected = false super if @refresh_thread @refresh_thread.kill @refresh_thread = nil end if @nodes @nodes.each do |member| member.disconnect end end @nodes = [] @read_pool = nil if @secondary_pools @secondary_pools.each do |pool| pool.close end end @secondaries = [] @secondary_pools = [] @arbiters = [] @tags_to_pools.clear @sockets_to_pools.clear end # If a ConnectionFailure is raised, this method will be called # to close the connection and reset connection values. # @deprecated def reset_connection close warn "ReplSetConnection#reset_connection is now deprecated. " + "Use ReplSetConnection#close instead." end # Returns +true+ if it's okay to read from a secondary node. # Since this is a replica set, this must always be true. # # This method exist primarily so that Cursor objects will # generate query messages with a slaveOkay value of +true+. # # @return [Boolean] +true+ def slave_ok? true end def authenticate_pools super @secondary_pools.each do |pool| pool.authenticate_existing end end def logout_pools(db) super @secondary_pools.each do |pool| pool.logout_existing(db) end end private def initiate_auto_refresh return unless @auto_refresh return if @refresh_thread && @refresh_thread.alive? @refresh_thread = Thread.new do while true do sleep(@refresh_interval) refresh end end end # Checkout a socket for reading (i.e., a secondary node). # Note that @read_pool might point to the primary pool # if no read pool has been defined. def checkout_reader connect unless connected? sync_synchronize(:SH) do socket = @read_pool.checkout @sockets_to_pools[socket] = @read_pool socket end end # Checkout a socket connected to a node with one of # the provided tags. If no such node exists, raise # an exception. def checkout_tagged(tags) sync_synchronize(:SH) do tags.each do |k, v| pool = @tags_to_pools[{k.to_s => v}] if pool socket = pool.checkout @sockets_to_pools[socket] = pool return socket end end end raise NodeWithTagsNotFound, "Could not find a connection tagged with #{tags}." end # Checkout a socket for writing (i.e., a primary node). def checkout_writer connect unless connected? sync_synchronize(:SH) do if @primary_pool socket = @primary_pool.checkout @sockets_to_pools[socket] = @primary_pool socket end end end # Checkin a socket used for reading. def checkin_reader(socket) warn "ReplSetConnection#checkin_writer is not deprecated and will be remove " + "in driver v2.0. Use ReplSetConnection#checkin instead." checkin(socket) end # Checkin a socket used for writing. def checkin_writer(socket) warn "ReplSetConnection#checkin_writer is not deprecated and will be remove " + "in driver v2.0. Use ReplSetConnection#checkin instead." checkin(socket) end def checkin(socket) if pool = @sockets_to_pools[socket] pool.checkin(socket) end end end end