Module: Mongo::Collection::View::Readable

# File 'lib/mongo/collection/view/readable.rb', line 80

def allow_disk_use
  configure(:allow_disk_use, true)
end

# File 'lib/mongo/collection/view/readable.rb', line 92

def allow_partial_results
  configure(:allow_partial_results, true)
end

# File 'lib/mongo/collection/view/readable.rb', line 104

def await_data
  configure(:await_data, true)
end

# File 'lib/mongo/collection/view/readable.rb', line 121

def batch_size(batch_size = nil)
  configure(:batch_size, batch_size)
end

# File 'lib/mongo/collection/view/readable.rb', line 752

def collation(doc = nil)
  configure(:collation, doc)
end

# File 'lib/mongo/collection/view/readable.rb', line 139

def comment(comment = nil)
  configure(:comment, comment)
end

# File 'lib/mongo/collection/view/readable.rb', line 175

def count(opts = {})
  opts = @options.merge(opts) unless Mongo.broken_view_options
  cmd = { :count => collection.name, :query => filter }
  cmd[:skip] = opts[:skip] if opts[:skip]
  cmd[:hint] = opts[:hint] if opts[:hint]
  cmd[:limit] = opts[:limit] if opts[:limit]
  if read_concern
    cmd[:readConcern] = Options::Mapper.transform_values_to_strings(
      read_concern)
  end
  cmd[:maxTimeMS] = opts[:max_time_ms] if opts[:max_time_ms]
  Mongo::Lint.validate_underscore_read_preference(opts[:read])
  read_pref = opts[:read] || read_preference
  selector = ServerSelector.get(read_pref || server_selector)
  with_session(opts) do |session|
    context = Operation::Context.new(
      client: client,
      session: session,
      operation_timeouts: operation_timeouts(opts)
    )
    read_with_retry(session, selector, context) do |server|
      Operation::Count.new(
        selector: cmd,
        db_name: database.name,
        options: {:limit => -1},
        read: read_pref,
        session: session,
        # For some reason collation was historically accepted as a
        # string key. Note that this isn't documented as valid usage.
        collation: opts[:collation] || opts['collation'] || collation,
        comment: opts[:comment],
      ).execute(
        server,
        context: context
      )
    end.n.to_i
  end
end

# File 'lib/mongo/collection/view/readable.rb', line 241

def count_documents(opts = {})
  opts = @options.merge(opts) unless Mongo.broken_view_options
  pipeline = [:'$match' => filter]
  pipeline << { :'$skip' => opts[:skip] } if opts[:skip]
  pipeline << { :'$limit' => opts[:limit] } if opts[:limit]
  pipeline << { :'$group' => { _id: 1, n: { :'$sum' => 1 } } }

  opts = opts.slice(:hint, :max_time_ms, :read, :collation, :session, :comment, :timeout_ms)
  opts[:collation] ||= collation

  first = aggregate(pipeline, opts).first
  return 0 unless first
  first['n'].to_i
end

# File 'lib/mongo/collection/view/readable.rb', line 660

def cursor_type(type = nil)
  configure(:cursor_type, type)
end

# File 'lib/mongo/collection/view/readable.rb', line 343

def distinct(field_name, opts = {})
  if field_name.nil?
    raise ArgumentError, 'Field name for distinct operation must be not nil'
  end
  opts = @options.merge(opts) unless Mongo.broken_view_options
  cmd = { :distinct => collection.name,
          :key => field_name.to_s,
          :query => filter, }
  cmd[:maxTimeMS] = opts[:max_time_ms] if opts[:max_time_ms]
  if read_concern
    cmd[:readConcern] = Options::Mapper.transform_values_to_strings(
      read_concern)
  end
  Mongo::Lint.validate_underscore_read_preference(opts[:read])
  read_pref = opts[:read] || read_preference
  selector = ServerSelector.get(read_pref || server_selector)
  with_session(opts) do |session|
    context = Operation::Context.new(
      client: client,
      session: session,
      operation_timeouts: operation_timeouts(opts)
    )
    read_with_retry(session, selector, context) do |server|
      Operation::Distinct.new(
        selector: cmd,
        db_name: database.name,
        options: {:limit => -1},
        read: read_pref,
        session: session,
        comment: opts[:comment],
        # For some reason collation was historically accepted as a
        # string key. Note that this isn't documented as valid usage.
        collation: opts[:collation] || opts['collation'] || collation,
      ).execute(
        server,
        context: context
      )
    end.first['values']
  end
