view release on metacpan or  search on metacpan

lib/Parallel/Downloader.pm  view on Meta::CPAN

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 };