<commit_before>package Mango::Cursor;
use Mojo::Base -base;

use Mango::BSON 'bson_doc';
use Mojo::IOLoop;

has [qw(batch_size limit skip)] => 0;
has [qw(collection hint id snapshot sort tailable)];
has [qw(fields query)] => sub { {} };

sub all {
  my ($self, $cb) = @_;

  # Non-blocking
  my @all;
  return $self->next(sub { shift->_collect(\@all, $cb, @_) }) if $cb;

  # Blocking
  while (my $next = $self->next) { push @all, $next }
  return \@all;
}

sub build_query {
  my ($self, $explain) = @_;

  my $query    = $self->query;
  my $sort     = $self->sort;
  my $hint     = $self->hint;
  my $snapshot = $self->snapshot;

  return $query unless $snapshot || $hint || $sort || $explain;

  $query = {'$query' => $query};
  $query->{'$explain'}  = 1     if $explain;
  $query->{'$orderby'}  = $sort if $sort;
  $query->{'$hint'}     = $hint if $hint;
  $query->{'$snapshot'} = 1     if $snapshot;

  return $query;
}

sub clone {
  my $self  = shift;
  my $clone = $self->new;
  $clone->$_($self->$_)
    for qw(batch_size collection fields hint limit query skip snapshot),
    qw(sort tailable);
  return $clone;
}

sub count {
  my $self = shift;
  my $cb = ref $_[-1] eq 'CODE' ? pop : undef;

  my $collection = $self->collection;
  my $count      = bson_doc
    count => $collection->name,
    query => $self->build_query,
    skip  => $self->skip,
    limit => $self->limit;

  # Non-blocking
  return $collection->db->command(
    $count => sub {
      my ($collection, $err, $doc) = @_;
      $self->$cb($err, $doc ? $doc->{n} : 0);
    }
  ) if $cb;

  # Blocking
  my $doc = $collection->db->command($count);
  return $doc ? $doc->{n} : 0;
}

sub distinct {
  my ($self, $key) = (shift, shift);
  my $cb = ref $_[-1] eq 'CODE' ? pop : undef;

  my $collection = $self->collection;
  my $distinct   = bson_doc
    distinct => $collection->name,
    key      => $key,
    query    => $self->build_query;

  # Non-blocking
  return $collection->db->command(
    $distinct => sub { shift; $self->$cb(shift, shift->{values}) })
    if $cb;

  # Blocking
  return $collection->db->command($distinct)->{values};
}

sub explain {
  my ($self, $cb) = @_;

  # Non-blocking
  my $clone = $self->clone->query($self->build_query(1))->sort(undef);
  return $clone->next(sub { shift; $self->$cb(@_) }) if $cb;

  # Blocking
  return $clone->next;
}

sub next {
  my ($self, $cb) = @_;
  return exists $self->{results} ? $self->_continue($cb) : $self->_start($cb);
}

sub rewind {
  my ($self, $cb) = @_;

  delete $self->{$_} for qw(num results);
  return $cb ? $self->_defer($cb) : undef unless my $id = $self->id;
  $self->id(undef);

  # Non-blocking
  my $mango = $self->collection->db->mango;
  return $mango->kill_cursors($id => sub { shift; $self->$cb(@_) }) if $cb;

  # Blocking
  $mango->kill_cursors($id);
}

sub _collect {
  my ($self, $all, $cb, $err, $doc) = @_;
  return $self->_defer($cb, $err, $all) if $err || !$doc;
  push @$all, $doc;
  $self->next(sub { shift->_collect($all, $cb, @_) });
}

sub _continue {
  my ($self, $cb) = @_;

  my $collection = $self->collection;
  my $name       = $collection->full_name;
  my $mango      = $collection->db->mango;

  # Non-blocking
  if ($cb) {
    return $self->_defer($cb, undef, $self->_dequeue) if $self->_enough;
    return $mango->get_more(($name, $self->_max, $self->id) =>
        sub { shift; $self->$cb(shift, $self->_enqueue(shift)) });
  }

  # Blocking
  return $self->_dequeue if $self->_enough;
  return $self->_enqueue($mango->get_more($name, $self->_max, $self->id));
}

