File Coverage

blib/lib/Hijk.pm

Criterion	Covered	Total	%
statement	163	195	83.5
branch	93	154	60.3
condition	55	107	51.4
subroutine	13	13	100.0
pod	0	1	0.0
total	324	470	68.9

package Hijk;
use strict;
use warnings;
use Time::HiRes;
use POSIX qw(:errno_h);
use Socket qw(PF_INET SOCK_STREAM pack_sockaddr_in $CRLF SOL_SOCKET SO_ERROR);
use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK);

our $VERSION = "0.28";

sub Hijk::Error::CONNECT_TIMEOUT         () { 1 << 0 } # 1
sub Hijk::Error::READ_TIMEOUT            () { 1 << 1 } # 2
sub Hijk::Error::TIMEOUT                 () { Hijk::Error::READ_TIMEOUT | Hijk::Error::CONNECT_TIMEOUT } # 3
sub Hijk::Error::CANNOT_RESOLVE          () { 1 << 2 } # 4
sub Hijk::Error::REQUEST_SELECT_ERROR    () { 1 << 3 } # 8
sub Hijk::Error::REQUEST_WRITE_ERROR     () { 1 << 4 } # 16
sub Hijk::Error::REQUEST_ERROR           () { Hijk::Error::REQUEST_SELECT_ERROR |  Hijk::Error::REQUEST_WRITE_ERROR } # 24
sub Hijk::Error::RESPONSE_READ_ERROR     () { 1 << 5 } # 32
sub Hijk::Error::RESPONSE_BAD_READ_VALUE () { 1 << 6 } # 64
sub Hijk::Error::RESPONSE_ERROR          () { Hijk::Error::RESPONSE_READ_ERROR | Hijk::Error::RESPONSE_BAD_READ_VALUE } # 96

sub _read_http_message {
    my ($fd, $read_length, $read_timeout, $parse_chunked, $head_as_array, $method) = @_;
    $read_timeout = undef if defined($read_timeout) && $read_timeout <= 0;

    my ($body,$buf,$decapitated,$nbytes,$proto);
    my $status_code = 0;
    my $header = $head_as_array ? [] : {};
    my $no_content_len = 0;
    my $head = "";
    my $method_has_no_content = do { no warnings qw(uninitialized); $method eq "HEAD" };
    my $close_connection;
    vec(my $rin = '', $fd, 1) = 1;
    do {
        return ($close_connection,undef,0,undef,undef, Hijk::Error::READ_TIMEOUT)
            if ((_select($rin, undef, undef, $read_timeout) != 1) || (defined($read_timeout) && $read_timeout <= 0));

        my $nbytes = POSIX::read($fd, $buf, $read_length);
        return ($close_connection, $proto, $status_code, $header, $body)
            if $no_content_len && $decapitated && (!defined($nbytes) || $nbytes == 0);
        if (!defined($nbytes)) {
            next if ($! == EWOULDBLOCK || $! == EAGAIN || $! == EINTR);
            return (
                $close_connection, undef, 0, undef, undef,
                Hijk::Error::RESPONSE_READ_ERROR,
                "Failed to read http " . ($decapitated ? "body": "head") . " from socket",
                $!+0,
                "$!",
            );
        }

        if ($nbytes == 0) {
            return (
                $close_connection, undef, 0, undef, undef,
                Hijk::Error::RESPONSE_BAD_READ_VALUE,
                "Wasn't expecting a 0 byte response for http " . ($decapitated ? "body": "head" ) . ". This shouldn't happen",
            );
        }

        if ($decapitated) {
            $body .= $buf;
            if (!$no_content_len) {
                $read_length -= $nbytes;
            }
        }
        else {
            $head .= $buf;
            my $neck_pos = index($head, "${CRLF}${CRLF}");
            if ($neck_pos > 0) {
                $decapitated = 1;
                $body = substr($head, $neck_pos+4);
                $head = substr($head, 0, $neck_pos);
                $proto = substr($head, 0, 8);
                $status_code = substr($head, 9, 3);
                $method_has_no_content = 1 if $status_code == 204; # 204 NO CONTENT, see http://tools.ietf.org/html/rfc2616#page-60
                substr($head, 0, index($head, $CRLF) + 2, ""); # 2 = length($CRLF)

                my ($doing_chunked, $content_length, $trailer_mode, $trailer_value_is_true);
                for (split /${CRLF}/o, $head) {
                    my ($key, $value) = split /: /, $_, 2;
                    my $key_lc = lc($key);

                    # Figure this out now so we don't need to scan the
                    # list later under $head_as_array, and just for
                    # simplicity and to avoid duplicating code later
                    # when !$head_as_array.
                    if ($key_lc eq 'transfer-encoding' and $value eq 'chunked') {
                        $doing_chunked = 1;
                    } elsif ( ($key_lc eq 'content-length') || (lc($key) eq 'content-length') ) {
                        $content_length = $value;
                    } elsif ($key_lc eq 'connection' and $value eq 'close') {
                        $close_connection = 1;
                    } elsif ($key_lc eq 'trailer' and $value) {
                        $trailer_value_is_true = 1;
                    }

                    if ($head_as_array) {
                        push @$header => $key, $value;
                    } else {
                        $header->{$key} = $value;
                    }
                }

                # We're processing the headers as a stream, and we
                # only want to turn on $trailer_mode if
                # Transfer-Encoding=chunked && Trailer=TRUE. However I
                # don't think there's any guarantee that
                # Transfer-Encoding comes before Trailer, so we're
                # effectively doing a second-pass here.
                if ($doing_chunked and $trailer_value_is_true) {
                    $trailer_mode = 1;
                }

                if ($doing_chunked) {
                    die "PANIC: The experimental Hijk support for chunked transfer encoding needs to be explicitly enabled with parse_chunked => 1"
                        unless $parse_chunked;

                    # if there is chunked encoding we have to ignore content length even if we have it
                    return (
                        $close_connection, $proto, $status_code, $header,
                        _read_chunked_body(
                            $body, $fd, $read_length, $read_timeout,
                            $head_as_array
                              ? $trailer_mode
                              : ($header->{Trailer} ? 1 : 0),
                        ),
                    );
                }

                if (defined $content_length) {
                    if ($content_length == 0) {
                        $read_length = 0;
                    } else {
                        $read_length = $content_length - length($body);
                    }
                } else {
                    $read_length = 10204;
                    $no_content_len = 1;
                }
            }
        }
    } while( !$decapitated || (!$method_has_no_content && ($read_length > 0 || $no_content_len)) );
    return ($close_connection, $proto, $status_code, $header, $body);
}

