Class: Mongo::Cluster

# File 'lib/mongo/cluster.rb', line 416

def connected?
  @update_lock.synchronize do
    !!@connected
  end
end

# File 'lib/mongo/cluster.rb', line 404

def connecting?
  @update_lock.synchronize do
    !!@connecting
  end
end

# File 'lib/mongo/cluster.rb', line 749

def has_writable_server?
  topology.has_writable_server?(self)
end

# File 'lib/mongo/cluster.rb', line 337

def load_balanced?
  topology.is_a?(Topology::LoadBalanced)
end

# File 'lib/mongo/cluster.rb', line 314

attr_reader :monitor_app_metadata

# File 'lib/mongo/cluster.rb', line 300

attr_reader :monitoring

# File 'lib/mongo/cluster.rb', line 297

attr_reader :options

# File 'lib/mongo/cluster.rb', line 317

attr_reader :push_monitor_app_metadata

# File 'lib/mongo/cluster.rb', line 325

attr_reader :seeds

# File 'lib/mongo/cluster.rb', line 480

attr_reader :server_selection_semaphore

# File 'lib/mongo/cluster.rb', line 328

attr_reader :session_pool

# File 'lib/mongo/cluster.rb', line 350

attr_reader :srv_monitor

# File 'lib/mongo/cluster.rb', line 305

attr_reader :topology

# File 'lib/mongo/cluster.rb', line 302

attr_reader :tracer

# File 'lib/mongo/cluster.rb', line 825

def add(host, add_options = nil)
  address = Address.new(host, options)
  return if addresses.include?(address)

  opts = options.merge(monitor: false)
  # If we aren't starting the monitoring threads, we also don't want to
  # start the pool's populator thread.
  opts.merge!(populator_io: false) unless options.fetch(:monitoring_io, true)
  # Note that in a load-balanced topology, every server must be a
  # load balancer (load_balancer: true is specified in the options)
  # but this option isn't set here because we are required by the
  # specifications to pretend the server started out as an unknown one
  # and publish server description change event into the load balancer
  # one. The actual correct description for this server will be set
  # by the fabricate_lb_sdam_events_and_set_server_type method.
  server = Server.new(address, self, @monitoring, event_listeners, opts)
  @update_lock.synchronize do
    # Need to recheck whether server is present in @servers, because
    # the previous check was not under a lock.
    # Since we are under the update lock here, we cannot call servers_list.
    return if @servers.map(&:address).include?(address)

    @servers.push(server)
  end
  server.start_monitoring if add_options.nil? || add_options[:monitor] != false
  server
end

# File 'lib/mongo/cluster.rb', line 443

def addresses
  servers_list.map(&:address)
end

# File 'lib/mongo/cluster.rb', line 498

def close
  @state_change_lock.synchronize do
    return nil unless connecting? || connected?

    if options[:cleanup] != false
      session_pool.end_sessions
      @periodic_executor.stop!
    end
    @srv_monitor_lock.synchronize do
      @srv_monitor.stop! if @srv_monitor
    end
    @servers.each do |server|
      next unless server.connected?

      server.close
      publish_sdam_event(
        Monitoring::SERVER_CLOSED,
        Monitoring::Event::ServerClosed.new(server.address, topology)
      )
    end
    publish_sdam_event(
      Monitoring::TOPOLOGY_CLOSED,
      Monitoring::Event::TopologyClosed.new(topology)
    )
    @update_lock.synchronize do
      @connecting = @connected = false
    end
  end
  nil
end

# File 'lib/mongo/cluster.rb', line 929

def disconnect_server_if_connected(server)
  return unless server.connected?

  server.clear_description
  server.disconnect!
  publish_sdam_event(
    Monitoring::SERVER_CLOSED,
    Monitoring::Event::ServerClosed.new(server.address, topology)
  )
end

# File 'lib/mongo/cluster.rb', line 1040

def fabricate_lb_sdam_events_and_set_server_type
  # Although there is no monitoring connection in load balanced mode,
  # we must emit the following series of SDAM events.
  server = @servers.first
  # We are guaranteed to have the server here.
  server.publish_opening_event
  server_desc = server.description
  # This is where a load balancer actually gets its correct server
  # description.
  server.update_description(
    Server::Description.new(server.address, {},
                            load_balancer: true,
                            force_load_balancer: options[:connect] == :load_balanced)
  )
  publish_sdam_event(
    Monitoring::SERVER_DESCRIPTION_CHANGED,
    Monitoring::Event::ServerDescriptionChanged.new(
      server.address,
      topology,
      server_desc,
      server.description
    )
  )
  recreate_topology(topology, topology)
end

# File 'lib/mongo/cluster.rb', line 737

def has_readable_server?(server_selector = nil)
  topology.has_readable_server?(self, server_selector)
end

# File 'lib/mongo/cluster.rb', line 395

def heartbeat_interval
  options[:heartbeat_frequency] || Server::Monitor::DEFAULT_HEARTBEAT_INTERVAL
