File Coverage

blib/lib/IO/Stream/HTTP/Persistent.pm

Criterion	Covered	Total	%
statement	114	132	86.3
branch	19	40	47.5
condition	8	20	40.0
subroutine	18	18	100.0
pod	1	3	33.3
total	160	213	75.1

package IO::Stream::HTTP::Persistent;
use 5.010001;
use warnings;
use strict;
use utf8;
use Carp;

our $VERSION = 'v0.2.2';

use Scalar::Util qw( dualvar );
use Data::Alias 0.08;
use IO::Stream::const;

use constant HTTP_SENT      => 1<<16;
use constant HTTP_RECV      => 1<<17;

use constant HTTP_EREQINCOMPLETE => dualvar(-10000, 'incomplete HTTP request headers');
use constant HTTP_ERESINCOMPLETE => dualvar(-10001, 'incomplete HTTP response');


#
# Export constants.
#
# Usage: use IO::Stream::HTTP::Persistent qw( :ALL :DEFAULT :Event :Error HTTP_RECV ... )
#
my %TAGS = (
    Event   => [ qw( HTTP_SENT HTTP_RECV ) ],
    Error   => [ qw( HTTP_EREQINCOMPLETE HTTP_ERESINCOMPLETE ) ],
);
$TAGS{ALL} = $TAGS{DEFAULT} = [ map { @{$_} } values %TAGS ];
my %KNOWN = map { $_ => 1 } @{ $TAGS{ALL} };

sub import {
    my (undef, @p) = @_;
    if (!@p) {
        @p = (':DEFAULT');
    }
    @p = map { /\A:(\w+)\z/xms ? @{ $TAGS{$1} || [] } : $_ } @p;
    my $pkg = caller;
    no strict 'refs';
    for my $const (@p) {
        next if !$KNOWN{$const};
        *{"${pkg}::$const"} = \&{$const};
    }
    return;
}


sub new {
    my ($class) = @_;
    my $self = bless {
        out_buf     => q{},     # modified on: OUT
        out_pos     => undef,   # modified on: OUT
        out_bytes   => 0,       # modified on: OUT
        in_buf      => q{},     # modified on: IN
        in_bytes    => 0,       # modified on: IN
        ip          => undef,   # modified on: RESOLVED
        is_eof      => undef,   # modified on: EOF
        out_sizes   => [],      # modified on: HTTP_SENT
        in_sizes    => [],      # modified on: HTTP_RECV
        _out_len    => 0,       # current length of {out_buf}
                                #   used to detect how many bytes was added to
                                #   {out_buf} in write() and increase {_out_todo}
        _out_todo   => 0,       # size of incomplete request at end of {out_buf}
                                #   used to find complete requests appended to
                                #   {out_buf} and add their sizes to {_out_queue}
                                #   can be negative, if we detected size
                                #   of next request but it isn't appended to
                                #   {out_buf} completely yet
        _out_queue  => [],      # sizes of unsent complete requests in {out_buf}
                                #   will be moved to {out_sizes} after sending
        _out_sent   => 0,       # how many bytes of {_out_queue}[0] already sent
                                #   if it become >= {_out_queue}[0] then it's
                                #   time to move from {_out_queue} to {out_sizes}
        _out_broken => 0,       # if true, disable HTTP_SENT and {out_sizes} support
        _in_todo    => 0,       # size of incomplete response at end of {in_buf}
                                #   used to find complete responses appended to
                                #   {in_buf} and add their sizes to {in_sizes}
        _wait_eof   => 0,       # flag: response end expected on EOF
        _wait_length=> 0,       # expected response length
        _wait_chunk => 0,       # known partial response length before next
                                # chunk header (or end of response sign)
    }, $class;
    return $self;
}

sub PREPARE {
    my ($self, $fh, $host, $port) = @_;
    for (qw( out_buf out_pos in_buf ip is_eof )) {
        alias $self->{$_} = $self->{_master}->{$_};
    }
    $self->{_slave}->PREPARE($fh, $host, $port);
    return;
}