sub _read_chunked_body {
    my ($buf,$fd,$read_length,$read_timeout,$true_trailer_header) = @_;
    my $chunk_size   = 0;
    my $body         = "";
    my $trailer_mode = 0;
    my $wait_for_last_clrf = 0;
    vec(my $rin = '', $fd, 1) = 1;
    while(1) {
        # just read a 10k block and process it until it is consumed
        if (length($buf) < 3 || length($buf) < $chunk_size || $wait_for_last_clrf > 0) {
            return (undef, Hijk::Error::READ_TIMEOUT)
                if ((_select($rin, undef, undef, $read_timeout) != 1) || (defined($read_timeout) && $read_timeout <= 0));
            my $current_buf = "";
            my $nbytes = POSIX::read($fd, $current_buf, $read_length);
            if (!defined($nbytes)) {
                next if ($! == EWOULDBLOCK || $! == EAGAIN || $! == EINTR);
                return (
                    undef,
                    Hijk::Error::RESPONSE_READ_ERROR,
                    "Failed to chunked http body from socket",
                    $!+0,
                    "$!",
                );
            }

            if ($nbytes == 0) {
                return (
                    undef,
                    Hijk::Error::RESPONSE_BAD_READ_VALUE,
                    "Wasn't expecting a 0 byte response for chunked http body. This shouldn't happen, buf:<$buf>, current_buf:<$current_buf>",
                );
            }

            $buf .= $current_buf;
        }

        if ($wait_for_last_clrf > 0) {
            $wait_for_last_clrf -= length($buf);
            return $body if ($wait_for_last_clrf <= 0);
        }

        if ($trailer_mode) {
            # http://tools.ietf.org/html/rfc2616#section-14.40
            # http://tools.ietf.org/html/rfc2616#section-3.6.1
            #   A server using chunked transfer-coding in a response MUST NOT use the
            #   trailer for any header fields unless at least one of the following is
            #   true:

            #   a)the request included a TE header field that indicates "trailers" is
            #     acceptable in the transfer-coding of the  response, as described in
            #     section 14.39; or,

            #   b)the server is the origin server for the response, the trailer
            #     fields consist entirely of optional metadata, and the recipient
            #     could use the message (in a manner acceptable to the origin server)
            #     without receiving this metadata.  In other words, the origin server
            #     is willing to accept the possibility that the trailer fields might
            #     be silently discarded along the path to the client.

            # in case of trailer mode, we just read everything until the next CRLFCRLF
            my $neck_pos = index($buf, "${CRLF}${CRLF}");
            if ($neck_pos > 0) {
                return $body;
            }
        } else {
            if ($chunk_size > 0 && length($buf) >= $chunk_size) {
                $body .= substr($buf, 0, $chunk_size - 2); # our chunk size includes the following CRLF
                $buf = substr($buf, $chunk_size);
                $chunk_size = 0;
            } else {
                my $neck_pos = index($buf, ${CRLF});
                if ($neck_pos > 0) {
                    $chunk_size = hex(substr($buf, 0, $neck_pos));
                    if ($chunk_size == 0) {
                        if ($true_trailer_header) {
                            $trailer_mode = 1;
                        } else {
                            $buf = substr($buf, $neck_pos + 2);
                            # in case we are missing the ending CLRF, we have to wait for it
                            # otherwise it is left int he socket
                            if (length($buf) < 2) {
                                $wait_for_last_clrf = 2 - length($buf);
                            } else {
                                return $body;
                            }
                        }
                    } else {
                        $chunk_size += 2; # include the following CRLF
                        $buf = substr($buf, $neck_pos + 2);
                    }
                } elsif($neck_pos == 0) {
                    return (
                        undef,
                        Hijk::Error::RESPONSE_BAD_READ_VALUE,
                        "Wasn't expecting CLRF without chunk size. This shouldn't happen, buf:<$buf>",
                    );
                }
            }
        }
    }
}