end

# File 'lib/mongo/cluster.rb', line 465

def inspect
  "#<Mongo::Cluster:0x#{object_id} servers=#{servers} topology=#{topology.summary}>"
end

# File 'lib/mongo/cluster.rb', line 366

def max_read_retries
  options[:max_read_retries] || MAX_READ_RETRIES
end

# File 'lib/mongo/cluster.rb', line 768

def next_primary(_ping = nil, session = nil, timeout: nil)
  ServerSelector.primary.select_server(
    self,
    nil,
    session,
    timeout: timeout
  )
end

# File 'lib/mongo/cluster.rb', line 788

def pool(server)
  server.pool
end

# File 'lib/mongo/cluster.rb', line 1086

def possibly_warn_about_compatibility!
  if topology.server_hosts_match_any?(COSMOSDB_HOST_PATTERNS)
    log_info COSMOSDB_LOG_MESSAGE
    return
  end

  return unless topology.server_hosts_match_any?(DOCUMENTDB_HOST_PATTERNS)

  log_info DOCUMENTDB_LOG_MESSAGE
  nil
end

# File 'lib/mongo/cluster.rb', line 1024

def raise_sessions_not_supported
  # Intentionally using @servers instead of +servers+ here because we
  # are supposed to be already holding the @update_lock and we cannot
  # recursively acquire it again.
  offending_servers = @servers.select do |server|
    server.description.data_bearing? && server.logical_session_timeout.nil?
  end
  reason = if offending_servers.empty?
             "There are no known data bearing servers (current seeds: #{@servers.map(&:address).map(&:seed).join(', ')})"
           else
             "The following servers have null logical session timeout: #{offending_servers.map(&:address).map(&:seed).join(', ')}"
           end
  msg = "The deployment that the driver is connected to does not support sessions: #{reason}"
  raise Error::SessionsNotSupported, msg
end

# File 'lib/mongo/cluster.rb', line 384

def read_retry_interval
  options[:read_retry_interval] || READ_RETRY_INTERVAL
end

# File 'lib/mongo/cluster.rb', line 539

def reconnect!
  @state_change_lock.synchronize do
    @update_lock.synchronize do
      @connecting = true
    end
    scan!
    servers.each do |server|
      server.reconnect!
    end
    @periodic_executor.restart!
    @srv_monitor_lock.synchronize do
      @srv_monitor.run! if @srv_monitor
    end
    @update_lock.synchronize do
      @connecting = false
      @connected = true
    end
  end
end

# File 'lib/mongo/cluster.rb', line 1066

def recreate_topology(new_topology_template, previous_topology)
  @topology = topology.class.new(new_topology_template.options, new_topology_template.monitoring, self)
  publish_sdam_event(
    Monitoring::TOPOLOGY_CHANGED,
    Monitoring::Event::TopologyChanged.new(previous_topology, @topology)
  )
end

# File 'lib/mongo/cluster.rb', line 872

def remove(host, disconnect: true)
  address = Address.new(host)
  removed_servers = []
  @update_lock.synchronize do
    @servers.delete_if do |server|
      (server.address == address).tap do |delete|
        removed_servers << server if delete
      end
    end
  end
  if disconnect != false
    removed_servers.each do |server|
      disconnect_server_if_connected(server)
    end
  end
  if disconnect == false
    removed_servers
  else
    removed_servers.any?
  end
end

# File 'lib/mongo/cluster.rb', line 623

def run_sdam_flow(previous_desc, updated_desc, options = {})
  if load_balanced?
    if updated_desc.config.empty? && !options[:keep_connection_pool]
      servers_list.each do |server|
        # TODO: should service id be taken out of updated_desc?
        # We could also assert that
        # options[:service_id] == updated_desc.service_id
        err = options[:scan_error]
        interrupt = err && (err.is_a?(Error::SocketError) || err.is_a?(Error::SocketTimeoutError))
        server.clear_connection_pool(service_id: options[:service_id], interrupt_in_use_connections: interrupt)
      end
    end
    return
  end

  @sdam_flow_lock.synchronize do
    flow = SdamFlow.new(self, previous_desc, updated_desc,
                        awaited: options[:awaited])
    flow.server_description_changed

    # SDAM flow may alter the updated description - grab the final
    # version for the purposes of broadcasting if a server is available
    updated_desc = flow.updated_desc

    if !options[:keep_connection_pool] && flow.became_unknown?
      servers_list.each do |server|
        next unless server.address == updated_desc.address

        err = options[:scan_error]
        interrupt = err && (err.is_a?(Error::SocketError) || err.is_a?(Error::SocketTimeoutError))
        server.clear_connection_pool(interrupt_in_use_connections: interrupt)
      end
    end

    start_stop_srv_monitor
  end

  # Some updated descriptions, e.g. a mismatched me one, result in the
  # server whose description we are processing being removed from
  # the topology. When this happens, the server's monitoring thread gets
  # killed. As a result, any code after the flow invocation may not run
  # a particular monitor instance, hence there should generally not be
  # any code in this method past the flow invocation.
  #
  # However, this broadcast call can be here because if the monitoring
  # thread got killed the server should have been closed and no client
  # should be currently waiting for it, thus not signaling the semaphore
  # shouldn't cause any problems.
  return if updated_desc.unknown?

  server_selection_semaphore.broadcast
