Class: Mongo::Client

# File 'lib/mongo/client.rb', line 139

attr_reader :cluster

# File 'lib/mongo/client.rb', line 142

attr_reader :database

# File 'lib/mongo/client.rb', line 149

attr_reader :encrypter

# File 'lib/mongo/client.rb', line 145

attr_reader :options

# File 'lib/mongo/client.rb', line 195

def [](collection_name, options = {})
  database[collection_name, options]
end

# File 'lib/mongo/client.rb', line 1707

def assert_not_closed
  if closed?
    raise Error::ClientClosed, "The client was closed and is not usable for operations. Call #reconnect to reset this client instance or create a new client instance"
  end
end

# File 'lib/mongo/client.rb', line 1268

def build_encrypter
  @encrypter = Crypt::AutoEncrypter.new(
    @options[:auto_encryption_options].merge(client: self)
  )
end

# File 'lib/mongo/client.rb', line 865

def close
  @connect_lock.synchronize do
    @closed = true
    do_close
  end
  true
end

# File 'lib/mongo/client.rb', line 876

def close_encrypter
  @encrypter.close if @encrypter

  true
end

# File 'lib/mongo/client.rb', line 1352

def cluster_modifying?(new_options)
  cluster_options = new_options.reject do |name|
    CRUD_OPTIONS.include?(name.to_sym)
  end
  cluster_options.any? do |name, value|
    options[name] != value
  end
end

# File 'lib/mongo/client.rb', line 611

def cluster_options
  # We share clusters when a new client with different CRUD_OPTIONS
  # is requested; therefore, cluster should not be getting any of these
  # options upon instantiation
  options.reject do |key, value|
    CRUD_OPTIONS.include?(key.to_sym)
  end.merge(
    # but need to put the database back in for auth...
    database: options[:database],

    # Put these options in for legacy compatibility, but note that
    # their values on the client and the cluster do not have to match -
    # applications should read these values from client, not from cluster
    max_read_retries: options[:max_read_retries],
    read_retry_interval: options[:read_retry_interval],
  ).tap do |options|
    # If the client has a cluster already, forward srv_uri to the new
    # cluster to maintain SRV monitoring. If the client is brand new,
    # its constructor sets srv_uri manually.
    if cluster
      options.update(srv_uri: cluster.options[:srv_uri])
    end
  end
end

# File 'lib/mongo/client.rb', line 932

def database_names(filter = {}, opts = {})
  list_databases(filter, true, opts).collect{ |info| info['name'] }
end

# File 'lib/mongo/client.rb', line 1276

def default_options(options)
  Database::DEFAULT_OPTIONS.dup.tap do |default_options|
    if options[:auth_mech] || options[:user]
      default_options[:auth_source] = Auth::User.default_auth_source(options)
    end

    if options[:auth_mech] == :gssapi
      default_options[:auth_mech_properties] = { service_name: 'mongodb' }
    end

    default_options[:retry_reads] = true
    default_options[:retry_writes] = true
  end
end

# File 'lib/mongo/client.rb', line 1292

def do_close
  @cluster.close
  close_encrypter
end

# File 'lib/mongo/client.rb', line 1178

def encrypted_fields_map
  @encrypted_fields_map ||= @options.fetch(:auto_encryption_options, {})[:encrypted_fields_map]
end

# File 'lib/mongo/client.rb', line 182

alias_method :eql?, :==

# File 'lib/mongo/client.rb', line 1118

def get_session(options = {})
  get_session!(options)
rescue Error::SessionsNotSupported
  nil
end

# File 'lib/mongo/client.rb', line 1319

def get_session!(options = {})
  if options[:session]
    return options[:session].validate!(self)
  end

  cluster.validate_session_support!(timeout: timeout_sec)

  options = {implicit: true}.update(options)

  server_session = if options[:implicit]
    nil
  else
    cluster.session_pool.checkout
  end

  Session.new(server_session, self, options)
end

# File 'lib/mongo/client.rb', line 207

def hash
  [cluster, options].hash
end

# File 'lib/mongo/client.rb', line 1343

def initialize_copy(original)
  @options = original.options.dup
  @connect_lock = Mutex.new
  @monitoring = @cluster ? monitoring : Monitoring.new(options)
  @database = nil
  @read_preference = nil
  @write_concern = nil
end

# File 'lib/mongo/client.rb', line 674

def inspect
  "#<Mongo::Client:0x#{object_id} cluster=#{cluster.summary}>"
end

# File 'lib/mongo/client.rb', line 963

