File Coverage

blib/lib/Perinci/Access/Simple/Client.pm

Criterion	Covered	Total	%
statement	102	192	53.1
branch	40	86	46.5
condition	14	23	60.8
subroutine	17	23	73.9
pod	6	6	100.0
total	179	330	54.2

package Perinci::Access::Simple::Client;

our $DATE = '2016-10-07'; # DATE
our $VERSION = '0.22'; # VERSION

use 5.010001;
use strict;
use warnings;
use Log::Any '$log';

use Cwd qw(abs_path);
use Perinci::AccessUtil qw(strip_riap_stuffs_from_res);
use POSIX qw(:sys_wait_h);
use Tie::Cache;
use URI::Split qw(uri_split);
use URI::Escape;

use parent qw(Perinci::Access::Base);

my @logging_methods = Log::Any->logging_methods();

sub new {
    my $class = shift;

    my $self = $class->SUPER::new(@_);

    # attributes
    $self->{retries}         //= 2;
    $self->{retry_delay}     //= 3;
    $self->{conn_cache_size} //= 32;

    # connection cache, key="tcp:HOST:PORT" OR "unix:ABSPATH" or "pipe:ABSPATH
    # ARGS". value=hash, for tcp & unix {socket=>...} and for pipe {pid=>...,
    # chld_out=>..., chld_in=>...}
    tie my(%conns), 'Tie::Cache', $self->{conn_cache_size};
    $self->{_conns} = \%conns;

    $self;
}

# for older Perinci::Access::Base 0.28-, to remove later
sub _init {}

sub _delete_cache {
    my ($self, $wanted) = @_;
    my $conns = $self->{_conns};
    return unless $conns;

    for my $k ($wanted ? ($wanted) : (keys %$conns)) {
        if ($k =~ /^pipe:/) {
            waitpid($conns->{$k}{pid}, WNOHANG);
        }
        delete $self->{_conns}{$k};
    }
}

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

    #$self->_delete_cache;
}

sub request {
    my $self = shift;
    $self->_parse_or_request('request', @_);
}

sub _parse {
    my $self = shift;
    $self->_parse_or_request('parse2', @_);
}

