File Coverage

blib/lib/Parallel/Downloader.pm

Criterion	Covered	Total	%
statement	81	88	92.0
branch	8	14	57.1
condition	1	2	50.0
subroutine	15	16	93.7
pod	2	2	100.0
total	107	122	87.7

use strictures;

package Parallel::Downloader;

our $VERSION = '0.132071'; # VERSION

# ABSTRACT: simply download multiple files at once

#
# This file is part of Parallel-Downloader
#
#
# Christian Walde has dedicated the work to the Commons by waiving all of his
# or her rights to the work worldwide under copyright law and all related or
# neighboring legal rights he or she had in the work, to the extent allowable by
# law.
#
# Works under CC0 do not require attribution. When citing the work, you should
# not imply endorsement by the author.
#


use Moo;
use MooX::Types::MooseLike::Base qw( Bool Int HashRef CodeRef ArrayRef );

sub {
    has requests       => ( is => 'ro', isa => ArrayRef, required => 1 );
    has workers        => ( is => 'ro', isa => Int,      default  => sub { 10 } );
    has conns_per_host => ( is => 'ro', isa => Int,      default  => sub { 4 } );
    has aehttp_args    => ( is => 'ro', isa => HashRef,  default  => sub { {} } );
    has debug          => ( is => 'ro', isa => Bool,     default  => sub { 0 } );
    has logger         => ( is => 'ro', isa => CodeRef,  default  => sub { \&_default_log } );
    has build_response => ( is => 'ro', isa => CodeRef,  default  => sub { \&_default_build_response } );
    has sorted         => ( is => 'ro', isa => Bool,     default  => sub { 1 } );

    has _consumables => ( is => 'lazy', isa => ArrayRef, builder => '_requests_interleaved_by_host' );

    has _responses => ( is => 'ro', isa => ArrayRef, default => sub { [] } );
    has _cv => ( is => 'ro', isa => sub { $_[0]->isa( 'AnyEvent::CondVar' ) }, default => sub { AnyEvent->condvar } );
  }
  ->();

use AnyEvent::HTTP;
use Sub::Exporter::Simple 'async_download';


sub async_download {
    return __PACKAGE__->new( @_ )->run;
}

sub _requests_interleaved_by_host {
    my ( $self, $requests ) = @_;

    my %hosts;
    for ( @{ $self->requests } ) {
        my $host_name = $_->uri->host;
        my $host = $hosts{$host_name} ||= [];
        push @{$host}, $_;
    }

    my @interleaved_list;
    while ( keys %hosts ) {
        push @interleaved_list, shift @{$_} for values %hosts;
        for ( keys %hosts ) {
            next if @{ $hosts{$_} };
            delete $hosts{$_};
        }
    }

    return \@interleaved_list;
}


sub run {
    my ( $self ) = @_;

    local $AnyEvent::HTTP::MAX_PER_HOST = $self->conns_per_host;

    for ( 1 .. $self->_sanitize_worker_max ) {
        $self->_cv->begin;
        $self->_log( msg => "$_ started", type => "WorkerStart", worker_id => $_ );
        $self->_add_request( $_ );
    }

    $self->_cv->recv;

    return @{ $self->_responses } if !$self->sorted;

    my %unsorted = map { 0 + $_->[2] => $_ } @{ $self->_responses };
    my @sorted = map { $unsorted{ 0 + $_ } } @{ $self->requests };

    return @sorted;
}

sub _add_request {
    my ( $self, $worker_id ) = @_;

    my $req = shift @{ $self->_consumables };
    return $self->_end_worker( $worker_id ) if !$req;

    my $post_download_sub = $self->_make_post_download_sub( $worker_id, $req );

    http_request(
        $req->method,
        $req->uri->as_string,
        body    => $req->content,
        headers => $req->{_headers},
        %{ $self->aehttp_args },
        $post_download_sub
    );

    my $host_name = $req->uri->host;
    $self->_log(
        msg       => "$worker_id accepted new request for $host_name",
        type      => "WorkerRequestAdd",
        worker_id => $worker_id,
        req       => $req
    );

    return;
}

sub _make_post_download_sub {
    my ( $self, $worker_id, $req ) = @_;

    my $post_download_sub = sub {
        push @{ $self->_responses }, $self->build_response->( @_, $req );

        my $host_name = $req->uri->host;
        $self->_log(
            msg       => "$worker_id completed a request for $host_name",
            type      => "WorkerRequestEnd",
            worker_id => $worker_id,
            req       => $req
        );

        $self->_add_request( $worker_id );
        return;
    };

    return $post_download_sub;
}

sub _default_build_response {
    my ( $body, $hdr, $req ) = @_;
    return [ $body, $hdr, $req ];
}

sub _end_worker {
    my ( $self, $worker_id ) = @_;
    $self->_log( msg => "$worker_id ended", type => "WorkerEnd", worker_id => $worker_id );
    $self->_cv->end;
    return;
}

sub _sanitize_worker_max {
    my ( $self ) = @_;

    die "max should be 0 or more" if $self->workers < 0;

    my $request_count = @{ $self->requests };

    return $request_count if !$self->workers;                    # 0 = as many parallel as possible
    return $request_count if $self->workers > $request_count;    # not more than the request count

    return $self->workers;
}

sub _log {
    my ( $self, %msg ) = @_;
    return if !$self->debug;
    $self->logger->( $self, \%msg );
    return;
}

sub _default_log {
    my ( $self, $msg ) = @_;
    print "$msg->{msg}\n";
    return;
}

1;

__END__