view release on metacpan or  search on metacpan

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

sub ensure_indexes {
    state $check = compile(Object);
    my ($self) = $check->(@_);
    state $aoa_check = compile( slurpy ArrayRef [ArrayRef] );
    my ($aoa) = $aoa_check->( $self->class->_indexes );
    my $index_view = $self->_mongo_collection->indexes;
    my @indexes;
    for my $index (@$aoa) {
        my @copy = @$index;
        my $options = ref $copy[0] eq 'HASH' ? shift @copy : undef;
        if ( @copy % 2 != 0 ) {
            $self->_croak(
                "_indexes must provide a list of key/value pairs, with an optional leading hashref");
        }
        my $spec = Tie::IxHash->new(@copy);
        push @indexes, { keys => $spec, ( $options ? ( options => $options ) : () ) };
    }
    $self->_try_mongo_op( ensure_indexes => sub { $index_view->create_many(@indexes) } );
    return 1;
}

#--------------------------------------------------------------------------#
# Semi-private methods on individual objects; typically called by object to
# modify itself and synchronize with the database
#--------------------------------------------------------------------------#

sub remove {
    state $check = compile( Object, Object );
    my ( $self, $obj ) = $check->(@_);
    $self->_try_mongo_op(
        remove => sub { $self->_mongo_collection->delete_one( { _id => $obj->_id } ) } );
    $obj->_set_removed(1);
    return 1;
}

sub reinsert {
    state $check = compile( Object, Object );
    my ( $self, $obj ) = $check->(@_);
    $self->_save($obj);
    $obj->_set_removed(0);
    return 1;
}

sub sync {
    state $check = compile( Object, Object );
    my ( $self, $obj ) = $check->(@_);
    my $data = $self->_try_mongo_op(
        sync => sub { $self->_mongo_collection->find_one( { _id => $obj->_id } ) } );
    if ($data) {
        $self->_sync( $data => $obj );
        $obj->_set_removed(0);
        return 1;
    }
    else {
        $obj->_set_removed(1);
        return; # false means removed
    }
}

sub update {
    state $check = compile( Object, Object, HashRef );
    my ( $self, $obj, $update ) = $check->(@_);
    my $data = $self->_try_mongo_op(
        update => sub {
            $self->_mongo_collection->find_one_and_update( { _id => $obj->_id },
                $update, { returnDocument => "after" } );
        },
    );

    if ( ref $data ) {
        $self->_sync( $data => $obj );
        return 1;
    }
    else {
        $obj->_set_removed(1);
        return; # false means removed
    }
}

sub thaw_object {
    state $check = compile( Object, HashRef );
    my ( $self, $data ) = $check->(@_);
    $data->{__CLASS__}   = $self->class;
    $data->{_collection} = $self;
    return $self->class->unpack($data);
}

#--------------------------------------------------------------------------#
# Private methods
#--------------------------------------------------------------------------#

sub _mongo_collection {
    state $check = compile(Object);
    my ($self) = $check->(@_);
    return $self->meerkat->mongo_collection( $self->collection_name );
}

sub _try_mongo_op {
    state $check = compile( Object, Str, CodeRef );
    my ( $self, $action, $code, $rest ) = $check->(@_);
    # call &retry to bypass prototype
    return &retry(
        $code, @$rest,
        retry_if { /not connected/ },
        delay_exp { 5, 1e6 },
        on_retry { $self->mongo_clear_caches },
        catch { croak "$action error: $_" }
    );
}

sub _save {
    state $check = compile( Object, Object );
    my ( $self, $obj ) = $check->(@_);
    my $pack = $obj->pack;
    delete $pack->{$_} for qw/__CLASS__ _collection _removed/;
    return $self->_try_mongo_op(
        sync => sub {
            !!$self->_mongo_collection->replace_one( { _id => $pack->{_id} },
                $pack, { upsert => 1 } );
        }