view release on metacpan or  search on metacpan

lib/AnyEvent/MultiDownload.pm  view on Meta::CPAN

#!/usr/bin/perl
package AnyEvent::MultiDownload;
use strict;
use utf8;
use Moo;
use AE;
use Asset::File;
use AnyEvent::Digest;
use List::Util qw/shuffle/;
use AnyEvent::HTTP qw/http_get/;

our $VERSION = '1.13';

has path => (
    is => 'ro',
    isa => sub {
        die "文件存在" if -e $_[0];
    },
    required => 1,
);

has url => (
    is  => 'ro',
    required => 1,
);


has mirror => (
    is => 'rw',
    predicate => 1,
    isa => sub {
        return 1 if ref $_[0] eq 'ARRAY';
    },
);

has digest => (
    is => 'rw',
    isa => sub {
        return 1 if $_[0] =~ /Digest::(SHA|MD5)/;
    }
);


has on_finish => (
    is => 'rw',
    required => 1,
    isa => sub {
        return 2 if ref $_[0] eq 'CODE';
    }
);

has on_error => (
    is => 'rw',
    isa => sub {
        return 2 if ref $_[0] eq 'CODE';
    },
    default => sub {  sub { 1 } },
);

has on_block_finish => (
    is => 'rw',
    isa => sub {
        return 2 if ref $_[0] eq 'CODE';
    },

lib/AnyEvent/MultiDownload.pm  view on Meta::CPAN

sub retry {
    my ($self, $task, $retry) = @_;
    my $w;$w = AE::timer( $self->retry_interval, 0, sub {
        $task->{pos}  = $task->{ofs}; # 重下本块时要 seek 回零
        $task->{size} = 0;
        $task->{ctx}  = undef;
        $self->fetch_block( $task, ++$retry );
        undef $w;
    });
}

sub split_range {
    my $self    = shift;
    my $length  = shift;

    # 每个请求的段大小的范围,字节
    my $block_size   = $self->block_size;
    my $segments   = int($length / $block_size);

    # 要处理的字节的总数
    my $len_remain = $length;

    my @ranges;
    my $block = 0;
    while ( $len_remain > 0 ) {
        # 每个 segment  的大小
        my $seg_len = $block_size;

        # 偏移长度
        my $ofs = $length - $len_remain;
        
        # 剩余字节
        $len_remain -= $seg_len;

        my $tail  = $ofs + $seg_len - 1; 
        if ( $length-1  < $tail) {
            $tail = $length-1;
        }

        my $task  = { 
            block => $block, # 当前块编号
            ofs   => $ofs,   # 当前的偏移量
            pos   => $ofs,   # 本块的起点
            tail  => $tail,  # 本块的结束
            range => 'bytes=' . $ofs . '-' . $tail, 
            size  => 0,      # 总共下载的长度
        }; 

        $self->tasks->[$block] = $task; 
        $block++;
    }
}

1;

__END__

=pod
 
=encoding utf8

=head1 NAME

AnyEvent::MultiDownload - 非阻塞的多线程多地址文件下载的模块

=head1 SYNOPSIS

这是一个全非阻塞的多线程多地址文件下载的模块, 可以象下面这个应用一样, 同时下载多个文件, 并且整个过程都是异步事件解发, 不会阻塞主进程.

下面是个简单的例子, 同时从多个地址下载同一个文件.

    use AE;
    use AnyEvent::MultiDownload;

    my @urls = (
        'http://mirrors.163.com/centos/7/isos/x86_64/CentOS-7.0-1406-x86_64-DVD.iso',
        'http://mirrors.sohu.com/centos/7/isos/x86_64/CentOS-7.0-1406-x86_64-DVD.iso',
    );
    
    my $cv = AE::cv;
    my $MultiDown = AnyEvent::MultiDownload->new( 
        url     => pop @urls, 
        mirror  => \@urls, 
        path  => '/tmp/ubuntu.iso',
        block_size => 1 * 1024 * 1024, # 1M
        on_block_finish => sub {
            my ($hdr, $block_ref, $md5) = @_;
            if ($md5 eq $src_md5) {
                return 1;
            }
            0
        },
        on_finish => sub {
            my $len = shift;
            $cv->send;
        },
        on_error => sub {
            my ($error, $hdr) = @_;
            $cv->send;
        }
    )->start;
    
    $cv->recv;


下面是异步同时下载多个文件的实例. 整个过程异步.

    use AE;
    use AnyEvent::MultiDownload;
    
    my $cv = AE::cv;
    
    $cv->begin;
    my $MultiDown = AnyEvent::MultiDownload->new( 
        url     => 'http://xxx/file1',
        path  => "/tmp/file2",
        on_finish => sub {
            my $len = shift;
            $cv->end;
        },