sub WRITE {
    my ($self) = @_;
    my $m = $self->{_master};

    my $l = length $self->{out_buf};
    $self->{_out_todo} += $l - $self->{_out_len};
    $self->{_out_len}   = $l;

    while (!$self->{_out_broken} && $self->{_out_todo} > 0) {
        pos $self->{out_buf} = $self->{_out_len} - $self->{_out_todo};
        if ($self->{out_buf} =~ /\G((?:[^\r\n]+\r?\n)+\r?\n)/xms) {
            my $h = $1;
            my $c_len = $h =~ /^Content-Length:\s*(\d+)\s*\n/ixms ? $1 : 0;
            my $size = length($h) + $c_len;
            push @{ $self->{_out_queue} }, $size;
            $self->{_out_todo} -= $size;
        }
        else {
            $self->{_out_broken} = 1;
            $m->EVENT(0, HTTP_EREQINCOMPLETE);
            last;
        }
    }

    $self->{_slave}->WRITE();
    return;
}

sub EVENT { ## no critic (ProhibitExcessComplexity)
    my ($self, $e, $err) = @_;
    my $m = $self->{_master};

    if ($e & OUT) {
        $self->{_out_len}   = length $self->{out_buf};

        $m->{out_bytes}    += $self->{out_bytes};
        $self->{_out_sent} += $self->{out_bytes};
        $self->{out_bytes}  = 0;

        while (@{ $self->{_out_queue} } && $self->{_out_sent} >= $self->{_out_queue}[0]) {
            $e |= HTTP_SENT;
            $self->{_out_sent} -= $self->{_out_queue}[0];
            push @{ $self->{out_sizes} }, shift @{ $self->{_out_queue} };
        }
    }

    if ($e & IN) {
        $m->{in_bytes}     += $self->{in_bytes};
        $self->{_in_todo}  += $self->{in_bytes};
        $self->{in_bytes}   = 0;
    }

    while ($self->{_in_todo} > 0) {
        if ($e & IN) {
            if (!$self->{_wait_eof} && !$self->{_wait_length} && !$self->{_wait_chunk}) {
                pos $self->{in_buf} = length($self->{in_buf}) - $self->{_in_todo};
                if ($self->{in_buf} =~ /\G((?:[^\r\n]+\r?\n)+\r?\n)/xms) {
                    my $h = $1;
                    if ($h =~ /^Content-Length:\s*(\d+)\s*\n/ixms) {
                        $self->{_wait_length} = length($h) + $1;
                    }
                    elsif ($h =~ /^Transfer-Encoding:\s*chunked\s*\n/ixms) {
                        $self->{_wait_chunk} = length $h;
                    }
                    else {
                        $self->{_wait_eof} = 1;
                    }
                }
            }
            while ($self->{_wait_chunk} && $self->{_in_todo} > $self->{_wait_chunk}) {
                pos $self->{in_buf} = length($self->{in_buf}) - $self->{_in_todo} + $self->{_wait_chunk};
                if ($self->{in_buf} =~ /\G((?:\r?\n)?([\dA-Fa-f]+)[ \t]*\r?\n)/xms) {
                    my $chunk = hex $2;
                    $self->{_wait_chunk} += length($1) + $chunk;
                  next if $chunk > 0;
                    $self->{_wait_length} = $self->{_wait_chunk};
                    $self->{_wait_chunk}  = 0;
                }
                last;
            }
        }
        if ($e & EOF) {
            if ($self->{_wait_eof}) {
                $self->{_wait_length} = $self->{_in_todo};
                $self->{_wait_eof} = 0;
            }
        }
        if ($self->{_wait_length} && $self->{_in_todo} >= $self->{_wait_length}) {
            $self->{_in_todo} -= $self->{_wait_length};
            push @{ $self->{in_sizes} }, $self->{_wait_length};
            $self->{_wait_length} = 0;
            $e |= HTTP_RECV;
            next;
        }
        last;
    }

    if ($e & EOF) {
        if ($self->{_in_todo}) {
            $err ||= HTTP_ERESINCOMPLETE;
        }
    }

    $m->EVENT($e, $err);
    return;
}


1; # Magic true value required at end of module
__END__