def list_databases(filter = {}, name_only = false, opts = {})
  cmd = { listDatabases: 1 }
  cmd[:nameOnly] = !!name_only
  cmd[:filter] = filter unless filter.empty?
  cmd[:authorizedDatabases] = true if opts[:authorized_databases]
  use(Database::ADMIN).database.read_command(cmd, opts).first[Database::DATABASES]
end

# File 'lib/mongo/client.rb', line 986

def list_mongo_databases(filter = {}, opts = {})
  database_names(filter, opts).collect do |name|
    Database.new(self, name, options)
  end
end

# File 'lib/mongo/client.rb', line 642

def max_read_retries
  options[:max_read_retries] || Cluster::MAX_READ_RETRIES
end

# File 'lib/mongo/client.rb', line 662

def max_write_retries
  options[:max_write_retries] || Cluster::MAX_WRITE_RETRIES
end

# File 'lib/mongo/client.rb', line 159

def monitoring
  if cluster
    cluster.monitoring
  else
    @monitoring
  end
end

# File 'lib/mongo/client.rb', line 1208

def process_addresses(addresses, options)
  if addresses.is_a?(String)
    process_addresses_string(addresses, options)
  else
    process_addresses_array(addresses, options)
  end
end

# File 'lib/mongo/client.rb', line 1252

def process_addresses_array(addresses, options)
  {}.tap do |processed|
    processed[:addresses] = addresses
    processed[:options] = options

    addresses.each do |addr|
      if addr =~ /\Amongodb(\+srv)?:\/\//i
        raise ArgumentError, "Host '#{addr}' should not contain protocol. Did you mean to not use an array?"
      end
    end

    @srv_records = nil
  end
end

# File 'lib/mongo/client.rb', line 1224

def process_addresses_string(addresses, options)
  {}.tap do |processed|
    processed[:uri] = uri = URI.get(addresses, options)
    processed[:addresses] = uri.servers

    uri_options = uri.client_options.dup
    # Special handing for :write and :write_concern: allow client Ruby
    # options to override URI options, even when the Ruby option uses the
    # deprecated :write key and the URI option uses the current
    # :write_concern key
    if options[:write]
      uri_options.delete(:write_concern)
    end

    processed[:options] = uri_options.merge(options)

    @srv_records = uri.srv_records
  end
end

# File 'lib/mongo/client.rb', line 839

def read_concern
  options[:read_concern]
end

# File 'lib/mongo/client.rb', line 722

def read_preference
  @read_preference ||= options[:read]
end

# File 'lib/mongo/client.rb', line 652

def read_retry_interval
  options[:read_retry_interval] || Cluster::READ_RETRY_INTERVAL
end

# File 'lib/mongo/client.rb', line 890

def reconnect
  addresses = cluster.addresses.map(&:to_s)

  @connect_lock.synchronize do
    do_close rescue nil

    @cluster = Cluster.new(addresses, monitoring, cluster_options)

    if @options[:auto_encryption_options]
      build_encrypter
    end

    @closed = false
  end

  true
end

# File 'lib/mongo/client.rb', line 700

def server_selector
  @server_selector ||= if read_preference
    ServerSelector.get(read_preference)
  else
    ServerSelector.primary
  end
end

# File 'lib/mongo/client.rb', line 1011

def start_session(options = {})
  session = get_session!(options.merge(implicit: false))
  if block_given?
    begin
      yield session
    ensure
      session.end_session
    end
  else
    session
  end
end

# File 'lib/mongo/client.rb', line 686

def summary
  "#<Client cluster=#{cluster.summary}>"
end

# File 'lib/mongo/client.rb', line 1184

def timeout_ms
  @options[:timeout_ms]
end

# File 'lib/mongo/client.rb', line 1190

def timeout_sec
  if timeout_ms.nil?
    nil
  else
    timeout_ms / 1_000.0
  end
end

# File 'lib/mongo/client.rb', line 789

def update_options(new_options)
  old_options = @options

  new_options = self.class.canonicalize_ruby_options(new_options || {})

  validate_new_options!(new_options).tap do |opts|
    # Our options are frozen
    options = @options.dup
    if options[:write] && opts[:write_concern]
      options.delete(:write)
    end
    if options[:write_concern] && opts[:write]
      options.delete(:write_concern)
    end

    options.update(opts)
    @options = options.freeze

    auto_encryption_options_changed =
      @options[:auto_encryption_options] != old_options[:auto_encryption_options]

    # If there are new auto_encryption_options, create a new encrypter.
    # Otherwise, allow the new client to share an encrypter with the
    # original client.
    #
    # If auto_encryption_options are nil, set @encrypter to nil, but do not
    # close the encrypter because it may still be used by the original client.
    if @options[:auto_encryption_options] && auto_encryption_options_changed
      @connect_lock.synchronize do
        build_encrypter
      end
    elsif @options[:auto_encryption_options].nil?
      @connect_lock.synchronize do
        @encrypter = nil
      end
    end

    validate_options!
    validate_authentication_options!
  end
