view release on metacpan or  search on metacpan

lib/AnyEvent/Net/Curl/Queued.pm  view on Meta::CPAN

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.049'; # 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) = @_;