Class: Mongo::Server::Monitor Private

# File 'lib/mongo/server/monitor.rb', line 108

attr_reader :options

# File 'lib/mongo/server/monitor.rb', line 350

def rtt_measurement_only?
  return false if connection.nil?

  # Only suppress the check while the server is in a known state and the
  # PushMonitor is the authoritative streaming source. If the server is
  # Unknown (e.g. an operation error or a streaming failure just marked
  # it so), the polling Monitor must run a full check to recover it
  # rather than waiting for the next streaming response - otherwise the
  # server can stay Unknown long enough to fail server selection.
  return false if server.unknown?

  pm = push_monitor
  !pm.nil? && pm.running?
end

# File 'lib/mongo/server/monitor.rb', line 98

attr_reader :server

# File 'lib/mongo/server/monitor.rb', line 451

def streaming_enabled?
  mode = options[:server_monitoring_mode] || :auto
  case mode
  when :poll
    false
  when :stream
    true
  when :auto
    !Server::AppMetadata::Environment.new.faas?
  end
end

# File 'lib/mongo/server/monitor.rb', line 365

def check
  # Snapshot the connection under the lock. A concurrent cancel_check!
  # may nil @connection from another thread; working on a local copy keeps
  # this check consistent, and the guarded writeback below never clobbers
  # a connection the monitor thread did not itself establish.
  connection = self.connection

  if connection && connection.pid != Process.pid
    log_warn("Detected PID change - Mongo client should have been reconnected (old pid #{connection.pid}, new pid #{Process.pid}")
    connection.disconnect!
    clear_connection(connection)
    connection = nil
  end

  if connection
    result = server.round_trip_time_calculator.measure do
      doc = connection.check_document
      cmd = Protocol::Query.new(
        Database::ADMIN, Database::COMMAND, doc, limit: -1
      )
      message = connection.dispatch_bytes(cmd.serialize.to_s)
      message.documents.first
    rescue Mongo::Error
      connection.disconnect!
      clear_connection(connection)
      raise
    end
  else
    connection = Connection.new(server.address, options)
    connection.connect!
    result = server.round_trip_time_calculator.measure do
      connection.handshake!
    end
    store_connection(connection)
    if (tv_doc = result['topologyVersion'])
      if streaming_enabled?
        # Run the instance we just created rather than re-reading the
        # push_monitor getter: a concurrent cancel_check! may have nil'd
        # @push_monitor between the two calls.
        push_monitor = create_push_monitor!(TopologyVersion.new(tv_doc), connection)
        push_monitor.run!
      else
        stop_push_monitor!
      end
    else
      # Failed response or pre-4.4 server
      stop_push_monitor!
    end
    result
  end
  result
end

# File 'lib/mongo/server/monitor.rb', line 428

def clear_connection(connection)
  @connection_lock.synchronize do
    @connection = nil if @connection.equal?(connection)
  end
end

# File 'lib/mongo/server/monitor.rb', line 103

def connection
  @connection_lock.synchronize { @connection }
end

# File 'lib/mongo/server/monitor.rb', line 175

def create_push_monitor!(topology_version, connection)
  @update_mutex.synchronize do
    @push_monitor = nil if @push_monitor && !@push_monitor.running?

    @push_monitor ||= PushMonitor.new(
      self,
      topology_version,
      monitoring,
      **Utils.shallow_symbolize_keys(options.merge(
                                       socket_timeout: heartbeat_interval + connection.socket_timeout,
                                       app_metadata: options[:push_monitor_app_metadata],
                                       check_document: connection.check_document
                                     ))
    )
  end
end

# File 'lib/mongo/server/monitor.rb', line 326

def do_scan(publish_heartbeat: true)
  if publish_heartbeat
    monitoring.publish_heartbeat(server) do
      check
    end
  else
    check
  end
rescue StandardError => e
  msg = "Error checking #{server.address}"
  Utils.warn_bg_exception(msg, e,
                          logger: options[:logger],
                          log_prefix: options[:log_prefix],
                          bg_error_backtrace: options[:bg_error_backtrace])
  raise e
end