end

# File 'lib/mongo/collection/view/readable.rb', line 276

def estimated_document_count(opts = {})
  unless view.filter.empty?
    raise ArgumentError, "Cannot call estimated_document_count when querying with a filter"
  end

  %i[limit skip].each do |opt|
    if options.key?(opt) || opts.key?(opt)
      raise ArgumentError, "Cannot call estimated_document_count when querying with #{opt}"
    end
  end

  opts = @options.merge(opts) unless Mongo.broken_view_options
  Mongo::Lint.validate_underscore_read_preference(opts[:read])
  read_pref = opts[:read] || read_preference
  selector = ServerSelector.get(read_pref || server_selector)
  with_session(opts) do |session|
    context = Operation::Context.new(
      client: client,
      session: session,
      operation_timeouts: operation_timeouts(opts)
    )
    read_with_retry(session, selector, context) do |server|
      cmd = { count: collection.name }
      cmd[:maxTimeMS] = opts[:max_time_ms] if opts[:max_time_ms]
      if read_concern
        cmd[:readConcern] = Options::Mapper.transform_values_to_strings(read_concern)
      end
      result = Operation::Count.new(
        selector: cmd,
        db_name: database.name,
        read: read_pref,
        session: session,
        comment: opts[:comment],
      ).execute(server, context: context)
      result.n.to_i
    end
  end
rescue Error::OperationFailure::Family => exc
  if exc.code == 26
    # NamespaceNotFound
    # This should only happen with the aggregation pipeline path
    # (server 4.9+). Previous servers should return 0 for nonexistent
    # collections.
    0
  else
    raise
  end
end

# File 'lib/mongo/collection/view/readable.rb', line 394

def hint(hint = nil)
  configure(:hint, hint)
end

# File 'lib/mongo/collection/view/readable.rb', line 408

def limit(limit = nil)
  configure(:limit, limit)
end

# File 'lib/mongo/collection/view/readable.rb', line 424

def map_reduce(map, reduce, options = {})
  MapReduce.new(self, map, reduce, @options.merge(options))
end

# File 'lib/mongo/collection/view/readable.rb', line 632

def max_await_time_ms(max = nil)
  configure(:max_await_time_ms, max)
end

# File 'lib/mongo/collection/view/readable.rb', line 441

def max_scan(value = nil)
  configure(:max_scan, value)
end

# File 'lib/mongo/collection/view/readable.rb', line 646

def max_time_ms(max = nil)
  configure(:max_time_ms, max)
end

# File 'lib/mongo/collection/view/readable.rb', line 455

def max_value(value = nil)
  configure(:max_value, value)
end

# File 'lib/mongo/collection/view/readable.rb', line 469

def min_value(value = nil)
  configure(:min_value, value)
end

# File 'lib/mongo/collection/view/readable.rb', line 613

def modifiers(doc = nil)
  if doc.nil?
    Operation::Find::Builder::Modifiers.map_server_modifiers(options)
  else
    new(options.merge(Operation::Find::Builder::Modifiers.map_driver_options(BSON::Document.new(doc))))
  end
end

# File 'lib/mongo/collection/view/readable.rb', line 482

def no_cursor_timeout
  configure(:no_cursor_timeout, true)
end

# File 'lib/mongo/collection/view/readable.rb', line 702

