File Coverage

blib/lib/AnyEvent/MultiDownload.pm

Criterion	Covered	Total	%
statement	131	144	90.9
branch	27	48	56.2
condition	22	42	52.3
subroutine	20	22	90.9
pod	1	7	14.2
total	201	263	76.4

#!/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.11';

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';
    },
    default => sub { return sub {1} }
);

has cv => (
    is       => 'rw',
    lazy     => 1,
    default => sub { AE::cv },
);

has fh       => (
    is      => 'rw',
    lazy    => 1,
    default => sub {
        return Asset::File->new;
    },
);

has retry_interval => is => 'rw', default => sub { 10 };
has max_retries    => is => 'rw', default => sub { 3 };
has block_size       => is => 'rw', default => sub { 1 * 1024 * 1024 };
has timeout        => is => 'rw', default => sub { 10 };
has recurse        => is => 'rw', default => sub { 6 };
has headers        => is => 'rw', default => sub {{}};
has tasks          => is => 'rw', default => sub { [] };
has error          => is => 'rw', default => sub {};
has task_lists     => is => 'rw', default => sub {[]};
has max_per_host   => is => 'rw', default => sub { 8 };
has url_status  => (
    is      => 'rw',
    lazy    => 1,
    default => sub {
        my $self = shift;
        my %hash;
        if ($self->has_mirror) {
            %hash = map {
                        $_ => 0
                    } @{ $self->mirror }, $self->url;
        }
        else {
            $hash{$self->url} = 0;
        }
        return \%hash;
    }
);

sub start  {
    my $self = shift;
    my $cb   = shift;

    $self->cv->cb(sub{
        my $cv = shift;
        my ($info, $hdr) = $cv->recv;
        if ($info) {
            AE::log debug => $info;
            return $self->on_error->($info, $hdr);
        }
        $self->fh->move_to($self->path);
        $self->on_finish->($self->fh->size);
    });

    $self->cv->begin;
    $self->first_request(0);
}


sub first_request {
    my ($self, $retry) = @_;
    my $url = $self->shuffle_url;

    my $first_task;
    my $ev; $ev = http_get $url,
        headers     => $self->headers,
        timeout     => $self->timeout,
        recurse     => $self->recurse,
        on_header   => sub {
            my ($hdr) = @_;
            if ( $hdr->{Status} == 200 ) {
                my $len = $hdr->{'content-length'};
                return $self->cv->send(("Cannot find a content-length header.", $hdr)) if !defined($len);

                # 准备开始下载的信息
                my $ranges = $self->split_range($len);

                # 除了第一个块, 其它块现在开始下载
                # 事件开始, 但这个回调会在最后才调用.
                $first_task = shift @{ $self->tasks };
                $first_task->{block} = $first_task->{block} || 0;
                $first_task->{ofs}   = $first_task->{ofs}   || 0;
                return 1 if $len <= $self->block_size;

                for ( 1 .. $self->max_per_host ) {
                    my $block_task = shift @{ $self->tasks };
                    last unless defined $block_task;
                    $self->cv->begin;
                    $self->fetch_block($block_task) ;

                }
            }
            1
        },
        on_body   => sub {
            my ($partial_body, $hdr) = @_;
            if ( $self->on_body($first_task)->($partial_body, $hdr) ) {
                # 如果是第一个块的话, 下载到指定的大小就需要断开
                if ( ( $hdr->{'content-length'} <= $self->block_size and  $first_task->{size} == $hdr->{'content-length'} )
                        or
                        $first_task->{size} >= $self->block_size
                    ) {

                    $self->cv->send(("The 0 block the compared failure", $hdr))
                        if !$self->on_block_finish->( $hdr, $first_task, $self->digest ? $first_task->{ctx}->hexdigest : '');
                    $self->cv->end;
                    return 0
                }
            }
            return 1;
        },
        sub {
            my (undef, $hdr) = @_;
            undef $ev;
            my $status = $hdr->{Status};
            # on_body 正常的下载
            return if ( $hdr->{OrigStatus} and $hdr->{OrigStatus} == 200 ) or $hdr->{Status} == 200 or $hdr->{Status} == 416;

            if ( ($status == 500 or $status == 503 or $status =~ /^59/) and $retry < $self->max_retries ) {
                my $w; $w = AE::timer( $self->retry_interval, 0, sub {
                    $first_task->{pos}  = $first_task->{ofs}; # 重下本块时要 seek 回零
                    $first_task->{size} = 0;
                    $first_task->{ctx}  = undef;
                     $self->first_request(++$retry);
                    undef $w;
                });
                AE::log debug => "地址 $url 的块 0 下载出错, 重试";
                return;
            }

            return $self->cv->send((
                sprintf("Status: %s, Reason: %s.", $status ? $status : '500', $hdr->{Reason} ? $hdr->{Reason} : ' '),
                $hdr)
            );
        }
}


