Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Created an internall scroller for the client. We only use this one feature (scrolled searches) and we have it as a dependency to function as an API to an API we could just use directly. The motivation is the breaking changes in latest 5.x versions of Search::Elasticsearch and the fact it's incompatiable with the current client code.
- Loading branch information
Showing
6 changed files
with
184 additions
and
30 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,167 @@ | ||
| package MetaCPAN::Client::Scroll; | ||
|
|
||
| use Moo; | ||
| use Carp; | ||
| use Ref::Util qw< is_arrayref is_hashref is_ref >; | ||
| use JSON::MaybeXS qw< decode_json encode_json >; | ||
|
|
||
| has ua => ( | ||
| is => 'ro', | ||
| required => 1, | ||
| ); | ||
|
|
||
| has size => ( | ||
| is => 'ro', | ||
| isa => sub { | ||
| is_ref($_[0]) and croak 'size must be a scalar'; | ||
| }, | ||
| ); | ||
|
|
||
| has time => ( | ||
| is => 'ro', | ||
| isa => sub { | ||
| is_ref($_[0]) and croak 'time must be a scalar'; | ||
| }, | ||
| ); | ||
|
|
||
| has base_url => ( | ||
| is => 'ro', | ||
| isa => sub { | ||
| is_ref($_[0]) and croak 'base_url must be a scalar'; | ||
| }, | ||
| required => 1, | ||
| ); | ||
|
|
||
| has type => ( | ||
| is => 'ro', | ||
| isa => sub { | ||
| is_ref($_[0]) and croak 'type must be a scalar'; | ||
| }, | ||
| required => 1, | ||
| ); | ||
|
|
||
| has body => ( | ||
| is => 'ro', | ||
| isa => sub { | ||
| is_hashref($_[0]) or croak 'body must be a hashref'; | ||
| }, | ||
| required => 1, | ||
| ); | ||
|
|
||
| has id => ( | ||
| is => 'ro', | ||
| isa => sub { | ||
| is_ref($_[0]) and croak 'id must be a scalar'; | ||
| }, | ||
| required => 1, | ||
| ); | ||
|
|
||
| has total => ( | ||
| is => 'ro', | ||
| isa => sub { | ||
| is_ref($_[0]) or $_[0] =~ /[^0-9]/ | ||
| and croak 'total must be a number'; | ||
| }, | ||
| required => 1, | ||
| ); | ||
|
|
||
| has _read => ( | ||
| is => 'ro', | ||
| isa => sub { | ||
| is_ref($_[0]) or $_[0] =~ /[^0-9]/ | ||
| and croak 'total must be a number'; | ||
| }, | ||
| default => sub { 0 }, | ||
| ); | ||
|
|
||
| has buffer => ( | ||
| is => 'ro', | ||
| isa => sub { | ||
| is_arrayref($_[0]) or croak 'buffer must be an array ref'; | ||
| }, | ||
| default => sub { [] }, | ||
| ); | ||
|
|
||
| has aggregations => ( | ||
| is => 'ro', | ||
| isa => sub { | ||
| is_hashref($_[0]) or croak 'aggregations must be an hash ref'; | ||
| }, | ||
| default => sub { +{} }, | ||
| ); | ||
|
|
||
| sub BUILDARGS { | ||
| my ( $class, %args ) = @_; | ||
| $args{time} //= '5m'; | ||
| $args{size} //= '100'; | ||
|
|
||
| my ( $ua, $base_url, $type, $body, $time, $size ) = | ||
| @args{qw< ua base_url type body time size >}; | ||
|
|
||
| # fetch | ||
| my $res = $ua->post( | ||
| sprintf( '%s/%s/_search?scroll=%s&size=%s', $base_url, $type, $time, $size ), | ||
| { content => encode_json $body } | ||
| ); | ||
|
|
||
| croak "failed to create a scrolled search" | ||
| unless $res->{status} == 200; | ||
|
|
||
| my $content = decode_json $res->{content}; | ||
|
|
||
| # set id | ||
| my $id = $content->{_scroll_id}; | ||
| $args{id} = $id; | ||
|
|
||
| # set total | ||
| $args{total} = $content->{hits}{total}; | ||
|
|
||
| # set buffer | ||
| push @{ $args{buffer} } => @{ $content->{hits}{hits} }; | ||
|
|
||
| # fecth aggregations | ||
| if ( $content->{aggregations} and is_hashref( $content->{aggregations} ) ) { | ||
| $args{aggregations} = $content->{aggregations}; | ||
| } | ||
|
|
||
| return \%args; | ||
| } | ||
|
|
||
| sub next { | ||
| my $self = shift; | ||
| return if $self->_read >= $self->total; | ||
|
|
||
| $self->_fetch_next unless @{ $self->buffer }; | ||
|
|
||
| return shift @{ $self->buffer }; | ||
| } | ||
|
|
||
| sub _fetch_next { | ||
| my $self = shift; | ||
|
|
||
| my $res = $self->ua->post( | ||
| sprintf( '%s/_search/scroll?scroll=%s&size=%s', $self->base_url, $self->time, $self->size ), | ||
| { content => $self->id } | ||
| ); | ||
|
|
||
| croak "failed to fetch next scolled batch" | ||
| unless $res->{status} == 200; | ||
|
|
||
| my $content = decode_json $res->{content}; | ||
|
|
||
| push @{ $self->buffer } => @{ $content->{hits}{hits} }; | ||
| } | ||
|
|
||
| sub DEMOLISH { | ||
| my $self = shift; | ||
|
|
||
| my $res = $self->ua->delete( | ||
| sprintf( '%s/_search/scroll?scroll=%s', $self->base_url, $self->time ), | ||
| { content => $self->id } | ||
| ); | ||
|
|
||
| warn "failed to delete scroller" | ||
| unless $res->{status} == 200; | ||
| } | ||
|
|
||
| 1; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters