File Coverage

blib/lib/Mango/Bulk.pm

Criterion	Covered	Total	%
statement	12	99	12.1
branch	0	32	0.0
condition	0	28	0.0
subroutine	4	21	19.0
pod	8	8	100.0
total	24	188	12.7

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);
    }
  );
}

sub _op {
  my ($self, $type, $doc) = @_;

  # Pre-encode documents
  my $mango     = $self->collection->db->mango;
  my $bson_max  = $mango->max_bson_size;
  my $batch_max = $mango->max_write_batch_size;
  my $ops       = $self->{ops} ||= [];
  my $previous  = @$ops ? $ops->[-1] : [];
  my $bson      = bson_encode $doc;
  my $size      = length $bson;
  my $new       = ($self->{size} // 0) + $size;
  my $limit     = $new > $bson_max || @$previous >= $batch_max + 2;

  # Group documents based on type and limits
  push @$ops, [$type, $self->{offset} || 0] and delete $self->{size}
    if !@$previous || $previous->[0] ne $type || $limit;
  push @{$ops->[-1]}, bson_raw $bson;
  $self->{size} += $size;
  $self->{offset}++;

  return $self;
}

sub _remove {
  my ($self, $limit) = @_;
  my $query = delete $self->{query} // {};
  return $self->_op(delete => {q => $query, limit => $limit});
}

sub _set {
  my ($self, $key, $value) = @_;
  $self->{$key} = $value;
  return $self;
}

sub _update {
  my ($self, $multi, $update) = @_;
  my $query = delete $self->{query} // {};
  my $upsert = delete $self->{upsert} ? \1 : \0;
  return $self->_op(
    update => {q => $query, u => $update, multi => $multi, upsert => $upsert});
}

1;

=encoding utf8

=head1 NAME

Mango::Bulk - MongoDB bulk operations

=head1 SYNOPSIS

  use Mango::Bulk;

  my $bulk = Mango::Bulk->new(collection => $collection);
  $bulk->insert({foo => 'bar'})->insert({foo => 'baz'})->execute;

=head1 DESCRIPTION

L is a container for MongoDB bulk operations, all operations will
be automatically grouped so they don't exceed L.

=head1 ATTRIBUTES

L implements the following attributes.

=head2 collection

  my $collection = $bulk->collection;
  $bulk          = $bulk->collection(Mango::Collection->new);

L object this bulk operation belongs to.

=head2 ordered

  my $bool = $bulk->ordered;
  $bulk    = $bulk->ordered($bool);

Bulk operations are ordered, defaults to a true value.

=head1 METHODS

L inherits all methods from L and implements the
following new ones.

=head2 execute

  my $results = $bulk->execute;

Execute bulk operations. You can also append a callback to perform operation
non-blocking.

  $bulk->execute(sub {
    my ($bulk, $err, $results) = @_;
    ...
  });
  Mojo::IOLoop->start unless Mojo::IOLoop->is_running;

=head2 find

  $bulk = $bulk->find({foo => 'bar'});

Query for next update or remove operation.

=head2 insert

  $bulk = $bulk->insert({foo => 'bar'});

Insert document.

=head2 remove

  $bulk = $bulk->remove;

Remove multiple documents.

=head2 remove_one

  $bulk = $bulk->remove_one;

Remove one document.

=head2 update

  $bulk = $bulk->update({foo => 'bar'});

Update multiple documents.

=head2 update_one

  $bulk = $bulk->update_one({foo => 'baz'});

Update one document.

=head2 upsert

  $bulk = $bulk->upsert;

Next update operation will be an C.

=head1 SEE ALSO

L, L, L.

=cut