Module: Mongo::Collection::View::Iterable

# File 'lib/mongo/collection/view/iterable.rb', line 183

def use_query_cache?
  QueryCache.enabled? && !collection.system_collection?
end

# File 'lib/mongo/collection/view/iterable.rb', line 110

def cached_cursor
  QueryCache.get(**cache_options)
end

# File 'lib/mongo/collection/view/iterable.rb', line 72

def close_query
  return unless @cursor

  @cursor.close
end

# File 'lib/mongo/collection/view/iterable.rb', line 209

def compute_limit_for_cached_query
  return nil unless use_query_cache? && respond_to?(:limit)

  # If a query with a limit is performed, the query cache will
  # re-use results from an earlier query with the same or larger
  # limit, and then impose the lower limit during iteration.
  QueryCache.normalized_limit(limit)
end

# File 'lib/mongo/collection/view/iterable.rb', line 41

def each(&block)
  @cursor = prefer_cached_cursor? ? cached_cursor : new_cursor_for_iteration
  return @cursor.to_enum unless block_given?

  limit_for_cached_query = compute_limit_for_cached_query

  # TODO: rather than pulling the entire (limited) result set into
  # memory, this should tell the cursor about the limit and then let
  # the cursor limit the iteration as necessary.
  cursor_to_iterate = if limit_for_cached_query
                        @cursor.to_a[0...limit_for_cached_query]
                      else
                        @cursor
                      end

  cursor_to_iterate.each(&block)
end

# File 'lib/mongo/collection/view/iterable.rb', line 130

def initial_query_op(session)
  spec = {
    coll_name: collection.name,
    filter: filter,
    projection: projection,
    db_name: database.name,
    session: session,
    collation: collation,
    sort: sort,
    skip: skip,
    let: options[:let],
    limit: limit,
    allow_disk_use: options[:allow_disk_use],
    allow_partial_results: options[:allow_partial_results],
    read: read,
    read_concern: options[:read_concern] || read_concern,
    batch_size: batch_size,
    hint: options[:hint],
    max_scan: options[:max_scan],
    max_value: options[:max_value],
    min_value: options[:min_value],
    no_cursor_timeout: options[:no_cursor_timeout],
    return_key: options[:return_key],
    show_disk_loc: options[:show_disk_loc],
    comment: options[:comment],
    oplog_replay: oplog_replay
  }

  if spec[:oplog_replay]
    collection.client.log_warn('The :oplog_replay option is deprecated and ignored by MongoDB 4.4 and later')
  end

  maybe_set_tailable_options(spec)

  if explained?
    spec[:explain] = options[:explain]
    Operation::Explain.new(spec)
  else
    Operation::Find.new(spec)
  end
end

# File 'lib/mongo/collection/view/iterable.rb', line 77

alias kill_cursors close_query

# File 'lib/mongo/collection/view/iterable.rb', line 221

def maybe_set_tailable_options(spec)
  case cursor_type
  when :tailable
    spec[:tailable] = true
  when :tailable_await
    spec[:tailable] = true
    spec[:await_data] = true
  end
end

# File 'lib/mongo/collection/view/iterable.rb', line 198

def new_cursor_for_iteration
  session = client.get_session(@options)
  select_cursor(session).tap do |cursor|
    if use_query_cache?
      # No need to store the cursor in the query cache if there is
      # already a cached cursor stored at this key.
      QueryCache.set(cursor, **cache_options)
    end
  end
end

# File 'lib/mongo/collection/view/iterable.rb', line 233

def oplog_replay
  v = options[:oplog_replay]
  v.nil? ? collection.options[:oplog_replay] : v
end

# File 'lib/mongo/collection/view/iterable.rb', line 81

def select_cursor(session)
  context = Operation::Context.new(
    client: client,
    session: session,
    operation_timeouts: operation_timeouts,
    view: self
  )
  op = initial_query_op(session)
  tracer.trace_operation(op, context) do
    if respond_to?(:write?, true) && write?
      retry_enabled = collection.client.options[:retry_writes] != false
      with_overload_retry(context: context, retry_enabled: retry_enabled) do
        server = server_selector.select_server(cluster, nil, session, write_aggregation: true)
        result = send_initial_query(server, context, operation: op)

        if use_query_cache?
          CachingCursor.new(view, result, server, session: session, context: context)
        else
          Cursor.new(view, result, server, session: session, context: context)
        end
      end
    else
      read_with_retry_cursor(session, server_selector, view, context: context) do |server|
        send_initial_query(server, context, operation: op)
      end
    end
  end
end

# File 'lib/mongo/collection/view/iterable.rb', line 172

def send_initial_query(server, context, operation: nil)
  operation ||= initial_query_op(context.session)
  if server.load_balancer?
    # Connection will be checked in when cursor is drained.
    connection = server.pool.check_out(context: context)
    operation.execute_with_connection(connection, context: context)
  else
    operation.execute(server, context: context)
  end
end