def parallel_scan(cursor_count, options = {})
  if options[:session]
    # The session would be overwritten by the one in +options+ later.
    session = client.get_session(@options)
  else
    session = nil
  end
  server = server_selector.select_server(cluster, nil, session)
  spec = {
    coll_name: collection.name,
    db_name: database.name,
    cursor_count: cursor_count,
    read_concern: read_concern,
    session: session,
  }.update(options)
  session = spec[:session]
  op = Operation::ParallelScan.new(spec)
  # Note that the context object shouldn't be reused for subsequent
  # GetMore operations.
  context = Operation::Context.new(client: client, session: session)
  result = op.execute(server, context: context)
  result.cursor_ids.map do |cursor_id|
    spec = {
      cursor_id: cursor_id,
      coll_name: collection.name,
      db_name: database.name,
      session: session,
      batch_size: batch_size,
      to_return: 0,
      # max_time_ms is not being passed here, I assume intentionally?
    }
    op = Operation::GetMore.new(spec)
    context = Operation::Context.new(
      client: client,
      session: session,
      connection_global_id: result.connection_global_id,
    )
    result = if server.load_balancer?
               # Connection will be checked in when cursor is drained.
               connection = server.pool.check_out(context: context)
               op.execute_with_connection(connection, context: context)
             else
               op.execute(server, context: context)
             end
    Cursor.new(self, result, server, session: session)
  end
end

# File 'lib/mongo/collection/view/readable.rb', line 500

def projection(document = nil)
  validate_doc!(document) if document
  configure(:projection, document)
end

# File 'lib/mongo/collection/view/readable.rb', line 516

def read(value = nil)
  return read_preference if value.nil?
  configure(:read, value)
end

# File 'lib/mongo/collection/view/readable.rb', line 674

def read_concern
  if options[:session] && options[:session].in_transaction?
    options[:session].send(:txn_read_concern) || collection.client.read_concern
  else
    collection.read_concern
  end
end

# File 'lib/mongo/collection/view/readable.rb', line 683

def read_preference
  @read_preference ||= begin
    # Operation read preference is always respected, and has the
    # highest priority. If we are in a transaction, we look at
    # transaction read preference and default to client, ignoring
    # collection read preference. If we are not in transaction we
    # look at collection read preference which defaults to client.
    rp = if options[:read]
      options[:read]
    elsif options[:session] && options[:session].in_transaction?
      options[:session].txn_read_preference || collection.client.read_preference
    else
      collection.read_preference
    end
    Lint.validate_underscore_read_preference(rp)
    rp
  end
end

# File 'lib/mongo/collection/view/readable.rb', line 531

def return_key(value = nil)
  configure(:return_key, value)
end

# File 'lib/mongo/collection/view/readable.rb', line 756

def server_selector
  @server_selector ||= if options[:session] && options[:session].in_transaction?
    ServerSelector.get(read_preference || client.server_selector)
  else
    ServerSelector.get(read_preference || collection.server_selector)
  end
end

# File 'lib/mongo/collection/view/readable.rb', line 546

def show_disk_loc(value = nil)
  configure(:show_disk_loc, value)
end

# File 'lib/mongo/collection/view/readable.rb', line 549

alias :show_record_id :show_disk_loc

# File 'lib/mongo/collection/view/readable.rb', line 562

def skip(number = nil)
  configure(:skip, number)
end

# File 'lib/mongo/collection/view/readable.rb', line 580

def snapshot(value = nil)
  configure(:snapshot, value)
end

# File 'lib/mongo/collection/view/readable.rb', line 595

def sort(spec = nil)
  configure(:sort, spec)
end

# File 'lib/mongo/collection/view/readable.rb', line 669

def timeout_ms(timeout_ms = nil)
  configure(:timeout_ms, timeout_ms)
end

# File 'lib/mongo/collection/view/readable.rb', line 764

def validate_doc!(doc)
  raise Error::InvalidDocument.new unless doc.respond_to?(:keys)
end