end

# File 'lib/mongo/client.rb', line 741

def use(name)
  with(database: name)
end

# File 'lib/mongo/client.rb', line 1629

def valid_compressors(compressors)
  compressors.select do |compressor|
    if !VALID_COMPRESSORS.include?(compressor)
      log_warn("Unsupported compressor '#{compressor}' in list '#{compressors}'. " +
                   "This compressor will not be used.")
      false
    else

      true
    end
  end
end

# File 'lib/mongo/client.rb', line 1574

def validate_authentication_options!
  auth_mech = options[:auth_mech]
  user = options[:user]
  password = options[:password]
  auth_source = options[:auth_source]
  mech_properties = options[:auth_mech_properties]

  if auth_mech.nil?
    if user && user.empty?
      raise Mongo::Auth::InvalidConfiguration, 'Empty username is not supported for default auth mechanism'
    end

    if auth_source == ''
      raise Mongo::Auth::InvalidConfiguration, 'Auth source cannot be empty for default auth mechanism'
    end

    return
  end

  if !Mongo::Auth::SOURCES.key?(auth_mech)
    raise Mongo::Auth::InvalidMechanism.new(auth_mech)
  end

  if user.nil? && !%i(aws mongodb_x509).include?(auth_mech)
    raise Mongo::Auth::InvalidConfiguration, "Username is required for auth mechanism #{auth_mech}"
  end

  if password.nil? && !%i(aws gssapi mongodb_x509).include?(auth_mech)
    raise Mongo::Auth::InvalidConfiguration, "Password is required for auth mechanism #{auth_mech}"
  end

  if password && auth_mech == :mongodb_x509
    raise Mongo::Auth::InvalidConfiguration, 'Password is not supported for :mongodb_x509 auth mechanism'
  end

  if auth_mech == :aws && user && !password
    raise Mongo::Auth::InvalidConfiguration, 'Username is provided but password is not provided for :aws auth mechanism'
  end

  if %i(aws gssapi mongodb_x509).include?(auth_mech)
    if !['$external', nil].include?(auth_source)
      raise Mongo::Auth::InvalidConfiguration, "#{auth_source} is an invalid auth source for #{auth_mech}; valid options are $external and nil"
    end
  else
    # Auth source is the database name, and thus cannot be the empty string.
    if auth_source == ''
      raise Mongo::Auth::InvalidConfiguration, "Auth source cannot be empty for auth mechanism #{auth_mech}"
    end
  end

  if mech_properties && !%i(aws gssapi).include?(auth_mech)
    raise Mongo::Auth::InvalidConfiguration, ":mechanism_properties are not supported for auth mechanism #{auth_mech}"
  end
end

# File 'lib/mongo/client.rb', line 1677

def validate_max_connecting!(option, opts)
  if option == :max_connecting && opts.key?(:max_connecting)
    max_connecting = opts[:max_connecting] || Server::ConnectionPool::DEFAULT_MAX_CONNECTING
    if max_connecting <= 0
      raise Error::InvalidMaxConnecting.new(opts[:max_connecting])
    end
  end
  true
end

# File 'lib/mongo/client.rb', line 1660

def validate_max_min_pool_size!(option, opts)
  if option == :min_pool_size && opts[:min_pool_size]
    max = opts[:max_pool_size] || Server::ConnectionPool::DEFAULT_MAX_SIZE
    if max != 0 && opts[:min_pool_size] > max
      raise Error::InvalidMinPoolSize.new(opts[:min_pool_size], max)
    end
  end
  true
end

# File 'lib/mongo/client.rb', line 1365