sub _construct_socket {
    my ($host, $port, $connect_timeout) = @_;

    # If we can't find the IP address there'll be no point in even
    # setting up a socket.
    my $addr;
    {
        my $inet_aton = gethostbyname($host);
        return (undef, {error => Hijk::Error::CANNOT_RESOLVE}) unless defined $inet_aton;
        $addr = pack_sockaddr_in($port, $inet_aton);
    }

    my $tcp_proto = getprotobyname("tcp");
    my $soc;
    socket($soc, PF_INET, SOCK_STREAM, $tcp_proto) || die "Failed to construct TCP socket: $!";
    my $flags = fcntl($soc, F_GETFL, 0) or die "Failed to set fcntl F_GETFL flag: $!";
    fcntl($soc, F_SETFL, $flags | O_NONBLOCK) or die "Failed to set fcntl O_NONBLOCK flag: $!";

    if (!connect($soc, $addr) && $! != EINPROGRESS) {
        die "Failed to connect $!";
    }

    $connect_timeout = undef if defined($connect_timeout) && $connect_timeout <= 0;
    vec(my $rout = '', fileno($soc), 1) = 1;
    if (_select(undef, $rout, undef, $connect_timeout) != 1) {
        if (defined($connect_timeout)) {
            return (undef, {error => Hijk::Error::CONNECT_TIMEOUT});
        } else {
            return (
                undef,
                {
                    error         => Hijk::Error::REQUEST_SELECT_ERROR,
                    error_message => "select() error on constructing the socket",
                    errno_number  => $!+0,
                    errno_string  => "$!",
                },
            );
        }
    }

    if ($! = unpack("L", getsockopt($soc, SOL_SOCKET, SO_ERROR))) {
        die $!;
    }

    return $soc;
}

sub _build_http_message {
    my $args = $_[0];
    my $path_and_qs = ($args->{path} || "/") . ( defined($args->{query_string}) ? ("?".$args->{query_string}) : "" );
    return join(
        $CRLF,
        ($args->{method} || "GET")." $path_and_qs " . ($args->{protocol} || "HTTP/1.1"),
        ($args->{no_default_host_header}
         ? ()
         : ("Host: $args->{host}")),
        defined($args->{body}) ? ("Content-Length: " . length($args->{body})) : (),
        ($args->{head} and @{$args->{head}}) ? (
            map {
                $args->{head}[2*$_] . ": " . $args->{head}[2*$_+1]
            } 0..$#{$args->{head}}/2
        ) : (),
        ""
    ) . $CRLF . (defined($args->{body}) ? $args->{body} : "");
}