sub shuffle_url {
    my $self = shift;
    my $urls = $self->url_status;
    return (shuffle keys %$urls)[0];
}

sub on_body {
    my ($self, $task) = @_;
    return sub {
        my ($partial_body, $hdr) = @_;
        return 0 unless ($hdr->{Status} == 206 || $hdr->{Status} == 200);

        my $len = length($partial_body);
        # 主要是用于解决第一个块会超过写的位置
        if ( $task->{size} + $len > $self->block_size ) {
            my $spsize = $len - ( $task->{size} + $len - $self->block_size );
            $partial_body = substr($partial_body, 0, $spsize);
            $len = $spsize;
        }

        $self->fh->start_range($task->{pos});
        $self->fh->add_chunk($partial_body);

        if ( $self->digest ) {
            $task->{ctx} ||= AnyEvent::Digest->new($self->digest);
            $task->{ctx}->add_async($partial_body);
        }

        $task->{pos}   += $len;
        $task->{size}  += $len;
        return 1;
    }
}

sub fetch_block {
    my ($self, $task, $retry) = @_;
    $retry ||= 0;
    my $url   = $self->shuffle_url;

    my $ev; $ev = http_get $url,
        timeout     => $self->timeout,
        recurse     => $self->recurse,
        persistent  => 1,
        keepalive   => 1,
        headers     => {
            %{ $self->headers },
            Range => $task->{range}
        },
        on_body => $self->on_body($task),
        sub {
            my ($hdl, $hdr) = @_;
            my $status = $hdr->{Status};
            undef $ev;

            # 成功下载到的流程
            # 1. 需要对比大小是否一致, 接着对比块较检
            # 2. 开始下一个任务的下载
            # 3. 当前块就退出, 不然下面会重试
            if ( $status == 200 || $status == 206  ) { # 第一个块, 这二个都有可能
                # not ok 块较检不相等 | 直接失败
                return $self->cv->send(("The $task->{block} block the compared failure", $hdr))
                    if ($task->{size} != ( $task->{tail} -$task->{ofs} + 1 )
                        or !$self->on_block_finish->($hdl, $task, $self->digest ? $task->{ctx}->hexdigest : ''));
                my $block_task = shift @{ $self->tasks };
                # 完成, 标记结束本次请求
                # ok  大小相等, 块较检相等, 当前块下载完成, 开始下载新的
                AE::log debug => "地址 $url 的块 $task->{block} 下载完成 $$";
                # 处理接下来的一个请求
                $block_task ? $self->fetch_block($block_task) : $self->cv->end;
                return;
            }

            # 是否重试的流程
            my $error  = sprintf(
                "Block %s the size is wrong, expect the size: %s actual size: %s, The %s try again,  Status: %s, Reason: %s.",
                $task->{block},
                $self->block_size,
                $task->{size},
                $retry,
		    	$status ? $status : '500',
                $hdr->{Reason} ? $hdr->{Reason} : ' ', );
            AE::log warn => $error;

            # 失败
            # 如果有可能还连接上的响应, 就需要重试, 直到达到重试, 如果不可能连接的响应, 就直接快速的退出
            return $self->cv->send(($error, $hdr))
                if $status !~ /^(59.|503|500|502|200|206|)$/ or $retry > $self->max_retries;

            $self->retry($task, $retry);
        }
};

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__