view release on metacpan or  search on metacpan

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

package Mango::Bulk;
use Mojo::Base -base;

use Carp 'croak';
use Mango::BSON qw(bson_doc bson_encode bson_oid bson_raw);
use Mojo::IOLoop;

has 'collection';
has ordered => 1;

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

  # Full results shared with all operations
  my $full = {upserted => [], writeConcernErrors => [], writeErrors => []};
  $full->{$_} = 0 for qw(nInserted nMatched nModified nRemoved nUpserted);

  # Non-blocking
  if ($cb) {
    return Mojo::IOLoop->next_tick(sub { shift; $self->$cb(undef, $full) })
      unless my $group = shift @{$self->{ops}};
    return $self->_next($group, $full, $cb);
  }

  # Blocking
  my $db       = $self->collection->db;
  my $protocol = $db->mango->protocol;
  while (my $group = shift @{$self->{ops}}) {
    my ($type, $offset, $command) = $self->_group($group);
    _merge($type, $offset, $full, $db->command($command));
    if (my $err = $protocol->write_error($full)) { croak $err }
  }

  return $full;
}

sub find { shift->_set(query => shift) }

sub insert {
  my ($self, $doc) = @_;
  $doc->{_id} //= bson_oid;
  return $self->_op(insert => $doc);
}

sub remove     { shift->_remove(0) }
sub remove_one { shift->_remove(1) }

sub update     { shift->_update(\1, @_) }
sub update_one { shift->_update(\0, @_) }

sub upsert { shift->_set(upsert => 1) }

sub _group {
  my ($self, $group) = @_;

  my ($type, $offset) = splice @$group, 0, 2;
  my $collection = $self->collection;
  return $type, $offset, bson_doc $type => $collection->name,
    $type eq 'insert' ? 'documents' : "${type}s" => $group,
    ordered => $self->ordered ? \1 : \0,
    writeConcern => $collection->db->build_write_concern;
}

sub _merge {
  my ($type, $offset, $full, $result) = @_;

  # Insert
  if ($type eq 'insert') { $full->{nInserted} += $result->{n} }

  # Update
  elsif ($type eq 'update') {
    $full->{nModified} += $result->{n};

    # Upsert
    if (my $upserted = $result->{upserted}) {
      push @{$full->{upserted}}, map { $_->{index} += $offset; $_ } @$upserted;
      $full->{nUpserted} += @$upserted;
      $full->{nMatched}  += $result->{n} - @$upserted;
    }

    else { $full->{nMatched} += $result->{n} }
  }

  # Delete
  elsif ($type eq 'delete') { $full->{nRemoved} += $result->{n} }

  # Errors
  push @{$full->{writeConcernErrors}}, $result->{writeConcernError}
    if $result->{writeConcernError};
  push @{$full->{writeErrors}},
    map { $_->{index} += $offset; $_ } @{$result->{writeErrors}};
}

sub _next {
  my ($self, $group, $full, $cb) = @_;

  my ($type, $offset, $command) = $self->_group($group);
  $self->collection->db->command(
    $command => sub {
      my ($db, $err, $result) = @_;

      _merge($type, $offset, $full, $result) unless $err;
      $err ||= $self->collection->db->mango->protocol->write_error($full);
      return $self->$cb($err, $full) if $err;

      return $self->$cb(undef, $full) unless my $next = shift @{$self->{ops}};
      $self->_next($next, $full, $cb);
    }
  );