Class | Mongo::Connection |
In: |
lib/mongo/connection.rb
|
Parent: | Object |
Instantiates and manages connections to MongoDB.
TCPSocket | = | ::TCPSocket |
Mutex | = | ::Mutex |
ConditionVariable | = | ::ConditionVariable |
DEFAULT_PORT | = | 27017 |
STANDARD_HEADER_SIZE | = | 16 |
RESPONSE_HEADER_SIZE | = | 20 |
BINARY_ENCODING | = | Encoding.find("binary") |
auths | [R] | |
connect_timeout | [R] | |
host_to_try | [R] | |
logger | [R] | |
pool_size | [R] | |
primary | [R] | |
primary_pool | [R] | |
safe | [R] | |
size | [R] | |
socket_class | [R] |
Initialize a connection to MongoDB using the MongoDB URI spec:
@param uri [String]
A string of the format mongodb://[username:password@]host1[:port1][,host2[:port2],...[,hostN[:portN]]][/database]
@param opts Any of the options available for Connection.new
@return [Mongo::Connection, Mongo::ReplSetConnection]
# File lib/mongo/connection.rb, line 148 def self.from_uri(string, extra_opts={}) uri = URIParser.new(string) opts = uri.connection_options opts.merge!(extra_opts) if uri.nodes.length == 1 opts.merge!({:auths => uri.auths}) Connection.new(uri.nodes[0][0], uri.nodes[0][1], opts) elsif uri.nodes.length > 1 nodes = uri.nodes.clone nodes_with_opts = nodes << opts ReplSetConnection.new(*nodes_with_opts) else raise MongoArgumentError, "No nodes specified. Please ensure that you've provided at least one node." end end
DEPRECATED
Initialize a connection to a MongoDB replica set using an array of seed nodes.
The seed nodes specified will be used on the initial connection to the replica set, but note that this list of nodes will be replced by the list of canonical nodes returned by running the is_master command on the replica set.
@param nodes [Array] An array of arrays, each of which specifies a host and port. @param opts [Hash] Any of the available options that can be passed to Connection.new.
@option opts [String] :rs_name (nil) The name of the replica set to connect to. An exception will be
raised if unable to connect to a replica set with this name.
@option opts [Boolean] :read_secondary (false) When true, this connection object will pick a random slave
to send reads to.
@example
Connection.multi([["db1.example.com", 27017], ["db2.example.com", 27017]])
@example This connection will read from a random secondary node.
Connection.multi([["db1.example.com", 27017], ["db2.example.com", 27017], ["db3.example.com", 27017]], :read_secondary => true)
@return [Mongo::Connection]
@deprecated
# File lib/mongo/connection.rb, line 133 def self.multi(nodes, opts={}) warn "Connection.multi is now deprecated and will be removed in v2.0. Please use ReplSetConnection.new instead." nodes << opts ReplSetConnection.new(*nodes) end
Create a connection to single MongoDB instance.
You may specify whether connection to slave is permitted. In all cases, the default host is "localhost" and the default port is 27017.
If you‘re connecting to a replica set, you‘ll need to use ReplSetConnection.new instead.
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 [String, Hash] host. @param [Integer] port specify a port number here if only one host is being specified.
@option opts [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 opts [Boolean] :slave_ok (false) Must be set to true when connecting
to a single, slave node.
@option opts [Logger, debug] :logger (nil) A Logger instance for debugging driver ops. Note that
logging negatively impacts performance; therefore, it should not be used for high-performance apps.
@option opts [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 opts [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 (which in Ruby are rare).
@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.
@example localhost, 27017
Connection.new
@example localhost, 27017
Connection.new("localhost")
@example localhost, 3000, max 5 connections, with max 5 seconds of wait time.
Connection.new("localhost", 3000, :pool_size => 5, :timeout => 5)
@example localhost, 3000, where this node may be a slave
Connection.new("localhost", 3000, :slave_ok => true)
@see 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.
@core connections
# File lib/mongo/connection.rb, line 95 def initialize(host=nil, port=nil, opts={}) @host_to_try = format_pair(host, port) # Host and port of current master. @host = @port = nil # slave_ok can be true only if one node is specified @slave_ok = opts[:slave_ok] setup(opts) end
Determine if the connection is active. In a normal case the server_info operation will be performed without issues, but if the connection was dropped by the server or for some reason the sockets are unsynchronized, a ConnectionFailure will be raised and the return will be false.
@return [Boolean]
# File lib/mongo/connection.rb, line 548 def active? return false unless connected? ping true rescue ConnectionFailure false end
Save an authentication to this connection. When connecting, the connection will attempt to re-authenticate on every db specificed in the list of auths. This method is called automatically by DB#authenticate.
Note: this method will not actually issue an authentication command. To do that, either run Connection#apply_saved_authentication or DB#authenticate.
@param [String] db_name @param [String] username @param [String] password
@return [Hash] a hash representing the authentication just added.
# File lib/mongo/connection.rb, line 233 def add_auth(db_name, username, password) remove_auth(db_name) auth = {} auth['db_name'] = db_name auth['username'] = username auth['password'] = password @auths << auth auth end
Apply each of the saved database authentications.
@return [Boolean] returns true if authentications exist and succeeed, false
if none exists.
@raise [AuthenticationError] raises an exception if any one
authentication fails.
# File lib/mongo/connection.rb, line 211 def apply_saved_authentication(opts={}) return false if @auths.empty? @auths.each do |auth| self[auth['db_name']].issue_authentication(auth['username'], auth['password'], false, :socket => opts[:socket]) end true end
# File lib/mongo/connection.rb, line 265 def authenticate_pools @primary_pool.authenticate_existing end
Check a socket back into its pool.
# File lib/mongo/connection.rb, line 621 def checkin(socket) if @primary_pool @primary_pool.checkin(socket) end end
Checkin a socket used for reading. Note: this is overridden in ReplSetConnection.
# File lib/mongo/connection.rb, line 606 def checkin_reader(socket) warn "Connection#checkin_writer is not deprecated and will be removed " + "in driver v2.0. Use Connection#checkin instead." checkin(socket) end
Checkin a socket used for writing. Note: this is overridden in ReplSetConnection.
# File lib/mongo/connection.rb, line 614 def checkin_writer(socket) warn "Connection#checkin_writer is not deprecated and will be removed " + "in driver v2.0. Use Connection#checkin instead." checkin(socket) end
Checkout a socket for reading (i.e., a secondary node). Note: this is overridden in ReplSetConnection.
# File lib/mongo/connection.rb, line 592 def checkout_reader connect unless connected? @primary_pool.checkout end
Checkout a socket for writing (i.e., a primary node). Note: this is overridden in ReplSetConnection.
# File lib/mongo/connection.rb, line 599 def checkout_writer connect unless connected? @primary_pool.checkout end
Remove all authenication information stored in this connection.
@return [true] this operation return true because it always succeeds.
# File lib/mongo/connection.rb, line 260 def clear_auths @auths = [] true end
Close the connection to the database.
# File lib/mongo/connection.rb, line 576 def close @primary_pool.close if @primary_pool @primary_pool = nil @primary = nil end
Create a new socket and attempt to connect to master. If successful, sets host and port to master and returns the socket.
If connecting to a replica set, this method will replace the initially-provided seed list with any nodes known to the set.
@raise [ConnectionFailure] if unable to connect to any host or port.
# File lib/mongo/connection.rb, line 510 def connect close config = check_is_master(@host_to_try) if config if config['ismaster'] == 1 || config['ismaster'] == true @read_primary = true elsif @slave_ok @read_primary = false end @max_bson_size = config['maxBsonObjectSize'] || Mongo::DEFAULT_MAX_BSON_SIZE set_primary(@host_to_try) end if !connected? raise ConnectionFailure, "Failed to connect to a master node at #{@host_to_try[0]}:#{@host_to_try[1]}" end end
It‘s possible that we defined connected as all nodes being connected??? NOTE: Do check if this needs to be more stringent. Probably not since if any node raises a connection failure, all nodes will be closed.
# File lib/mongo/connection.rb, line 538 def connected? @primary_pool && @primary_pool.host && @primary_pool.port end
Copy the database from to to on localhost. The from database is assumed to be on localhost, but an alternate host can be specified.
@param [String] from name of the database to copy from. @param [String] to name of the database to copy to. @param [String] from_host host of the ‘from’ database. @param [String] username username for authentication against from_db (>=1.3.x). @param [String] password password for authentication against from_db (>=1.3.x).
# File lib/mongo/connection.rb, line 330 def copy_database(from, to, from_host="localhost", username=nil, password=nil) oh = BSON::OrderedHash.new oh[:copydb] = 1 oh[:fromhost] = from_host oh[:fromdb] = from oh[:todb] = to if username || password unless username && password raise MongoArgumentError, "Both username and password must be supplied for authentication." end nonce_cmd = BSON::OrderedHash.new nonce_cmd[:copydbgetnonce] = 1 nonce_cmd[:fromhost] = from_host result = self["admin"].command(nonce_cmd) oh[:nonce] = result["nonce"] oh[:username] = username oh[:key] = Mongo::Support.auth_key(username, password, oh[:nonce]) end self["admin"].command(oh) end
Return a hash with all database names and their respective sizes on disk.
@return [Hash]
# File lib/mongo/connection.rb, line 277 def database_info doc = self['admin'].command({:listDatabases => 1}) doc['databases'].each_with_object({}) do |db, info| info[db['name']] = db['sizeOnDisk'].to_i end end
Return an array of database names.
@return [Array]
# File lib/mongo/connection.rb, line 287 def database_names database_info.keys end
Return a database with the given name. See DB#new for valid options hash parameters.
@param [String] db_name a valid database name. @param [Hash] opts options to be passed to the DB constructor.
@return [Mongo::DB]
@core databases db-instance_method
# File lib/mongo/connection.rb, line 300 def db(db_name, opts={}) DB.new(db_name, self, opts) end
Fsync, then lock the mongod process against writes. Use this to get the datafiles in a state safe for snapshotting, backing up, etc.
@return [BSON::OrderedHash] the command response
# File lib/mongo/connection.rb, line 183 def lock! cmd = BSON::OrderedHash.new cmd[:fsync] = 1 cmd[:lock] = true self['admin'].command(cmd) end
Is this database locked against writes?
@return [Boolean]
# File lib/mongo/connection.rb, line 193 def locked? self['admin']['$cmd.sys.inprog'].find_one['fsyncLock'] == 1 end
Returns the maximum BSON object size as returned by the core server. Use the 4MB default when the server doesn‘t report this.
@return [Integer]
# File lib/mongo/connection.rb, line 586 def max_bson_size @max_bson_size end
The value of the read preference. Because this is a single-node connection, the value is +:primary+, and the connection will read from whichever type of node it‘s connected to.
# File lib/mongo/connection.rb, line 571 def read_preference :primary end
Determine whether we‘re reading from a primary node. If false, this connection connects to a secondary node and @slave_ok is true.
@return [Boolean]
# File lib/mongo/connection.rb, line 562 def read_primary? @read_primary end
Sends a message to the database and waits for the response.
@param [Integer] operation a MongoDB opcode. @param [BSON::ByteBuffer] message a message to send to the database. @param [String] log_message this is currently a no-op and will be removed. @param [Socket] socket a socket to use in lieu of checking out a new one. @param [Boolean] command (false) indicate whether this is a command. If this is a command,
the message will be sent to the primary node.
@param [Boolean] command (false) indicate whether the cursor should be exhausted. Set
this to true only when the OP_QUERY_EXHAUST flag is set.
@return [Array]
An array whose indexes include [0] documents returned, [1] number of document received, and [3] a cursor_id.
# File lib/mongo/connection.rb, line 470 def receive_message(operation, message, log_message=nil, socket=nil, command=false, read=:primary, exhaust=false) request_id = add_message_headers(message, operation) packed_message = message.to_s begin if socket sock = socket should_checkin = false else if command sock = checkout_writer elsif read == :primary sock = checkout_writer elsif read == :secondary sock = checkout_reader else sock = checkout_tagged(read) end should_checkin = true end result = '' @safe_mutexes[sock].synchronize do send_message_on_socket(packed_message, sock) result = receive(sock, request_id, exhaust) end ensure if should_checkin checkin(sock) end end result end
Send a message to MongoDB, adding the necessary headers.
@param [Integer] operation a MongoDB opcode. @param [BSON::ByteBuffer] message a message to send to the database.
@option opts [Symbol] :connection (:writer) The connection to which
this message should be sent. Valid options are :writer and :reader.
@return [Integer] number of bytes sent
# File lib/mongo/connection.rb, line 391 def send_message(operation, message, opts={}) if opts.is_a?(String) warn "Connection#send_message no longer takes a string log message. " + "Logging is now handled within the Collection and Cursor classes." opts = {} end connection = opts.fetch(:connection, :writer) begin add_message_headers(message, operation) packed_message = message.to_s if connection == :writer socket = checkout_writer else socket = checkout_reader end send_message_on_socket(packed_message, socket) ensure checkin(socket) end end
Sends a message to the database, waits for a response, and raises an exception if the operation has failed.
@param [Integer] operation a MongoDB opcode. @param [BSON::ByteBuffer] message a message to send to the database. @param [String] db_name the name of the database. used on call to get_last_error. @param [Hash] last_error_params parameters to be sent to getLastError. See DB#error for
available options.
@see DB#get_last_error for valid last error params.
@return [Hash] The document returned by the call to getlasterror.
# File lib/mongo/connection.rb, line 428 def send_message_with_safe_check(operation, message, db_name, log_message=nil, last_error_params=false) docs = num_received = cursor_id = '' add_message_headers(message, operation) last_error_message = BSON::ByteBuffer.new build_last_error_message(last_error_message, db_name, last_error_params) last_error_id = add_message_headers(last_error_message, Mongo::Constants::OP_QUERY) packed_message = message.append!(last_error_message).to_s begin sock = checkout_writer @safe_mutexes[sock].synchronize do send_message_on_socket(packed_message, sock) docs, num_received, cursor_id = receive(sock, last_error_id) end ensure checkin(sock) end if num_received == 1 && (error = docs[0]['err'] || docs[0]['errmsg']) close if error == "not master" error = "wtimeout" if error == "timeout" raise OperationFailure.new(docs[0]['code'].to_s + ': ' + error, docs[0]['code'], docs[0]) end docs[0] end
Get the build version of the current server.
@return [Mongo::ServerVersion]
object allowing easy comparability of version.
# File lib/mongo/connection.rb, line 371 def server_version ServerVersion.new(server_info["version"]) end
Unlock a previously fsync-locked mongod process.
@return [BSON::OrderedHash] command response
# File lib/mongo/connection.rb, line 200 def unlock! self['admin']['$cmd.sys.unlock'].find_one end
Generic initialization code.
# File lib/mongo/connection.rb, line 630 def setup(opts) # Default maximum BSON object size @max_bson_size = Mongo::DEFAULT_MAX_BSON_SIZE # Determine whether to use SSL. @ssl = opts.fetch(:ssl, false) if @ssl @socket_class = Mongo::SSLSocket else @socket_class = ::TCPSocket end # Authentication objects @auths = opts.fetch(:auths, []) # Lock for request ids. @id_lock = Mutex.new # Pool size and timeout. @pool_size = opts[:pool_size] || 1 if opts[:timeout] warn "The :timeout option has been deprecated " + "and will be removed in the 2.0 release. Use :pool_timeout instead." end @timeout = opts[:pool_timeout] || opts[:timeout] || 5.0 # Timeout on socket read operation. @op_timeout = opts[:op_timeout] || nil # Timeout on socket connect. @connect_timeout = opts[:connect_timeout] || nil # Mutex for synchronizing pool access # TODO: remove this. @connection_mutex = Mutex.new # Global safe option. This is false by default. @safe = opts[:safe] || false # Create a mutex when a new key, in this case a socket, # is added to the hash. @safe_mutexes = Hash.new { |h, k| h[k] = Mutex.new } # Condition variable for signal and wait @queue = ConditionVariable.new # Connection pool for primay node @primary = nil @primary_pool = nil @logger = opts[:logger] || nil if @logger @logger.debug("MongoDB logging. Please note that logging negatively impacts performance " + "and should be disabled for high-performance production apps.") end should_connect = opts.fetch(:connect, true) connect if should_connect end