view release on metacpan or  search on metacpan

lib/Catmandu/Importer/OAI.pm  view on Meta::CPAN

            my $_retried = $self->_retried();

            if ( $max_retries > 0 && $_retried < $max_retries  ){

                $_retried++;

                #exponential backoff:  [0 .. 2^c [
                my $n_seconds = int( 2**$_retried );
                $self->log->error("failed, retrying after $n_seconds");
                sleep $n_seconds;
                $self->_retried( $_retried );
                next;
            }
            else {
                my $err_msg = $self->url . " : " . $res->message." (stopped after ".$self->_retried()." retries)";
                $self->log->error( $err_msg );
                Catmandu::Error->throw( $err_msg );
            }
        }

        last;
    }

    $res;
}

sub _list_records {
    my $self = $_[0];
    my $args = $_[1];
    sub {
        state $stack = [];
        state $resumptionToken = $self->resumptionToken;
        state $resumptionData  = {};
        state $done  = 0;

        my $fill_stack = sub {
            push @$stack , shift;
        };

        if (@$stack <= 1 && $done == 0) {
            my %args = $args ? %$args : $self->_args;

            # Use the resumptionToken if one found on the last run, or if it was
            # undefined (last record)
            if (defined $resumptionToken) {
                my $verb = $args{verb};
                %args = (verb => $verb , resumptionToken => $resumptionToken);
            }

            my $sub = $self->listIdentifiers() ?
                sub { $self->oai->ListIdentifiers( %args , onRecord => $fill_stack ); } :
                sub { $self->oai->ListRecords( %args , onRecord => $fill_stack ); };

            my $res = $self->_retry( $sub );
            if (defined $res->resumptionToken) {
                $resumptionToken = $res->resumptionToken->resumptionToken;

                $resumptionData->{token}            = $resumptionToken;
                $resumptionData->{expirationDate}   = $res->resumptionToken->expirationDate;
                $resumptionData->{completeListSize} = $res->resumptionToken->completeListSize;
                $resumptionData->{cursor}           = $res->resumptionToken->cursor;
            }
            else {
                $resumptionToken = undef;
            }

            unless (defined $resumptionToken && length $resumptionToken) {
                $done = 1;
            }

            if ($self->sleep) {
                sleep $self->sleep;
            }
        }

        if (my $rec = shift @$stack) {
            if ($rec->isa('HTTP::OAI::Record')) {
                my $rec = $self->_map_record($rec);

                $rec->{_resumptionToken} = $resumptionToken if defined($resumptionToken);
                $rec->{_resumption} = $resumptionData if defined($resumptionData);

                return $rec;
            }
            else {
                my $rec =  {
                    _id => $rec->identifier,
                    _datestamp  => $rec->datestamp,
                    _status => $rec->status // "",
                };

                $rec->{_resumptionToken} = $resumptionToken if defined($resumptionToken);
                $rec->{_resumption} = $resumptionData if defined($resumptionData);

                return $rec;
            }
        }

        return undef;
    };
}

sub _list_sets {
    my $self = $_[0];
    sub {
        state $stack = [];
        state $done  = 0;

        my $fill_stack = sub {
            push @$stack , shift;
        };

        if (@$stack <= 1 && $done == 0) {
            my $sub = sub { $self->oai->ListSets( onRecord => $fill_stack ); };

            my $res = $self->_retry( $sub );
            $done = 1;
        }

        if (my $rec = shift @$stack) {
            return $self->_map_set($rec);