view release on metacpan or  search on metacpan

lib/Mango/Collection.pm  view on Meta::CPAN

  $self->_command($cmd, $cb, sub { shift->{cursor}->{firstBatch}->[0] });
}

sub remove {
  my $self  = shift;
  my $cb    = ref $_[-1] eq 'CODE' ? pop : undef;
  my $query = shift // {};
  my $flags = shift // {};

  ($query, $flags) = ({_id => $query}, {single => 1})
    if ref $query eq 'Mango::BSON::ObjectID';
  my $command = bson_doc
    delete       => $self->name,
    deletes      => [{q => $query, limit => $flags->{single} ? 1 : 0}],
    ordered      => \1,
    writeConcern => $self->db->build_write_concern;

  return $self->_command($command, $cb);
}

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

  my $admin = $self->db->mango->db('admin');
  my $dbname = $self->db->name;
  my $oldname = join '.', $dbname, $self->name;
  my $newname = join '.', $dbname, $name;

  my $cmd = bson_doc renameCollection => $oldname, to => $newname;

  # Non-blocking
  return $admin->command($cmd, sub {
    my ($admin_db, $err, $doc) = @_;
    my $newcol = $doc->{ok} ? $self->db->collection($name) : undef;
    return $cb->($self, $err, $newcol);
  }) if $cb;

  # Blocking
  my $doc = $admin->command($cmd);
  return $doc->{ok} ? $self->db->collection($name) : undef;
}

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

  # New document
  return $self->insert($doc, $cb) unless $doc->{_id};

  # Update non-blocking
  my @update = ({_id => $doc->{_id}}, $doc, {upsert => 1});
  return $self->update(@update => sub { shift->$cb(shift, $doc->{_id}) })
    if $cb;

  # Update blocking
  $self->update(@update);
  return $doc->{_id};
}

sub stats { $_[0]->_command(bson_doc(collstats => $_[0]->name), $_[1]) }

sub update {
  my ($self, $query, $update) = (shift, shift, shift);
  my $cb = ref $_[-1] eq 'CODE' ? pop : undef;
  my $flags = shift // {};

  $update = {
    q => ref $query eq 'Mango::BSON::ObjectID' ? {_id => $query} : $query,
    u => $update,
    upsert => $flags->{upsert} ? \1 : \0,
    multi  => $flags->{multi}  ? \1 : \0
  };
  my $command = bson_doc
    update       => $self->name,
    updates      => [$update],
    ordered      => \1,
    writeConcern => $self->db->build_write_concern;

  return $self->_command($command, $cb);
}

sub _aggregate {
  my ($self, $command, $doc) = @_;

  # Document (explain)
  return $doc if $command->{explain};

  # Collection
  my $out = $command->{pipeline}[-1]{'$out'};
  return $self->db->collection($out) if defined $out;

  # Cursor
  my $cursor = $doc->{cursor};
  return Mango::Cursor->new(collection => $self, id => $cursor->{id})
    ->add_batch($cursor->{firstBatch});
}

sub _command {
  my ($self, $command, $cb, $return) = @_;
  $return ||= sub {shift};

  # Non-blocking
  my $db       = $self->db;
  my $protocol = $db->mango->protocol;
  return $db->command(
    $command => sub {
      my ($db, $err, $doc) = @_;
      $err ||= $protocol->write_error($doc);
      $self->$cb($err, $return->($doc));
    }
  ) if $cb;

  # Blocking
  my $doc = $db->command($command);
  if (my $err = $protocol->write_error($doc)) { croak $err }
  return $return->($doc);
}

sub _map_reduce {
  my ($self, $doc) = @_;
  return $doc->{results} unless $doc->{result};
  return $self->db->collection($doc->{result});