our $SOCKET_CACHE = {};

sub request {
    my $args = $_[0];

    # Backwards compatibility for code that provided the old timeout
    # argument.
    $args->{connect_timeout} = $args->{read_timeout} = $args->{timeout} if exists $args->{timeout};

    # Ditto for providing a default socket cache, allow for setting it
    # to "socket_cache => undef" to disable the cache.
    $args->{socket_cache} = $SOCKET_CACHE unless exists $args->{socket_cache};

    # Provide a default for the read_length option
    $args->{read_length} = 10 * 2 ** 10 unless exists $args->{read_length};

    # Use $; so we can use the $socket_cache->{$$, $host, $port}
    # idiom to access the cache.
    my $cache_key; $cache_key = join($;, $$, @$args{qw(host port)}) if defined $args->{socket_cache};

    my $soc;
    if (defined $cache_key and exists $args->{socket_cache}->{$cache_key}) {
        $soc = $args->{socket_cache}->{$cache_key};
    } else {
        ($soc, my $error) = _construct_socket(@$args{qw(host port connect_timeout)});
        return $error if $error;
        $args->{socket_cache}->{$cache_key} = $soc if defined $cache_key;
        $args->{on_connect}->() if exists $args->{on_connect};
    }

    my $r = _build_http_message($args);
    my $total = length($r);
    my $left = $total;

    vec(my $rout = '', fileno($soc), 1) = 1;
    while ($left > 0) {
        if (_select(undef, $rout, undef, undef) != 1) {
            delete $args->{socket_cache}->{$cache_key} if defined $cache_key;
            return {
                error         => Hijk::Error::REQUEST_SELECT_ERROR,
                error_message => "Got error on select() before the write() when while writing the HTTP request the socket",
                errno_number  => $!+0,
                errno_string  => "$!",
            };
        }

        my $rc = syswrite($soc,$r,$left, $total - $left);
        if (!defined($rc)) {
            next if ($! == EWOULDBLOCK || $! == EAGAIN || $! == EINTR);
            delete $args->{socket_cache}->{$cache_key} if defined $cache_key;
            shutdown($soc, 2);
            return {
                error         => Hijk::Error::REQUEST_WRITE_ERROR,
                error_message => "Got error trying to write the HTTP request with write() to the socket",
                errno_number  => $!+0,
                errno_string  => "$!",
            };
        }
        $left -= $rc;
    }

    my ($close_connection,$proto,$status,$head,$body,$error,$error_message,$errno_number,$errno_string);
    eval {
        ($close_connection,$proto,$status,$head,$body,$error,$error_message,$errno_number,$errno_string) =
        _read_http_message(fileno($soc), @$args{qw(read_length read_timeout parse_chunked head_as_array method)});
        1;
    } or do {
        my $err = $@ || "zombie error";
        delete $args->{socket_cache}->{$cache_key} if defined $cache_key;
        shutdown($soc, 2);
        die $err;
    };

    if ($status == 0
        # We always close connections for 1.0 because some servers LIE
        # and say that they're 1.0 but don't close the connection on
        # us! An example of this. Test::HTTP::Server (used by the
        # ShardedKV::Storage::Rest tests) is an example of such a
        # server. In either case we can't cache a connection for a 1.0
        # server anyway, so BEGONE!
        or $close_connection
        or (defined $proto and $proto eq 'HTTP/1.0')) {
        delete $args->{socket_cache}->{$cache_key} if defined $cache_key;
        shutdown($soc, 2);
    }
    return {
        proto => $proto,
        status => $status,
        head => $head,
        body => $body,
        defined($error) ? ( error => $error ) : (),
        defined($error_message) ? ( error_message => $error_message ) : (),
        defined($errno_number) ? ( errno_number => $errno_number ) : (),
        defined($errno_string) ? ( errno_string => $errno_string ) : (),
    };
}

sub _select {
    my ($rbits, $wbits, $ebits, $timeout) = @_;
    while (1) {
        my $start = Time::HiRes::time();
        my $nfound = select($rbits, $wbits, $ebits, $timeout);
        if ($nfound == -1 && $! == EINTR) {
            $timeout -= Time::HiRes::time() - $start if $timeout;
            next;
        }
        return $nfound;
    }
}

1;

__END__