Class: Mongo::Cluster

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

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

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

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

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

def has_writable_server?
  topology.has_writable_server?(self)
end

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

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

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

attr_reader :monitor_app_metadata

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

attr_reader :monitoring

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

attr_reader :options

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

attr_reader :push_monitor_app_metadata

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

attr_reader :seeds

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

attr_reader :server_selection_semaphore

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

attr_reader :session_pool

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

attr_reader :srv_monitor

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

attr_reader :topology

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

def add(host, add_options=nil)
  address = Address.new(host, options)
  if !addresses.include?(address)
    opts = options.merge(monitor: false)
    # If we aren't starting the montoring 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 pretent 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
    if add_options.nil? || add_options[:monitor] != false
      server.start_monitoring
    end
    server
  end
end

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

def addresses
  servers_list.map(&:address)
end

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

def close
  @state_change_lock.synchronize do
    unless connecting? || connected?
      return nil
    end
    if options[:cleanup] != false
      session_pool.end_sessions
      @periodic_executor.stop!
    end
    @srv_monitor_lock.synchronize do
      if @srv_monitor
        @srv_monitor.stop!
      end
    end
    @servers.each do |server|
      if server.connected?
        server.close
        publish_sdam_event(
          Monitoring::SERVER_CLOSED,
          Monitoring::Event::ServerClosed.new(server.address, topology)
        )
      end
    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 953

def disconnect_server_if_connected(server)
  if server.connected?
    server.clear_description
    server.disconnect!
    publish_sdam_event(
      Monitoring::SERVER_CLOSED,
      Monitoring::Event::ServerClosed.new(server.address, topology)
    )
  end
end

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

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 757

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

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

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

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

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

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

def max_read_retries
  options[:max_read_retries] || MAX_READ_RETRIES
end

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

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 808

def pool(server)
  server.pool
end

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

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

  if topology.server_hosts_match_any?(DOCUMENTDB_HOST_PATTERNS)
    log_info DOCUMENTDB_LOG_MESSAGE
    return
  end
end

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

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 394

def read_retry_interval
  options[:read_retry_interval] || READ_RETRY_INTERVAL
end

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

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
      if @srv_monitor
        @srv_monitor.run!
      end
    end
    @update_lock.synchronize do
      @connecting = false
      @connected = true
    end
  end
end

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

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 894

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|
        if delete
          removed_servers << server
        end
      end
    end
  end
  if disconnect != false
    removed_servers.each do |server|
      disconnect_server_if_connected(server)
    end
  end
  if disconnect != false
    removed_servers.any?
  else
    removed_servers
  end
end

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

def run_sdam_flow(previous_desc, updated_desc, options = {})
  if load_balanced?
    if updated_desc.config.empty?
      unless 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
    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

    unless options[:keep_connection_pool]
      if flow.became_unknown?
        servers_list.each do |server|
          if 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
      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.
  unless updated_desc.unknown?
    server_selection_semaphore.broadcast
  end
end

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

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 441

def servers
  topology.servers(servers_list)
end

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

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

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

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|
      unless servers_list.any? { |server| server.address.seed == address_str }
        add(address_str)
      end
    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 1019

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
      if @srv_monitor
        @srv_monitor.stop!
      end
    end
  end
end

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

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

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

def update_cluster_time(result)
  if cluster_time_doc = result.cluster_time
    @cluster_time_lock.synchronize do
      advance_cluster_time(cluster_time_doc)
    end
  end
end

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

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 988

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

  @state_change_lock.synchronize do
    @sdam_flow_lock.synchronize do
      if topology.data_bearing_servers?
        unless topology.logical_session_timeout
          raise_sessions_not_supported
        end
      end
    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
      unless topology.logical_session_timeout
        raise_sessions_not_supported
      end
    end
  end
end