view release on metacpan or  search on metacpan

lib/AnyEvent/Curl/Multi.pm  view on Meta::CPAN

hashref containing some interesting statistics.

=item error => $cb->($client, $request, $errmsg, $stats);

Fired when an error is received.  

The arguments sent to your callback will be the client object, the original
request (untampered with), the error message, and a hashref containing some
interesting statistics.  (If the error was other than a timeout, the statistics
may be invalid.)

=back
  
=cut

sub new { 
    my $class = shift;

    my $self = $class->SUPER::new(
        multi_h => WWW::Curl::Multi->new,
        state => {},
        timer_w => undef,
        io_w => {},
        queue => [],
        max_concurrency => 0,
        max_redirects => 0,
        timeout => undef,
        proxy => undef,
        debug => undef,
        ipresolve => undef,
        @_
    );

    if (! $MS_TIMEOUT_SUPPORTED 
        && $self->{timeout}
        && $self->{timeout} != int($self->{timeout})) {
        croak "Subsecond timeout resolution is not supported by your " .
              "libcurl version.  Upgrade to 7.16.2 or later.";
    }

    return bless $self, $class;
}

sub request {
    my $self = shift;
    my ($req, %opts) = @_;

    my $easy_h; 

    if ($req->isa("HTTP::Request")) {
        # Convert to WWW::Curl::Easy
        $easy_h = $self->_gen_easy_h($req, %opts);
    } else {
        croak "Unsupported request type";
    }

    # Initialize easy curl handle
    my $id = refaddr $easy_h;
    my ($response, $header);
    $easy_h->setopt(CURLOPT_WRITEDATA, \$response);
    $easy_h->setopt(CURLOPT_WRITEHEADER, \$header);
    $easy_h->setopt(CURLOPT_PRIVATE, $id);

    my $obj = {
        easy_h => $easy_h,
        req => $req,
        response => \$response,
        header => \$header,
        cv => AE::cv,
    };

    push @{$self->{queue}}, $obj;

    $self->_dequeue;

    return bless $obj, 'AnyEvent::Curl::Multi::Handle';
}

sub _dequeue {
    my $self = shift;

    while ($self->{max_concurrency} == 0 || 
           scalar keys %{$self->{state}} < $self->{max_concurrency}) {
        if (my $dequeued = shift @{$self->{queue}}) {
            $self->{state}->{refaddr($dequeued->{easy_h})} = $dequeued;
            # Add it to our multi handle
            $self->{multi_h}->add_handle($dequeued->{easy_h});
        } else {
            last;
        }
    }
    
    # Start our timer
    $self->{timer_w} = AE::timer(0, 0.5, sub { $self->_perform });
}

sub _perform {
    my $self = shift;

    $self->{multi_h}->perform;

    while (my ($id, $rv) = $self->{multi_h}->info_read) {
        if ($id) {
            my $state = $self->{state}->{$id};
            my $req = $state->{req};
            my $easy_h = $state->{easy_h};
            my $stats = {
                total_time => $easy_h->getinfo(CURLINFO_TOTAL_TIME),
                dns_time => $easy_h->getinfo(CURLINFO_NAMELOOKUP_TIME),
                connect_time => $easy_h->getinfo(CURLINFO_CONNECT_TIME),
                start_transfer_time => 
                    $easy_h->getinfo(CURLINFO_STARTTRANSFER_TIME),
                download_bytes => 
                    $easy_h->getinfo(CURLINFO_SIZE_DOWNLOAD),
                upload_bytes => $easy_h->getinfo(CURLINFO_SIZE_UPLOAD),
            };
            if ($rv) {
                # Error
                $state->{cv}->croak($easy_h->errbuf);
                $req->event('error', $easy_h->errbuf, $stats) 
                    if $req->can('event');