File Coverage

blib/lib/Test/HTTP/AnyEvent/Server.pm

Criterion	Covered	Total	%
statement	173	186	93.0
branch	29	34	85.2
condition	3	8	37.5
subroutine	29	31	93.5
pod	2	4	50.0
total	236	263	89.7

package Test::HTTP::AnyEvent::Server;
# ABSTRACT: the async counterpart to Test::HTTP::Server


use feature qw(state switch);
use strict;
use utf8;
use warnings qw(all);

use AnyEvent;
use AnyEvent::Handle;
use AnyEvent::Log;
use AnyEvent::Socket;
use AnyEvent::Util;
use HTTP::Headers;
use HTTP::Request;
use HTTP::Response;
use HTTP::Status;
use Moo;
use MooX::Types::MooseLike::Base qw(:all);
use POSIX;

no if ($] >= 5.017010), warnings => q(experimental);

#$AnyEvent::Log::FILTER->level('debug');

our $VERSION = '0.013'; # VERSION

my %pool;


has address     => (is => 'ro', isa => Str, default => sub { '127.0.0.1' }, writer => 'set_address');


has port        => (is => 'ro', isa => Int, writer => 'set_port');


has maxconn     => (is => 'ro', isa => Int, default => sub { 10 });


has timeout     => (is => 'ro', isa => Int, default => sub { 60 });


has disable_proxy => (is => 'ro', isa => Bool, default => sub { 1 });


has https       => (is => 'ro', isa => HashRef);


has custom_handler => (is => 'ro', isa => CodeRef);


has forked      => (is => 'ro', isa => Bool, default => sub { 0 });


has forked_pid  => (is => 'ro', isa => Int, writer => 'set_forked_pid');


has server      => (is => 'ro', isa => Ref, writer => 'set_server');


sub BUILD {
    my ($self) = @_;

    ## no critic (RequireLocalizedPunctuationVars)
    @ENV{qw(no_proxy http_proxy https_proxy ftp_proxy all_proxy)} = (q(localhost,127.0.0.1), (q()) x 4)
        if $self->disable_proxy;

    unless ($self->forked) {
        $self->set_server(
            $self->start_server(sub {
                my (undef, $address, $port) = @_;
                $self->set_address($address);
                $self->set_port($port);
                AE::log info =>
                    'bound to ' . $self->uri;
            })
        );
    } else {
        my ($rh, $wh) = portable_pipe;

        given (fork) {
            when (undef) {
                AE::log fatal =>
                    "couldn't fork(): $!";
            } when (0) {
                # child
                close $rh;

                my $h = AnyEvent::Handle->new(
                    fh          => $wh,
                    on_error    => sub {
                        AE::log fatal =>
                            "couldn't syswrite() to pipe: $!";
                    },
                );

                $self->set_server(
                    $self->start_server(sub {
                        my (undef, $address, $port) = @_;
                        # have to postpone so the address/port gets actually bound
                        AE::postpone { $h->push_write(join("\t", $address, $port)) };
                    })
                );

                AE::cv->wait;
                POSIX::_exit(0);
                exit 1;
            } default {
                # parent
                my $pid = $_;
                close $wh;

                my $buf;
                my $len = sysread $rh, $buf, 65536;
                AE::log fatal =>
                    "couldn't sysread() from pipe: $!"
                        if not defined $len or not $len;

                my ($address, $port) = split m{\t}x, $buf;
                $self->set_address($address);
                $self->set_port($port);
                $self->set_forked_pid($pid);
                AE::log info =>
                    "forked as $pid and bound to " . $self->uri;
            }
        }
    }

    return;
}

sub DEMOLISH {
    my ($self) = @_;

    if ($self->forked) {
        my $pid = $self->forked_pid;
        kill 9 => $pid;
        AE::log info =>
            "killed $pid";
    }

    return;
}


sub uri {
    my ($self) = @_;
    return sprintf(
        '%s://%s:%d/',
        ($self->https ? 'https' : 'http'),
        $self->address,
        $self->port,
    );
}


sub start_server {
    my ($self, $cb) = @_;

    return tcp_server(
        $self->address => $self->port,
        sub {
            my ($fh, $host, $port) = @_;
            if (scalar keys %pool > $self->maxconn) {
                AE::log error =>
                    "deny connection from $host:$port (too many connections)\n";
                return;
            } else {
                AE::log warn =>
                    "new connection from $host:$port\n";
            }

            my $h = AnyEvent::Handle->new(
                fh          => $fh,
                on_eof      => \&_cleanup,
                on_error    => \&_cleanup,
                timeout     => $self->timeout,
                ($self->https ? (tls_ctx => $self->https) : ()),
            );

            $h->push_read(tls_autostart => 'accept') if $self->https;

            $pool{fileno($fh)} = $h;
            AE::log debug =>
                sprintf "%d connection(s) in pool\n", scalar keys %pool;

            $self->_start($h);
        } => $cb
    );
}


