File Coverage

blib/lib/AnyEvent/Net/Curl/Queued.pm

Criterion	Covered	Total	%
statement	22	24	91.6
branch			n/a
condition			n/a
subroutine	8	8	100.0
pod			n/a
total	30	32	93.7

package AnyEvent::Net::Curl::Queued;
# ABSTRACT: Moo wrapper for queued downloads via Net::Curl & AnyEvent


use strict;
use utf8;
use warnings qw(all);

use AnyEvent;
use Carp qw(confess);
use Moo;
use MooX::Types::MooseLike::Base qw(
    AnyOf
    ArrayRef
    Bool
    HashRef
    InstanceOf
    Int
    Num
    Object
    Str
    is_Int
);
use Net::Curl::Share;

use AnyEvent::Net::Curl::Queued::Multi;

our $VERSION = '0.047'; # VERSION


has allow_dups  => (is => 'ro', isa => Bool, default => sub { 0 });


has common_opts => (is => 'ro', isa => HashRef, default => sub { {} });


has http_response => (is => 'ro', isa => Bool, default => sub { 0 });


has completed  => (
    is          => 'ro',
    isa         => Int,
    default     => sub { 0 },
    writer      => 'set_completed',
);

sub inc_completed {
    my ($self) = @_;
    return $self->set_completed($self->completed + 1);
}


has cv          => (is => 'ro', isa => Object, default => sub { AE::cv }, lazy => 1, writer => 'set_cv');


has max         => (
    is          => 'rw',
    isa         => Int,
    coerce      => sub {
        confess 'At least 1 connection required'
            if not is_Int($_[0])
            or $_[0] < 1;
        return $_[0];
    },
    default     => sub { 4 },
);


has multi       => (is => 'ro', isa => InstanceOf['AnyEvent::Net::Curl::Queued::Multi'], writer => 'set_multi');


has queue       => (
    is          => 'ro',
    isa         => ArrayRef[Object],
    default     => sub { [] },
);

## no critic (RequireArgUnpacking)

sub queue_push      { return 0 + push @{shift->queue}, @_ }
sub queue_unshift   { return 0 + unshift @{shift->queue}, @_ }
sub dequeue         { return shift @{shift->queue} }
sub count           { return 0 + @{shift->queue} }


has share       => (
    is      => 'ro',
    isa     => InstanceOf['Net::Curl::Share'],
    default => sub { Net::Curl::Share->new({ stamp => time }) },
    lazy    => 1,
);


has stats       => (is => 'ro', isa => InstanceOf['AnyEvent::Net::Curl::Queued::Stats'], default => sub { AnyEvent::Net::Curl::Queued::Stats->new }, lazy => 1);


has timeout     => (is => 'ro', isa => Num, default => sub { 60.0 });


has unique      => (is => 'ro', isa => HashRef[Str], default => sub { {} });


has watchdog    => (is => 'ro', isa => AnyOf[ArrayRef, Object], writer => 'set_watchdog', clearer => 'clear_watchdog', predicate => 'has_watchdog', weak_ref => 0);


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

    $self->set_multi(
        AnyEvent::Net::Curl::Queued::Multi->new({
            max         => $self->max,
            timeout     => $self->timeout,
        })
    );

    $self->share->setopt(Net::Curl::Share::CURLSHOPT_SHARE, Net::Curl::Share::CURL_LOCK_DATA_COOKIE);   # 2
    $self->share->setopt(Net::Curl::Share::CURLSHOPT_SHARE, Net::Curl::Share::CURL_LOCK_DATA_DNS);      # 3

    ## no critic (RequireCheckingReturnValueOfEval)
    eval { $self->share->setopt(Net::Curl::Share::CURLSHOPT_SHARE, Net::Curl::Share::CURL_LOCK_DATA_SSL_SESSION) };

    return;
}

sub BUILDARGS {
    my $class = shift;
    if (@_ == 1 and q(HASH) eq ref $_[0]) {
        return shift;
    } elsif (@_ % 2 == 0) {
        return { @_ };
    } elsif (@_ == 1) {
        return { max => shift };
    } else {
        confess 'Should be initialized as ' . $class . '->new(Hash|HashRef|Int)';
    }
}


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

    # watchdog
    $self->set_watchdog(AE::timer 1, 1, sub {
        $self->multi->perform;
        $self->empty;
    });

    # populate queue
    $self->add($self->dequeue)
        while
            $self->count
            and ($self->multi->handles < $self->max);

    # check if queue is empty
    $self->empty;

    return;
}


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

    AE::postpone { $self->cv->send }
        if
            $self->completed > 0
            and $self->count == 0
            and $self->multi->handles == 0;

    return;
}



sub add {
    my ($self, $worker) = @_;

    # vivify the worker
    $worker = $worker->()
        if ref($worker) eq 'CODE';

    # self-reference & warmup
    $worker->queue($self);
    $worker->init;

    # check if already processed
    if ($self->allow_dups
        or $worker->force
        or ++$self->unique->{$worker->unique} == 1
    ) {
        # fire
        $self->multi->add_handle($worker);
    }

    return;
}


sub append {
    my ($self, $worker) = @_;

    $self->queue_push($worker);
    $self->start;

    return;
}


sub prepend {
    my ($self, $worker) = @_;

    $self->queue_unshift($worker);
    $self->start;

    return;
}


## no critic (ProhibitBuiltinHomonyms)
sub wait {
    my ($self) = @_;

    # handle queue
    $self->cv->recv;

    # stop the watchdog
    $self->clear_watchdog;

    # reload
    $self->set_cv(AE::cv);

    return;
}


1;

__END__