File Coverage

blib/lib/Net/MPRPC/Client.pm

Criterion	Covered	Total	%
statement	33	98	33.6
branch	0	36	0.0
condition	0	14	0.0
subroutine	11	20	55.0
pod	5	5	100.0
total	49	173	28.3

package Net::MPRPC::Client;
use strict;
use warnings;

our $VERSION = '0.02';

use IO::Select;
use IO::Socket::INET;

our $_HAVE_UNIX_SOCKET = 1;
eval q[use IO::Socket::INET; 1];
if ($@) { $_HAVE_UNIX_SOCKET = 0 }

use Try::Tiny;
use Carp;
use Data::MessagePack;

use constant MP_REQ_TYPE   => 0;
use constant MP_RES_ERROR  => 2;
use constant MP_RES_RESULT => 3;

sub new {
    my $class = shift;
    my $args  = @_ > 1 ? {@_} : $_[0];

    $args->{_id} = 0;
    $args->{timeout} ||= 30;
    $args->{_error} = q[];

    bless $args, $class;
}

sub connect {
    my $self = shift;
    my $args = @_ > 1 ? {@_} : $_[0];

    $self->disconnect if $self->{_sock};

    my $host = $args->{host} || $self->{host};
    my $port = $args->{post} || $self->{port};

    croak q[Required "host" parameter to connect] unless $host;
    croak q[Required "port" parameter to connect] unless $port;

    my $sock;
    try {
        if ($host eq 'unix/') {
            if (!$_HAVE_UNIX_SOCKET) {
                croak "This environment doesn't support UNIX socket";
            }

            $sock = IO::Socket::UNIX->new(
                Peer    => $port,
                Timeout => $self->{timeout},
            ) or die qq/Unable to connect unix socket "$port": $!/;
        }
        else {
            $sock = IO::Socket::INET->new(
                PeerAddr => $host,
                PeerPort => $port,
                Proto    => 'tcp',
                Timeout  => $self->{timeout},
            ) or die qq/Unable to connect "${host}:${port}": $!/;
        }

        $sock->autoflush(1);
        $self->{_sock} = $sock;
    } catch {
        $self->{_error} = $_;
    };

    return !!$self->{_sock};
}

sub disconnect {
    delete $_[0]->{_sock};
}

sub call {
    my ($self, $method, $param) = @_;

    $self->{_error} = q[];
    return unless $self->{_sock} or $self->connect;

    my $sock = $self->{_sock};
    my $req  = [
        MP_REQ_TYPE, ++$self->{_id},
        $method, $param,
    ];
    $sock->print(Data::MessagePack->pack($req));

    my $timeout = $sock->timeout;
    my $limit   = time + $timeout;
    my $buf     = q[];

    my $select = IO::Select->new or croak $!;
    $select->add($sock);

    my $unpacker = Data::MessagePack::Unpacker->new;
    my $nread    = 0;

    while ($limit >= time) {
        my @ready = $select->can_read( $limit - time )
            or last;

        croak q/Fatal error on select, $ready[0] isn't $sock/
            if $sock ne $ready[0];

        unless (my $l = $sock->sysread($buf, 512, length $buf)) {
            my $e = $!;
            $self->disconnect;
            croak qq/Error reading socket: $e/;
        }

        try {
            $nread = $unpacker->execute($buf, $nread);
        } catch {
            $self->{_error} = $_;
            $self->disconnect;
            $unpacker->reset;
        };
        return if $self->{_error};

        if ($unpacker->is_finished) {
            my $res = $unpacker->data;
            $unpacker->reset;

            unless ($res and ref $res eq 'ARRAY') {
                $self->{_error} = 'Invalid response from server';
                $self->disconnect;
                return;
            }

            if (my $error = $res->[MP_RES_ERROR]) {
                $self->{_error} = $error;
                return;
            }

            return $res->[MP_RES_RESULT];
        }
    }

    $self->disconnect;
    croak 'request timeout';
}

sub error {
    $_[0]->{_error};
}

1;

__END__