sub _start {
    my ($self, $my_handle) = @_;
    return $my_handle->push_read(regex => qr{(\015?\012){2}}x, sub {
        my ($h, $data) = @_;
        my ($req, $hdr) = split m{\015?\012}x, $data, 2;
        $req =~ s/\s+$//sx;
        AE::log debug => "request: [$req]\n";
        if ($hdr =~ m{\bContent-length:\s*(\d+)\b}isx) {
            AE::log debug => "expecting content\n";
            $h->push_read(chunk => int($1), sub {
                my ($_h, $_data) = @_;
                $self->_reply($_h, $req, $hdr, $_data);
            });
        } else {
            $self->_reply($h, $req, $hdr);
        }
    });
}


sub _cleanup {
    my ($h) = @_;
    AE::log debug => "closing connection\n";
    my $r = eval {
        ## no critic (ProhibitNoWarnings)
        no warnings;

        my $id = fileno($h->{fh});
        delete $pool{$id};
        shutdown $h->{fh}, 2;

        return 1;
    };
    AE::log warn => "shutdown() aborted\n"
        if not defined $r or $@;
    $h->destroy;
    return;
}


sub _reply {
    my ($self, $h, $req, $hdr, $content) = @_;
    state $timer = {};

    my $res = HTTP::Response->new(
        &HTTP::Status::RC_OK ,=> undef,
        HTTP::Headers->new(
            Connection      => 'close',
            Content_Type    => 'text/plain',
            Server          => __PACKAGE__ . "/@{[ $Test::HTTP::AnyEvent::Server::VERSION // 0 ]} AnyEvent/$AE::VERSION Perl/$] ($^O)",
        )
    );
    $res->date(time);
    $res->protocol('HTTP/1.0');

    if ($req =~ m{^(GET|POST)\s+(.+)\s+(HTTP/1\.[01])$}ix) {
        my ($method, $uri, $protocol) = ($1, $2, $3);
        AE::log debug => "sending response to $method ($protocol)\n";
        AE::log debug => "simulating connection to $1\n"
            if $uri =~ s{^(https?://[^/]+)}{}ix;
        for ($uri) {
            when (m{^/repeat/(\d+)/(.+)}x) {
                $res->content($2 x $1);
            } when (m{^/echo/head$}x) {
                $res->content(
                    join(
                        "\015\012",
                        qq($method $uri $protocol),
                        $hdr,
                    )
                );
            } when (m{^/echo/body$}x) {
                $res->content($content);
            } when (m{^/delay/(\d+)$}x) {
                $res->content(sprintf(qq(issued %s\n), scalar gmtime));
                $timer->{$h} = AE::timer $1, 0, sub {
                    delete $timer->{$h};
                    AE::log debug => "delayed response\n";
                    $h->push_write($res->as_string("\015\012"));
                    _cleanup($h);
                };
                return;
            } default {
                my $found;
                if ($self->custom_handler) {
                    $res->request(HTTP::Request->new(
                        $method,
                        $uri,
                        [
                            map {
                                m{^\s*([^:\s]+)\s*:\s*(.*)$}sx
                            } split m{\015?\012}x, $hdr
                        ],
                        $content,
                    ));
                    $found = eval { $self->custom_handler->($res) };
                    if ($@) {
                        AE::log error => "custom_handler died: $@";
                        $res->code(&HTTP::Status::RC_INTERNAL_SERVER_ERROR);
                        $res->content($@);
                        $found = 1;
                    }
                }
                unless ($found) {
                    $res->code(&HTTP::Status::RC_NOT_FOUND);
                    $res->content('Not Found');
                }
            }
        }
    } elsif ($req =~ m{^CONNECT\s+([\w\.\-]+):(\d+)\s+(HTTP/1\.[01])$}ix) {
        my ($peer_host, $peer_port, $protocol) = ($1, $2, $3);
        AE::log debug => "simulating connection to $peer_host:$peer_port ($protocol)\n";
        $res->message('Connection established');
        $h->push_write($res->as_string("\015\012"));
        if ($self->https) {
            AE::log debug => 'attempting to use TLS';
            $h->push_read(tls_autostart => 'accept');
        }
        $self->_start($h);
        return;
    } else {
        AE::log error => "bad request\n";
        $res->code(&HTTP::Status::RC_BAD_REQUEST);
        $res->content('Bad Request');
    }

    $h->push_write($res->as_string("\015\012"));
    _cleanup($h);
    return;
}


1;

__END__