# which: parse0 = quick parse (for parse_url(), parse2 = more thorough parse
# (for testing)
sub _parse_or_request {
    my ($self, $which, $action, $server_url, $extra) = @_;
    $log->tracef("=> %s\::request(action=%s, server_url=%s, extra=%s)",
                 __PACKAGE__, $action, $server_url, $extra);
    return [400, "Please specify server_url"] unless $server_url;

    my ($uri,
        $cache_key,
        $host, $port, # tcp
        $path,        # unix & pipe
        $args         # pipe
    );
    my ($srvsch, $srvauth, $srvpath, $srvquery, $srvfrag) =
        uri_split($server_url);
    $srvauth //= "";
    $srvpath //= "";
    return [400, "Please supply only riap+tcp/riap+unix/riap+pipe URL"]
        unless $srvsch =~ /\Ariap\+(tcp|unix|pipe)\z/;
    if ($srvsch eq 'riap+tcp') {
        if ($srvauth =~ m!^(.+):(\d+)$!) {
            ($host, $port) = ($1, $2, $3);
            $uri = $srvpath;
            $cache_key = "tcp:".lc($host).":$port";
        } else {
            return [400, "Invalid riap+tcp URL, please use this format: ".
                "riap+tcp://host:1234 or riap+tcp://host:1234/uri"];
        }
    } elsif ($srvsch eq 'riap+unix') {
        if ($srvpath =~ m!(.+)/(/.*)!) {
            ($path, $uri) = (uri_unescape($1), $2);
        } elsif ($srvpath =~ m!(.+)!) {
            $path = uri_unescape($1);
        }
        unless ($which eq 'parse0') {
            if (defined($path)) {
                my $apath = abs_path($path) or
                    return [500, "Can't find absolute path for $path"];
                $cache_key = "unix:$apath";
            } else {
                return [400, "Invalid riap+unix URL, please use this format: ".
                            ", e.g.: riap+unix:/path/to/unix/socket or ".
                                "riap+unix:/path/to/unix/socket//uri"];
            }
        }
    } elsif ($srvsch eq 'riap+pipe') {
        if ($srvpath =~ m!(.+?)//(.*?)/(/.*)!) {
            ($path, $args, $uri) = (uri_unescape($1), $2, $3);
        } elsif ($srvpath =~ m!(.+?)//(.*)!) {
            ($path, $args) = (uri_unescape($1), $2);
        } elsif ($srvpath =~ m!(.+)!) {
            $path = uri_unescape($1);
            $args = '';
        }
        $args = [map {uri_unescape($_)} split m!/!, $args // ''];
        unless ($which eq 'parse0') {
            if (defined($path)) {
                my $apath = abs_path($path) or
                    return [500, "Can't find absolute path for $path"];
                $cache_key = "pipe:$apath ".join(" ", @$args);
            } else {
                return [400, "Invalid riap+pipe URL, please use this format: ".
                            "riap+pipe:/path/to/prog or ".
                                "riap+pipe:/path/to/prog//arg1/arg2 or ".
                                    "riap+pipe:/path/to/prog//arg1/arg2//uri"];
            }
        }
    }

    my $req;
    my $res;

    unless ($which eq 'parse0') {
        $req = { v=>$self->{riap_version}, action=>$action, %{$extra // {}} };
        $uri ||= $req->{uri}; $req->{uri} //= $uri;
        $res = $self->check_request($req);
        return $res if $res;
    }

    if ($which =~ /parse/) {
        return [200, "OK", {
            args=>$args, host=>$host, path=>$path, port=>$port,
            scheme=>$srvsch, uri=>$uri,
        }];
    }

    $log->tracef("Parsed URI, scheme=%s, host=%s, port=%s, path=%s, args=%s, ".
                     "uri=%s", $srvsch, $host, $port, $path, $args, $uri);

    require JSON::MaybeXS;
    state $json = JSON::MaybeXS->new->allow_nonref;

    my $attempts = 0;
    my $do_retry;
    my $e;
    while (1) {
        $do_retry = 0;

        my ($in, $out);
        my $cache = $self->{_conns}{$cache_key};
        # check cache staleness
        if ($cache) {
            if ($srvsch =~ /tcp|unix/) {
                if ($cache->{socket}->connected) {
                    $in = $out = $cache->{socket};
                } else {
                    $log->infof("Stale socket cache (%s), discarded",
                                $cache_key);
                    $cache = undef;
                }
            } else {
                if (kill(0, $cache->{pid})) {
                    $in  = $cache->{chld_out};
                    $out = $cache->{chld_in};
                } else {
                    $log->infof(
                        "Process (%s) seems dead/unsignalable, discarded",
                        $cache_key);
                    $cache = undef;
                }
            }
        }
        # connect
        if (!$cache) {
            if ($srvsch =~ /tcp|unix/) {
                my $sock;
                if ($srvsch eq 'riap+tcp') {
                    require IO::Socket::INET;
                    $sock = IO::Socket::INET->new(
                        PeerHost => $host,
                        PeerPort => $port,
                        Proto    => 'tcp',
                    );
                } else {
                    use IO::Socket::UNIX;
                    $sock = IO::Socket::UNIX->new(
                        Type => SOCK_STREAM,
                        Peer => $path,
                    );
                }
                $e = $@;
                if ($sock) {
                    $self->{_conns}{$cache_key} = {socket=>$sock};
                    $in = $out = $sock;
                } else {
                    $e = $srvsch eq 'riap+tcp' ?
                        "Can't connect to TCP socket $host:$port: $e" :
                            "Can't connect to Unix socket $path: $e";
                    $do_retry++; goto RETRY;
                }
            } else {
                # taken from Modern::Perl. enable methods on filehandles;
                # unnecessary when 5.14 autoloads them
                require IO::File;
                require IO::Handle;

                require IPC::Open2;

                require String::ShellQuote;
                my $cmd = $path . (@$args ? " " . join(" ", map {
                    String::ShellQuote::shell_quote($_) } @$args) : "");
                $log->tracef("executing cmd: %s", $cmd);

                # using shell
                #my $pid = IPC::Open2::open2($in, $out, $cmd, @$args);

                # not using shell
                my $pid = IPC::Open2::open2($in, $out, $path, @$args);

                if ($pid) {
                    $self->{_conns}{$cache_key} = {
                        pid=>$pid, chld_out=>$in, chld_in=>$out};
                } else {
                    $e = "Can't open2 $cmd: $!";
                    $do_retry++; goto RETRY;
                }
            }
        }

        my $req_json;
        eval { $req_json = $json->encode($req) };
        $e = $@;
        return [400, "Can't encode request as JSON: $e"] if $e;

        $out->write("j$req_json\015\012");
        $log->tracef("Sent request to server: %s", $req_json);

        # XXX alarm/timeout
        my $line = $in->getline;
        $log->tracef("Got line from server: %s", $line);
        if (!$line) {
            $self->_delete_cache($cache_key);
            return [500, "Empty response from server"];
        } elsif ($line !~ /^j(.+)/) {
            $self->_delete_cache($cache_key);
            return [500, "Invalid response line from server: $line"];
        }
        eval { $res = $json->decode($1) };
        $e = $@;
        if ($e) {
            $self->_delete_cache($cache_key);
            return [500, "Invalid JSON response from server: $e"];
        }
        strip_riap_stuffs_from_res($res);
        return $res;

      RETRY:
        if ($do_retry && $attempts++ < $self->{retries}) {
            $log->tracef("Request failed ($e), waiting to retry #%s...",
                         $attempts);
            sleep $self->{retry_delay};
        } else {
            last;
        }
    }
    return [500, "$e (tried $attempts times)"];
}

sub request_tcp {
    my ($self, $action, $hostport, $extra) = @_;
    $self->request($action, "riap+tcp://$hostport->[0]:$hostport->[1]", $extra);
}

sub request_unix {
    my ($self, $action, $sockpath, $extra) = @_;
    $self->request($action => "riap+unix:" . uri_escape($sockpath), $extra);
}

sub request_pipe {
    my ($self, $action, $cmd, $extra) = @_;
    $self->request($action => "riap+pipe:" . uri_escape($cmd->[0]) . "//" .
                       join("/", map {uri_escape($_)} @$cmd[1..@$cmd-1]),
                   $extra);
}

sub parse_url {
    my ($self, $uri) = @_;

    my $res0 = $self->_parse_or_request('parse0', 'dummy', $uri);
    #use Data::Dump; dd $res0;
    die "Can't parse URL $uri: $res0->[0] - $res0->[1]" unless $res0->[0]==200;
    $res0 = $res0->[2];
    my $res = {proto=>$res0->{scheme}, path=>$res0->{uri}};
    if ($res->{proto} eq 'riap+unix') {
        $res->{unix_sock_path} = $res0->{path};
    } elsif ($res->{proto} eq 'riap+tcp') {
        $res->{host} = $res0->{host};
        $res->{port} = $res0->{port};
    } elsif ($res->{proto} eq 'riap+pipe') {
        $res->{prog_path} = $res0->{path};
        $res->{args} = $res0->{args};
    }

    $res;
}

1;
# ABSTRACT: Riap::Simple client

__END__