def validate_new_options!(opts)
  return Options::Redacted.new unless opts
  if opts[:read_concern]
    # Raise an error for non user-settable options
    if opts[:read_concern][:after_cluster_time]
      raise Mongo::Error::InvalidReadConcern.new(
        'The after_cluster_time read_concern option cannot be specified by the user'
      )
    end

    given_keys = opts[:read_concern].keys.map(&:to_s)
    allowed_keys = ['level']
    invalid_keys = given_keys - allowed_keys
    # Warn that options are invalid but keep it and forward to the server
    unless invalid_keys.empty?
      log_warn("Read concern has invalid keys: #{invalid_keys.join(',')}.")
    end
  end

  if server_api = opts[:server_api]
    unless server_api.is_a?(Hash)
      raise ArgumentError, ":server_api value must be a hash: #{server_api}"
    end

    extra_keys = server_api.keys - %w(version strict deprecation_errors)
    unless extra_keys.empty?
      raise ArgumentError, "Unknown keys under :server_api: #{extra_keys.map(&:inspect).join(', ')}"
    end

    if version = server_api[:version]
      unless VALID_SERVER_API_VERSIONS.include?(version)
        raise ArgumentError, "Unknown server API version: #{version}"
      end
    end
  end

  Lint.validate_underscore_read_preference(opts[:read])
  Lint.validate_read_concern_option(opts[:read_concern])
  opts.each.inject(Options::Redacted.new) do |_options, (k, v)|
    key = k.to_sym
    if VALID_OPTIONS.include?(key)
      validate_max_min_pool_size!(key, opts)
      validate_max_connecting!(key, opts)
      validate_read!(key, opts)
      if key == :compressors
        compressors = valid_compressors(v)

        if compressors.include?('snappy')
          validate_snappy_compression!
        end

        if compressors.include?('zstd')
          validate_zstd_compression!
        end

        _options[key] = compressors unless compressors.empty?
      elsif key == :srv_max_hosts
        if v && (!v.is_a?(Integer) || v < 0)
          log_warn("#{v} is not a valid integer for srv_max_hosts")
        else
          _options[key] = v
        end
      else
        _options[key] = v
      end
    else
      log_warn("Unsupported client option '#{k}'. It will be ignored.")
    end
    _options
  end
end

# File 'lib/mongo/client.rb', line 1440

def validate_options!(addresses = nil, is_srv: nil)
  if options[:write] && options[:write_concern] && options[:write] != options[:write_concern]
    raise ArgumentError, "If :write and :write_concern are both given, they must be identical: #{options.inspect}"
  end

  connect = options[:connect]&.to_sym

  if connect && !%i(direct replica_set sharded load_balanced).include?(connect)
    raise ArgumentError, "Invalid :connect option value: #{connect}"
  end

  if options[:direct_connection]
    if connect && connect != :direct
      raise ArgumentError, "Conflicting client options: direct_connection=true and connect=#{connect}"
    end
    # When a new client is created, we get the list of seed addresses
    if addresses && addresses.length > 1
      raise ArgumentError, "direct_connection=true cannot be used with multiple seeds"
    end
    # When a client is copied using #with, we have a cluster
    if cluster && !cluster.topology.is_a?(Mongo::Cluster::Topology::Single)
      raise ArgumentError, "direct_connection=true cannot be used with topologies other than Single (this client is #{cluster.topology.class.name.sub(/.*::/, '')})"
    end
  end

  if options[:load_balanced]
    if addresses && addresses.length > 1
      raise ArgumentError, "load_balanced=true cannot be used with multiple seeds"
    end

    if options[:direct_connection]
      raise ArgumentError, "direct_connection=true cannot be used with load_balanced=true"
    end

    if connect && connect != :load_balanced
      raise ArgumentError, "connect=#{connect} cannot be used with load_balanced=true"
    end

    if options[:replica_set]
      raise ArgumentError, "load_balanced=true cannot be used with replica_set option"
    end
  end

  if connect == :load_balanced
    if addresses && addresses.length > 1
      raise ArgumentError, "connect=load_balanced cannot be used with multiple seeds"
    end

    if options[:replica_set]
      raise ArgumentError, "connect=load_balanced cannot be used with replica_set option"
    end
  end

  if options[:direct_connection] == false && connect && connect == :direct
    raise ArgumentError, "Conflicting client options: direct_connection=false and connect=#{connect}"
  end

  %i(connect_timeout socket_timeout).each do |key|
    if value = options[key]
      unless Numeric === value
        raise ArgumentError, "#{key} must be a non-negative number: #{value}"
      end
      if value < 0
        raise ArgumentError, "#{key} must be a non-negative number: #{value}"
      end
    end
  end

  if value = options[:bg_error_backtrace]
    case value
    when Integer
      if value <= 0
        raise ArgumentError, ":bg_error_backtrace option value must be true, false, nil or a positive integer: #{value}"
      end
    when true
      # OK
    else
      raise ArgumentError, ":bg_error_backtrace option value must be true, false, nil or a positive integer: #{value}"
    end
  end

  if libraries = options[:wrapping_libraries]
    unless Array === libraries
      raise ArgumentError, ":wrapping_libraries must be an array of hashes: #{libraries}"
    end

    libraries = libraries.map do |library|
      Utils.shallow_symbolize_keys(library)
    end

    libraries.each do |library|
      unless Hash === library
        raise ArgumentError, ":wrapping_libraries element is not a hash: #{library}"
      end

      if library.empty?
        raise ArgumentError, ":wrapping_libraries element is empty"
      end

      unless (library.keys - %i(name platform version)).empty?
        raise ArgumentError, ":wrapping_libraries element has invalid keys (allowed keys: :name, :platform, :version): #{library}"
      end

      library.each do |key, value|
        if value.include?('|')
          raise ArgumentError, ":wrapping_libraries element value cannot include '|': #{value}"
        end
      end
    end
  end

  if options[:srv_max_hosts] && options[:srv_max_hosts] > 0
    if options[:replica_set]
      raise ArgumentError, ":srv_max_hosts > 0 cannot be used with :replica_set option"
    end

    if options[:load_balanced]
      raise ArgumentError, ":srv_max_hosts > 0 cannot be used with :load_balanced=true"
    end
  end

  unless is_srv.nil? || is_srv
    if options[:srv_max_hosts]
      raise ArgumentError, ":srv_max_hosts cannot be used on non-SRV URI"
    end

    if options[:srv_service_name]
      raise ArgumentError, ":srv_service_name cannot be used on non-SRV URI"
    end
  end