# File 'lib/mongo/server/monitor.rb', line 140

def do_work
  scan!
  # @next_wanted_scan may be updated by the push monitor.
  # However we need to check for termination flag so that the monitor
  # thread exits when requested.
  loop do
    delta = @next_wanted_scan - Time.now
    break unless delta > 0

    signaled = server.scan_semaphore.wait(delta)
    break if signaled || @stop_requested
  end
end

# File 'lib/mongo/server/monitor.rb', line 113

def heartbeat_interval
  options[:heartbeat_interval] || DEFAULT_HEARTBEAT_INTERVAL
end

# File 'lib/mongo/server/monitor.rb', line 322

def pre_stop
  server.scan_semaphore.signal
end

# File 'lib/mongo/server/monitor.rb', line 131

def push_monitor
  @update_mutex.synchronize do
    @push_monitor
  end
end

# File 'lib/mongo/server/monitor.rb', line 308

def restart!
  if @thread && @thread.alive?
    @thread
  else
    run!
  end
end

# File 'lib/mongo/server/monitor.rb', line 268

def run_sdam_flow(result, awaited: false, scan_error: nil, rtt_only: false)
  @sdam_mutex.synchronize do
    old_description = server.description

    # An RTT-only measurement (streaming protocol active) must not update
    # the topology or publish SDAM events. The RTT it gathered is
    # incorporated into the next streaming-hello description via the
    # shared RTT calculator. The scheduling below still runs so the
    # monitor keeps pacing its checks.
    unless rtt_only
      new_description = Description.new(
        server.address,
        result,
        average_round_trip_time: server.round_trip_time_calculator.average_round_trip_time,
        minimum_round_trip_time: server.round_trip_time_calculator.minimum_round_trip_time
      )
      server.cluster.run_sdam_flow(server.description, new_description, awaited: awaited, scan_error: scan_error)
    end

    server.description.tap do |new_description|
      unless awaited
        if new_description.unknown? && !old_description.unknown?
          @next_earliest_scan = @next_wanted_scan = Time.now
        else
          @next_earliest_scan = Time.now + MIN_SCAN_INTERVAL
          @next_wanted_scan = Time.now + heartbeat_interval
        end
      end
    end
  end
end

# File 'lib/mongo/server/monitor.rb', line 244

def scan!
  # Ordinarily the background thread would invoke this method.
  # But it is also possible to invoke scan! directly on a monitor.
  # Allow only one scan to be performed at a time.
  @mutex.synchronize do
    throttle_scan_frequency!

    # When the streaming protocol is active the PushMonitor is the
    # authoritative SDAM source and this scan only measures RTT on the
    # dedicated connection. Per the Server Monitoring spec, an RTT
    # command MUST NOT publish events or update the topology. Compute
    # this before do_scan, which may (re)connect and change the state.
    rtt_only = rtt_measurement_only?

    begin
      result = do_scan(publish_heartbeat: !rtt_only)
    rescue StandardError => e
      run_sdam_flow({}, scan_error: e, rtt_only: rtt_only)
    else
      run_sdam_flow(result, rtt_only: rtt_only)
    end
  end
end

# File 'lib/mongo/server/monitor.rb', line 160

def stop!
  stop_push_monitor!

  # Forward super's return value
  super.tap do
    # Important: disconnect should happen after the background thread
    # terminates.
    connection&.disconnect!
  end
end

# File 'lib/mongo/server/monitor.rb', line 192

def stop_push_monitor!
  @update_mutex.synchronize do
    if @push_monitor
      @push_monitor.stop!
      @push_monitor = nil
    end
  end
end

# File 'lib/mongo/server/monitor.rb', line 419

def store_connection(connection)
  @connection_lock.synchronize do
    @connection = connection
  end
end

# File 'lib/mongo/server/monitor.rb', line 436

def throttle_scan_frequency!
  delta = @next_earliest_scan - Time.now
  return unless delta > 0

  sleep(delta)
end

# File 'lib/mongo/server/monitor.rb', line 316

def to_s
  "#<#{self.class.name}:#{object_id} #{server.address}>"
end