sub _defer {
  my ($self, $cb, @args) = @_;
  Mojo::IOLoop->timer(0 => sub { $self->$cb(@args) });
}

sub _dequeue {
  my $self = shift;
  return undef if $self->_finished;
  $self->{num}++;
  return shift @{$self->{results}};
}

sub _enough { $_[0]->_finished ? 1 : !!@{$_[0]{results}} }

sub _finished {
  my $self = shift;
  return undef unless my $limit = $self->limit;
  $limit = $limit * -1 if $limit < 0;
  return ($self->{num} // 0) >= $limit ? 1 : undef;
}

sub _enqueue {
  my ($self, $reply) = @_;
  return unless $reply;
  push @{$self->{results} ||= []}, @{$reply->{docs}};
  return $self->id($reply->{cursor})->_dequeue;
}

sub _max {
  my $self  = shift;
  my $limit = $self->limit;
  my $size  = $self->batch_size;
  return $size if $limit == 0;
  return $size > $limit ? $limit : $size;
}

sub _start {
  my ($self, $cb) = @_;

  my $collection = $self->collection;
  my $name       = $collection->full_name;
  my $flags = $self->tailable ? {tailable_cursor => 1, await_data => 1} : {};
  my @args  = (
    $name, $flags, $self->skip, $self->_max, $self->build_query, $self->fields
  );

  # Non-blocking
  return $collection->db->mango->query(
    @args => sub { shift; $self->$cb(shift, $self->_enqueue(shift)) })
    if $cb;

  # Blocking
  my $reply = $collection->db->mango->query(@args);<commit_msg>flag slave_ok by default<commit_after>package Mango::Cursor;
use Mojo::Base -base;

use Mango::BSON 'bson_doc';
use Mojo::IOLoop;

has [qw(batch_size limit skip)] => 0;
has [qw(collection hint id snapshot sort tailable)];
has [qw(fields query)] => sub { {} };

sub all {
  my ($self, $cb) = @_;

  # Non-blocking
  my @all;
  return $self->next(sub { shift->_collect(\@all, $cb, @_) }) if $cb;

  # Blocking
  while (my $next = $self->next) { push @all, $next }
  return \@all;
}

sub build_query {
  my ($self, $explain) = @_;

  my $query    = $self->query;
  my $sort     = $self->sort;
  my $hint     = $self->hint;
  my $snapshot = $self->snapshot;

  return $query unless $snapshot || $hint || $sort || $explain;

  $query = {'$query' => $query};
  $query->{'$explain'}  = 1     if $explain;
  $query->{'$orderby'}  = $sort if $sort;
  $query->{'$hint'}     = $hint if $hint;
  $query->{'$snapshot'} = 1     if $snapshot;

  return $query;
}

sub clone {
  my $self  = shift;
  my $clone = $self->new;
  $clone->$_($self->$_)
    for qw(batch_size collection fields hint limit query skip snapshot),
    qw(sort tailable);
  return $clone;
}

sub count {
  my $self = shift;
  my $cb = ref $_[-1] eq 'CODE' ? pop : undef;

  my $collection = $self->collection;
  my $count      = bson_doc
    count => $collection->name,
    query => $self->build_query,
    skip  => $self->skip,
    limit => $self->limit;

  # Non-blocking
  return $collection->db->command(
    $count => sub {
      my ($collection, $err, $doc) = @_;
      $self->$cb($err, $doc ? $doc->{n} : 0);
    }
  ) if $cb;

  # Blocking
  my $doc = $collection->db->command($count);
  return $doc ? $doc->{n} : 0;
}

sub distinct {
  my ($self, $key) = (shift, shift);
  my $cb = ref $_[-1] eq 'CODE' ? pop : undef;

  my $collection = $self->collection;
  my $distinct   = bson_doc
    distinct => $collection->name,
    key      => $key,
    query    => $self->build_query;

  # Non-blocking
  return $collection->db->command(
    $distinct => sub { shift; $self->$cb(shift, shift->{values}) })
    if $cb;

  # Blocking
  return $collection->db->command($distinct)->{values};
}

sub explain {
  my ($self, $cb) = @_;

  # Non-blocking
  my $clone = $self->clone->query($self->build_query(1))->sort(undef);
  return $clone->next(sub { shift; $self->$cb(@_) }) if $cb;

  # Blocking
  return $clone->next;
}

sub next {
  my ($self, $cb) = @_;
  return exists $self->{results} ? $self->_continue($cb) : $self->_start($cb);
}

sub rewind {
  my ($self, $cb) = @_;

  delete $self->{$_} for qw(num results);
  return $cb ? $self->_defer($cb) : undef unless my $id = $self->id;
  $self->id(undef);

  # Non-blocking
  my $mango = $self->collection->db->mango;
  return $mango->kill_cursors($id => sub { shift; $self->$cb(@_) }) if $cb;

  # Blocking
  $mango->kill_cursors($id);
}

sub _collect {
  my ($self, $all, $cb, $err, $doc) = @_;
  return $self->_defer($cb, $err, $all) if $err || !$doc;
  push @$all, $doc;
  $self->next(sub { shift->_collect($all, $cb, @_) });
}

sub _continue {
  my ($self, $cb) = @_;

  my $collection = $self->collection;
  my $name       = $collection->full_name;
  my $mango      = $collection->db->mango;

  # Non-blocking
  if ($cb) {
    return $self->_defer($cb, undef, $self->_dequeue) if $self->_enough;
    return $mango->get_more(($name, $self->_max, $self->id) =>
        sub { shift; $self->$cb(shift, $self->_enqueue(shift)) });
  }

  # Blocking
  return $self->_dequeue if $self->_enough;
  return $self->_enqueue($mango->get_more($name, $self->_max, $self->id));
}

sub _defer {
  my ($self, $cb, @args) = @_;
  Mojo::IOLoop->timer(0 => sub { $self->$cb(@args) });
}

sub _dequeue {
  my $self = shift;
  return undef if $self->_finished;
  $self->{num}++;
  return shift @{$self->{results}};
}

sub _enough { $_[0]->_finished ? 1 : !!@{$_[0]{results}} }

sub _finished {
  my $self = shift;
  return undef unless my $limit = $self->limit;
  $limit = $limit * -1 if $limit < 0;
  return ($self->{num} // 0) >= $limit ? 1 : undef;
}

sub _enqueue {
  my ($self, $reply) = @_;
  return unless $reply;
  push @{$self->{results} ||= []}, @{$reply->{docs}};
  return $self->id($reply->{cursor})->_dequeue;
}

sub _max {
  my $self  = shift;
  my $limit = $self->limit;
  my $size  = $self->batch_size;
  return $size if $limit == 0;
  return $size > $limit ? $limit : $size;
}

sub _start {
  my ($self, $cb) = @_;

  my $collection = $self->collection;
  my $name       = $collection->full_name;
  my $flags = $self->tailable ? {tailable_cursor => 1, await_data => 1} : {};
  $flags->{ slave_ok } = 1;
  my @args  = (
    $name, $flags, $self->skip, $self->_max, $self->build_query, $self->fields
  );

  # Non-blocking
  return $collection->db->mango->query(
    @args => sub { shift; $self->$cb(shift, $self->_enqueue(shift)) })
    if $cb;

  # Blocking
  my $reply = $collection->db->mango->query(@args);
{'fraction_non_alphanumeric': 0.21510974987238388, 'fraction_numerical': 0.003981623277182236, 'mean_word_length': 3.2006861063464838, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 258, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '69900', 'n_tokens_mistral': 3824, 'n_tokens_neox': 3728, 'n_words': 909}