Class: Mongo::Server::Monitor Private

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

attr_reader :monitoring

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

attr_reader :options

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

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 93

attr_reader :server

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

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 162

def create_push_monitor!(topology_version)
  @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 288

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 131

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 104

def heartbeat_interval
  options[:heartbeat_interval] || DEFAULT_HEARTBEAT_INTERVAL
end

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

def pre_stop
  server.scan_semaphore.signal
end

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

def push_monitor
  @update_mutex.synchronize do
    @push_monitor
  end
end

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

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

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

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 206

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 151

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 179

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 372

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

  sleep(delta)
end

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

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