File Coverage

blib/lib/ElasticSearchX/UniqueKey.pm

Criterion	Covered	Total	%
statement	12	103	11.6
branch	0	38	0.0
condition	0	15	0.0
subroutine	4	22	18.1
pod	12	13	92.3
total	28	191	14.6

package ElasticSearchX::UniqueKey;
$ElasticSearchX::UniqueKey::VERSION = '0.05';
use strict;
use warnings;
use Carp;

#===================================
sub new {
#===================================
    my $class  = shift;
    my %params = (
        index => 'unique_key',
        ref $_[0] ? %{ shift() } : @_
    );
    my $self = bless {}, $class;
    for (qw(index es)) {
        $self->{"_$_"} = $params{$_}
            or croak "Missing required param $_";
    }
    return $self;
}

#===================================
sub create {
#===================================
    my $self = shift;
    my %params = $self->_params( 'create', @_ );

    eval {
        $self->es->create( %params, data => {} );
        1;
    }
        && return 1;
    return 0
        if $@->isa('Search::Elasticsearch::Error::Conflict')
        || $@->isa('ElasticSearch::Error::Conflict');
    croak $@;
}

#===================================
sub delete {
#===================================
    my $self = shift;
    my %params = $self->_params( 'delete', @_ );
    $self->es->delete( %params, ignore_missing => 1 );
}

#===================================
sub exists {
#===================================
    my $self = shift;
    my %params = $self->_params( 'exists', @_ );
    $self->es->exists(%params);
}

#===================================
sub update {
#===================================
    my $self   = shift;
    my %params = $self->_params( 'update', shift(), shift() );
    my $new_id = shift();
    croak "No new id passed to update()"
        unless defined $new_id and length $new_id;

    my ( $type, $old_id ) = @params{ 'type', 'id' };
    return 1 if $new_id eq $old_id;
    return unless $self->create( $type, $new_id );
    $self->delete( $type, $old_id );
    1;

}

#===================================
sub multi_create {
#===================================
    my ( $self, %keys ) = @_;

    my @docs = map { { type => $_, id => $keys{$_}, data => {} } } keys %keys;

    my %failed;
    $self->es->bulk_create(
        index       => $self->index,
        docs        => \@docs,
        on_conflict => sub {
            my ( $action, $doc ) = @_;
            $failed{ $doc->{type} } = $doc->{id};
        },
        on_error => sub {
            die "Error creating multi unique keys: $_[2]";
        }
    );
    if (%failed) {
        delete @keys{ keys %failed };
        $self->multi_delete(%keys);
    }
    return %failed;
}

#===================================
sub multi_delete {
#===================================
    my ( $self, %keys ) = @_;

    my @docs = map { { type => $_, id => $keys{$_} } } keys %keys;

    $self->es->bulk_delete(
        index    => $self->index,
        docs     => \@docs,
        on_error => sub {
            die "Error deleting multi unique keys: $_[2]";
        }
    );
    return 1;
}

#===================================
sub multi_update {
#===================================
    my $self = shift;
    my %old  = %{ shift() || {} };
    my %new  = %{ shift() || {} };
    for ( keys %new ) {
        no warnings 'uninitialized';
        next unless $old{$_} eq $new{$_};
        delete $old{$_};
        delete $new{$_};
    }
    my %failed = $self->multi_create(%new);
    $self->multi_delete(%old) unless %failed;
    return %failed;
}

#===================================
sub multi_exists {
#===================================
    my ( $self, %keys ) = @_;
    my @docs = map { { _type => $_, _id => $keys{$_} } } keys %keys;
    my $exists = $self->es->mget( index => $self->index, docs => \@docs );
    for (@$exists) {
        next unless $_->{exists};
        delete $keys{ $_->{_type} };
    }
    return %keys;
}

#===================================
sub _params {
#===================================
    my ( $self, $method, $type, $id ) = @_;
    croak "No type passed to ${method}()"
        unless defined $type and length $type;
    croak "No id passed to ${method}()"
        unless defined $id and length $id;

    return (
        index => $self->index,
        type  => $type,
        id    => $id
    );
}

#===================================
sub bootstrap {
#===================================
    my $self = shift;
    my %params = ref $_[0] eq 'HASH' ? %{ shift() } : @_;
    %params = (
        auto_expand_replicas => '0-all',
        number_of_shards     => 1,
    ) unless %params;

    my $es    = $self->es;
    my $index = $self->index;
    return if $es->index_exists( index => $index );

    $es->create_index(
        index    => $index,
        settings => \%params,
        mappings => {
            _default_ => {
                _all    => { enabled => 0 },
                _source => { enabled => 0 },
                _type   => { index   => 'no' },
                enabled => 0,
            }
        }
    );
    $es->cluster_health( wait_for_status => 'yellow' );
    return $self;
}

#===================================
sub index { shift->{_index} }
sub es    { shift->{_es} }
#===================================

#===================================
sub delete_type {
#===================================
    my $self = shift;
    my $type = shift;
    croak "No type passed to delete_type()"
        unless defined $type and length $type;

    $self->es->delete_mapping(
        index          => $self->index,
        type           => $type,
        ignore_missing => 1
    );
    return $self;
}

#===================================
sub delete_index {
#===================================
    my $self = shift;
    $self->es->delete_index( index => $self->index, ignore_missing => 1 );
    return $self;
}

1;

# ABSTRACT: Track unique keys in ElasticSearch

__END__