view release on metacpan or  search on metacpan

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

    }

    # 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');
                $self->event('error', $req, $easy_h->errbuf, $stats);
            } else {
                # libcurl appends subsequent response headers to the buffer
                # when following redirects.  We need to remove all but the
                # most recent header before we parse the response.
                my $last_header = (split(/\r?\n\r?\n/, 
                                   ${$state->{header}}))[-1];
                my $response = HTTP::Response->parse($last_header . 
                                                     "\n\n" . 
                                                     ${$state->{response}});
                $req->uri($easy_h->getinfo(CURLINFO_EFFECTIVE_URL));
                $response->request($req);
                $state->{cv}->send($response, $stats);
                $req->event('response', $response, $stats) 
                    if $req->can('event');
                $self->event('response', $req, $response, $stats);
            }
            delete $self->{state}->{$id};
            $self->_dequeue;
        }
    }

    # We must recalculate the number of active handles here, because
    # a user-provided callback may have added a new one.
    my $active_handles = scalar keys %{$self->{state}};
    if (! $active_handles) {
        # Nothing left to do - no point keeping the watchers around anymore.
        delete $self->{timer_w};
        delete $self->{io_w};
        return;
    }

    # Re-establish all I/O watchers
    foreach my $fd (keys %{$self->{io_w}}) {
        delete $self->{io_w}->{$fd};
    }

    my ($readfds, $writefds, $errfds) = $self->{multi_h}->fdset;

    foreach my $fd (@$writefds) {
        $self->{io_w}->{$fd} ||= AE::io($fd, 1, sub { $self->_perform }); 
    }
    foreach my $fd (@$readfds) {
        $self->{io_w}->{$fd} ||= AE::io($fd, 0, sub { $self->_perform }); 
    }
}

sub _gen_easy_h {
    my $self = shift;
    my $req = shift;
    my %opts = @_;

    my $easy_h = WWW::Curl::Easy->new;
    $easy_h->setopt(CURLOPT_URL, $req->uri);