end

# File 'lib/mongo/client.rb', line 1687

def validate_read!(option, opts)
  if option == :read && opts.has_key?(:read)
    read = opts[:read]
    # We could check if read is a Hash, but this would fail
    # for custom classes implementing key access ([]).
    # Instead reject common cases of strings and symbols.
    if read.is_a?(String) || read.is_a?(Symbol)
      raise Error::InvalidReadOption.new(read, "the read preference must be specified as a hash: { mode: #{read.inspect} }")
    end

    if mode = read[:mode]
      mode = mode.to_sym
      unless Mongo::ServerSelector::PREFERENCES.include?(mode)
        raise Error::InvalidReadOption.new(read, "mode #{mode} is not one of recognized modes")
      end
    end
  end
  true
end

# File 'lib/mongo/client.rb', line 1642

def validate_snappy_compression!
  return if defined?(Snappy)
  require 'snappy'
rescue LoadError => e
  raise Error::UnmetDependency, "Cannot enable snappy compression because the snappy gem " \
    "has not been installed. Add \"gem 'snappy'\" to your Gemfile and run " \
    "\"bundle install\" to install the gem. (#{e.class}: #{e})"
end

# File 'lib/mongo/client.rb', line 1651

def validate_zstd_compression!
  return if defined?(Zstd)
  require 'zstd-ruby'
rescue LoadError => e
  raise Error::UnmetDependency, "Cannot enable zstd compression because the zstd-ruby gem " \
    "has not been installed. Add \"gem 'zstd-ruby'\" to your Gemfile and run " \
    "\"bundle install\" to install the gem. (#{e.class}: #{e})"
end

# File 'lib/mongo/client.rb', line 1088

def watch(pipeline = [], options = {})
  return use(Database::ADMIN).watch(pipeline, options) unless database.name == Database::ADMIN

  view_options = options.dup
  view_options[:cursor_type] = :tailable_await if options[:max_await_time_ms]

  Mongo::Collection::View::ChangeStream.new(
    Mongo::Collection::View.new(self["#{Database::COMMAND}.aggregate"], {}, view_options),
    pipeline,
    Mongo::Collection::View::ChangeStream::CLUSTER,
    options)
end

# File 'lib/mongo/client.rb', line 763

def with(new_options = nil)
  clone.tap do |client|
    opts = client.update_options(new_options || Options::Redacted.new)
    Database.create(client)
    # We can't use the same cluster if some options that would affect it
    # have changed.
    if cluster_modifying?(opts)
      Cluster.create(client, monitoring: opts[:monitoring])
    end
  end
end

# File 'lib/mongo/client.rb', line 1144

def with_session(options = {}, &block)
  # TODO: Add this back in RUBY-3174.
  # assert_not_closed

  session = get_session(options)

  yield session
ensure
  if session && session.implicit?
    session.end_session
  end
end

# File 'lib/mongo/client.rb', line 852

def write_concern
  @write_concern ||= WriteConcern.get(options[:write_concern] || options[:write])
end