Class: Mongo::Server::PushMonitor Private

# File 'lib/mongo/server/push_monitor.rb', line 61

attr_reader :monitoring

# File 'lib/mongo/server/push_monitor.rb', line 64

attr_reader :options

# File 'lib/mongo/server/push_monitor.rb', line 58

attr_reader :topology_version

# File 'lib/mongo/server/push_monitor.rb', line 94

def do_work
  @lock.synchronize do
    return if @stop_requested
  end

  result = monitoring.publish_heartbeat(server, awaited: true) do
    check
  end
  new_description = monitor.run_sdam_flow(result, awaited: true)
  # When hello fails due to a fail point, the response does not
  # include topology version. In this case we need to keep our existing
  # topology version so that we can resume monitoring.
  # The spec does not appear to directly address this case but
  # https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-monitoring.rst#streaming-ismaster
  # says that topologyVersion should only be updated from successful
  # hello responses.
  if new_description.topology_version
    @topology_version = new_description.topology_version
  end
rescue IOError, SocketError, SystemCallError, Mongo::Error => exc
  stop_requested = @lock.synchronize { @stop_requested }
  if stop_requested
    # Ignore the exception, see RUBY-2771.
    return
  end

  msg = "Error running awaited hello on #{server.address}"
  Utils.warn_bg_exception(msg, exc,
    logger: options[:logger],
    log_prefix: options[:log_prefix],
    bg_error_backtrace: options[:bg_error_backtrace],
  )

  # If a request failed on a connection, stop push monitoring.
  # In case the server is dead we don't want to have two connections
  # trying to connect unsuccessfully at the same time.
  stop!

  # Request an immediate check on the monitor to get reinstated as
  # soon as possible in case the server is actually alive.
  server.scan_semaphore.signal
end

# File 'lib/mongo/server/push_monitor.rb', line 184

def read_response
  if timeout = options[:connect_timeout]
    if timeout < 0
      raise Mongo::SocketTimeoutError, "Requested to read with a negative timeout: #{}"
    elsif timeout > 0
      timeout += options[:heartbeat_frequency] || Monitor::DEFAULT_HEARTBEAT_INTERVAL
    end
  end
  # We set the timeout twice: once passed into read_socket which applies
  # to each individual read operation, and again around the entire read.
  Timeout.timeout(timeout, Error::SocketTimeoutError, "Failed to read an awaited hello response in #{timeout} seconds") do
    @lock.synchronize { @connection }.read_response(socket_timeout: timeout)
  end
end

# File 'lib/mongo/server/push_monitor.rb', line 69

def start!
  @lock.synchronize do
    super
  end
end

# File 'lib/mongo/server/push_monitor.rb', line 75

def stop!
  @lock.synchronize do
    @stop_requested = true
    if @connection
      # Interrupt any in-progress exhausted hello reads by
      # disconnecting the connection.
      @connection.send(:socket).close rescue nil
    end
  end
  super.tap do
    @lock.synchronize do
      if @connection
        @connection.disconnect!
        @connection = nil
      end
    end
  end
end

# File 'lib/mongo/server/push_monitor.rb', line 199

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

# File 'lib/mongo/server/push_monitor.rb', line 173

def write_check_command
  document = @check_document.merge(
    topologyVersion: topology_version.to_doc,
    maxAwaitTimeMS: monitor.heartbeat_interval * 1000,
  )
  command = Protocol::Msg.new(
    [:exhaust_allowed], {}, document.merge({'$db' => Database::ADMIN})
  )
  @lock.synchronize { @connection }.write_bytes(command.serialize.to_s)
end