File Coverage

blib/lib/Search/Elasticsearch/Transport/Async.pm

Criterion	Covered	Total	%
statement	43	43	100.0
branch	6	6	100.0
condition			n/a
subroutine	10	10	100.0
pod	1	1	100.0
total	60	60	100.0

# Licensed to Elasticsearch B.V. under one or more contributor
# license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright
# ownership. Elasticsearch B.V. licenses this file to you under
# the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

package Search::Elasticsearch::Transport::Async;
$Search::Elasticsearch::Transport::Async::VERSION = '8.00';
use Moo;
with 'Search::Elasticsearch::Role::Is_Async',
    'Search::Elasticsearch::Role::Transport';

use Time::HiRes qw(time);
use Search::Elasticsearch::Util qw(upgrade_error);
use Promises qw(deferred);
use namespace::clean;

#===================================
sub perform_request {
#===================================
    my $self   = shift;
    my $params = $self->tidy_request(@_);
    my $pool   = $self->cxn_pool;
    my $logger = $self->logger;

    my $deferred = deferred;

    my ( $start, $cxn );
    $pool->next_cxn

        # perform request
        ->then(
        sub {
            $cxn   = shift;
            $start = time();
            $cxn->perform_request($params);
        }
        )

        # log request regardless of success/failure
        ->finally( sub { $logger->trace_request( $cxn, $params ) } )

        ->done(
        # request succeeded
        sub {
            my ( $code, $response ) = @_;
            $pool->request_ok($cxn);
            $logger->trace_response( $cxn, $code, $response,
                time() - $start );
            $deferred->resolve($response);
        },

        # request failed
        sub {
            my $error = upgrade_error( shift(), { request => $params } );
            if ( $pool->request_failed( $cxn, $error ) ) {

                # log failed, then retry
                $logger->debugf( "[%s] %s", $cxn->stringify, "$error" );
                $logger->info('Retrying request on a new cxn');
                return $self->perform_request($params)->done(
                    sub { $deferred->resolve(@_) },
                    sub { $deferred->reject(@_) }
                );
            }
            if ($cxn) {
                $logger->trace_request( $cxn, $params );
                $logger->trace_error( $cxn, $error );
            }
            $error->is('NoNodes')
                ? $logger->critical($error)
                : $logger->error($error);
            $deferred->reject($error);
        }
        );
    return $deferred->promise;
}

1;

#ABSTRACT: Provides async interface between the client class and the Elasticsearch cluster

__END__