end

# File 'lib/mongo/cluster.rb', line 582

def scan!(sync = true)
  if sync
    servers_list.each do |server|
      if server.monitor
        server.monitor.scan!
      else
        log_warn("Synchronous scan requested on cluster #{summary} but server #{server} has no monitor")
      end
    end
  else
    servers_list.each do |server|
      server.scan_semaphore.signal
    end
  end
  true
end

# File 'lib/mongo/cluster.rb', line 431

def servers
  topology.servers(servers_list)
end

# File 'lib/mongo/cluster.rb', line 922

def servers_list
  @update_lock.synchronize do
    @servers.dup
  end
end

# File 'lib/mongo/cluster.rb', line 690

def set_server_list(server_address_strs)
  @sdam_flow_lock.synchronize do
    # If one of the new addresses is not in the current servers list,
    # add it to the servers list.
    server_address_strs.each do |address_str|
      add(address_str) unless servers_list.any? { |server| server.address.seed == address_str }
    end

    # If one of the servers' addresses are not in the new address list,
    # remove that server from the servers list.
    servers_list.each do |server|
      unless server_address_strs.any? { |address_str| server.address.seed == address_str }
        remove(server.address.seed)
      end
    end
  end
end

# File 'lib/mongo/cluster.rb', line 997

def start_stop_srv_monitor
  # SRV URI is either always given or not for a given cluster, if one
  # wasn't given we shouldn't ever have an SRV monitor to manage.
  return unless options[:srv_uri]

  if topology.is_a?(Topology::Sharded) || topology.is_a?(Topology::Unknown)
    # Start SRV monitor
    @srv_monitor_lock.synchronize do
      unless @srv_monitor
        monitor_options = Utils.shallow_symbolize_keys(options.merge(
                                                         timeout: options[:connect_timeout] || Server::CONNECT_TIMEOUT
                                                       ))
        @srv_monitor = _srv_monitor = Srv::Monitor.new(self, **monitor_options)
      end
      @srv_monitor.run!
    end
  else
    # Stop SRV monitor if running. This path is taken when the client
    # is given an SRV URI to a standalone/replica set; when the topology
    # is discovered, since it's not a sharded cluster, the SRV monitor
    # needs to be stopped.
    @srv_monitor_lock.synchronize do
      @srv_monitor.stop! if @srv_monitor
    end
  end
end

# File 'lib/mongo/cluster.rb', line 473

def summary
  '#<Cluster ' +
    "topology=#{topology.summary} " +
    "servers=[#{servers_list.map(&:summary).join(',')}]>"
end

# File 'lib/mongo/cluster.rb', line 988

def trigger_periodic_executor!
  return unless @periodic_executor

  @periodic_executor.execute
end

# File 'lib/mongo/cluster.rb', line 802

def update_cluster_time(result)
  return unless cluster_time_doc = result.cluster_time

  @cluster_time_lock.synchronize do
    advance_cluster_time(cluster_time_doc)
  end
end

# File 'lib/mongo/cluster.rb', line 895

def update_topology(new_topology)
  old_topology = nil
  @update_lock.synchronize do
    old_topology = topology
    @topology = new_topology
  end

  # If new topology has data bearing servers, we know for sure whether
  # sessions are supported - update our cached value.
  # If new topology has no data bearing servers, leave the old value
  # as it is and sessions_supported? method will perform server selection
  # to try to determine session support accurately, falling back to the
  # last known value.
  if topology.data_bearing_servers?
    sessions_supported = !!topology.logical_session_timeout
    @update_lock.synchronize do
      @sessions_supported = sessions_supported
    end
  end

  publish_sdam_event(
    Monitoring::TOPOLOGY_CHANGED,
    Monitoring::Event::TopologyChanged.new(old_topology, topology)
  )
end

# File 'lib/mongo/cluster.rb', line 964

def validate_session_support!(timeout: nil)
  return if topology.is_a?(Topology::LoadBalanced)

  @state_change_lock.synchronize do
    @sdam_flow_lock.synchronize do
      raise_sessions_not_supported if topology.data_bearing_servers? && !topology.logical_session_timeout
    end
  end

  # No data bearing servers known - perform server selection to try to
  # get a response from at least one of them, to return an accurate
  # assessment of whether sessions are currently supported.
  ServerSelector.get(mode: :primary_preferred).select_server(self, timeout: timeout)
  @state_change_lock.synchronize do
    @sdam_flow_lock.synchronize do
      raise_sessions_not_supported unless topology.logical_session_timeout
    end
  end
end